QueueRepository::findOldestEntryForProcess() - Code Metrics - Inspection of "Deprecation/count and count all" - AOEpeople/crawler - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#621)

by Tomas Norre

created 2020-10-12 18:02 UTC

QueueRepository::findOldestEntryForProcess() A

↳ Parent: QueueRepository

Complexity

Conditions	1
Paths	1

Size

Total Lines	3
Code Lines	1

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	1
nc	1
nop	1
dl	0
loc	3
ccs	2
cts	2
cp	1
crap	1
rs	10
c	0
b	0
f	0

<?php

declare(strict_types=1);

namespace AOE\Crawler\Domain\Repository;

/***************************************************************
 *  Copyright notice
 *
 *  (c) 2020 AOE GmbH <[email protected]>
 *
 *  All rights reserved
 *
 *  This script is part of the TYPO3 project. The TYPO3 project is
 *  free software; you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation; either version 3 of the License, or
 *  (at your option) any later version.
 *
 *  The GNU General Public License can be found at
 *  http://www.gnu.org/copyleft/gpl.html.
 *
 *  This script is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  This copyright notice MUST APPEAR in all copies of the script!
 ***************************************************************/

use AOE\Crawler\Configuration\ExtensionConfigurationProvider;
use AOE\Crawler\Domain\Model\Process;
use Psr\Log\LoggerAwareInterface;
use TYPO3\CMS\Core\Database\Connection;
use TYPO3\CMS\Core\Database\ConnectionPool;
use TYPO3\CMS\Core\Utility\GeneralUtility;
use TYPO3\CMS\Extbase\Object\ObjectManager;
use TYPO3\CMS\Extbase\Persistence\Repository;

/**
 * Class QueueRepository
 *
 * @package AOE\Crawler\Domain\Repository
 */
class QueueRepository extends Repository implements LoggerAwareInterface
{
    use \Psr\Log\LoggerAwareTrait;

    /**
     * @var string
     */
    protected $tableName = 'tx_crawler_queue';

    public function __construct()
    {
        $objectManager = GeneralUtility::makeInstance(ObjectManager::class);
        parent::__construct($objectManager);
    }

    public function unsetQueueProcessId(string $processId): void
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $queryBuilder
            ->update($this->tableName)
            ->where(
                $queryBuilder->expr()->eq('process_id', $queryBuilder->createNamedParameter($processId))
            )
            ->set('process_id', '')
            ->execute();
    }

    /**
     * This method is used to find the youngest entry for a given process.
     */
    public function findYoungestEntryForProcess(Process $process): array
    {
        return $this->getFirstOrLastObjectByProcess($process, 'exec_time');
    }

    /**
     * This method is used to find the oldest entry for a given process.
     */
    public function findOldestEntryForProcess(Process $process): array
    {
        return $this->getFirstOrLastObjectByProcess($process, 'exec_time', 'DESC');
    }

    /**
     * Counts all executed items of a process.
     *
     * @param Process $process
     */
    public function countExecutedItemsByProcess($process): int
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);

        return $queryBuilder
            ->count('*')
            ->from($this->tableName)
            ->where(
                $queryBuilder->expr()->eq('process_id_completed', $queryBuilder->createNamedParameter($process->getProcessId())),
                $queryBuilder->expr()->gt('exec_time', 0)
            )
            ->execute()
            ->fetchColumn(0);
    }

    /**
     * Counts items of a process which yet have not been processed/executed
     *
     * @param Process $process
     */
    public function countNonExecutedItemsByProcess($process): int
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);

        return $queryBuilder
            ->count('*')
            ->from($this->tableName)
            ->where(
                $queryBuilder->expr()->eq('process_id', $queryBuilder->createNamedParameter($process->getProcessId())),
                $queryBuilder->expr()->eq('exec_time', 0)
            )
            ->execute()
            ->fetchColumn(0);
    }

    /**
     * get items which have not been processed yet
     */
    public function getUnprocessedItems(): array
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);

        return $queryBuilder
            ->select('*')
            ->from($this->tableName)
            ->where(
                $queryBuilder->expr()->eq('exec_time', 0)
            )
            ->execute()->fetchAll();
    }

    /**
     * Count items which have not been processed yet
     */
    public function countUnprocessedItems(): int
    {
        return count($this->getUnprocessedItems());
    }

    /**
     * This method can be used to count all queue entrys which are
     * scheduled for now or a earlier date.
     */
    public function countAllPendingItems(): int
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);

        return $queryBuilder
            ->count('*')
            ->from($this->tableName)
            ->where(
                $queryBuilder->expr()->eq('process_scheduled', 0),
                $queryBuilder->expr()->eq('exec_time', 0),
                $queryBuilder->expr()->lte('scheduled', time())
            )
            ->execute()
            ->fetchColumn(0);
    }

    /**
     * This method can be used to count all queue entries which are
     * scheduled for now or a earlier date and are assigned to a process.
     */
    public function countAllAssignedPendingItems(): int
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);

        return $queryBuilder
            ->count('*')
            ->from($this->tableName)
            ->where(
                $queryBuilder->expr()->neq('process_id', '""'),
                $queryBuilder->expr()->eq('exec_time', 0),
                $queryBuilder->expr()->lte('scheduled', time())
            )
            ->execute()
            ->fetchColumn(0);
    }

    /**
     * This method can be used to count all queue entrys which are
     * scheduled for now or a earlier date and are not assigned to a process.
     */
    public function countAllUnassignedPendingItems(): int
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);

        return $queryBuilder
            ->count('*')
            ->from($this->tableName)
            ->where(
                $queryBuilder->expr()->eq('process_id', '""'),
                $queryBuilder->expr()->eq('exec_time', 0),
                $queryBuilder->expr()->lte('scheduled', time())
            )
            ->execute()
            ->fetchColumn(0);
    }

    /**
     * Count pending queue entries grouped by configuration key
     */
    public function countPendingItemsGroupedByConfigurationKey(): array
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $statement = $queryBuilder
            ->from($this->tableName)
            ->selectLiteral('count(*) as unprocessed', 'sum(process_id != \'\') as assignedButUnprocessed')
            ->addSelect('configuration')
            ->where(
                $queryBuilder->expr()->eq('exec_time', 0),
                $queryBuilder->expr()->lt('scheduled', time())
            )
            ->groupBy('configuration')
            ->execute();

        return $statement->fetchAll();
    }

    /**
     * Get set id with unprocessed entries
     *
     * @return array array of set ids
     */
    public function getSetIdWithUnprocessedEntries(): array
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $statement = $queryBuilder
            ->select('set_id')
            ->from($this->tableName)
            ->where(
                $queryBuilder->expr()->lt('scheduled', time()),
                $queryBuilder->expr()->eq('exec_time', 0)
            )
            ->addGroupBy('set_id')
            ->execute();

        $setIds = [];
        while ($row = $statement->fetch()) {
            $setIds[] = intval($row['set_id']);
        }

        return $setIds;
    }

    /**
     * Get total queue entries by configuration
     *
     * @return array totals by configuration (keys)
     */
    public function getTotalQueueEntriesByConfiguration(array $setIds): array
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $totals = [];
        if (count($setIds) > 0) {
            $statement = $queryBuilder
                ->from($this->tableName)
                ->selectLiteral('count(*) as c')
                ->addSelect('configuration')
                ->where(
                    $queryBuilder->expr()->in('set_id', implode(',', $setIds)),
                    $queryBuilder->expr()->lt('scheduled', time())
                )
                ->groupBy('configuration')
                ->execute();

            while ($row = $statement->fetch()) {
                $totals[$row['configuration']] = $row['c'];
            }
        }

        return $totals;
    }

    /**
     * Get the timestamps of the last processed entries
     *
     * @param int $limit
     */
    public function getLastProcessedEntriesTimestamps($limit = 100): array
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $statement = $queryBuilder
            ->select('exec_time')
            ->from($this->tableName)
            ->addOrderBy('exec_time', 'desc')
            ->setMaxResults($limit)
            ->execute();

        $rows = [];
        while ($row = $statement->fetch()) {
            $rows[] = $row['exec_time'];
        }

        return $rows;
    }

    /**
     * Get the last processed entries
     *
     * @param int $limit
     */
    public function getLastProcessedEntries($limit = 100): array
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $statement = $queryBuilder
            ->from($this->tableName)
            ->select('*')
            ->orderBy('exec_time', 'desc')
            ->setMaxResults($limit)
            ->execute();

        $rows = [];
        while (($row = $statement->fetch()) !== false) {
            $rows[] = $row;
        }

        return $rows;
    }

    /**
     * Get performance statistics data
     *
     * @param int $start timestamp
     * @param int $end timestamp
     *
     * @return array performance data
     */
    public function getPerformanceData($start, $end): array
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $statement = $queryBuilder
            ->from($this->tableName)
            ->selectLiteral('min(exec_time) as start', 'max(exec_time) as end', 'count(*) as urlcount')
            ->addSelect('process_id_completed')
            ->where(
                $queryBuilder->expr()->neq('exec_time', 0),
                $queryBuilder->expr()->gte('exec_time', $queryBuilder->createNamedParameter($start, \PDO::PARAM_INT)),
                $queryBuilder->expr()->lte('exec_time', $queryBuilder->createNamedParameter($end, \PDO::PARAM_INT))
            )
            ->groupBy('process_id_completed')
            ->execute();

        $rows = [];
        while ($row = $statement->fetch()) {
            $rows[$row['process_id_completed']] = $row;
        }

        return $rows;
    }

    /**
     * Determines if a page is queued
     */
    public function isPageInQueue(int $uid, bool $unprocessed_only = true, bool $timed_only = false, int $timestamp = 0): bool
    {
        $isPageInQueue = false;

        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $statement = $queryBuilder
            ->from($this->tableName)
            ->count('*')
            ->where(
                $queryBuilder->expr()->eq('page_id', $queryBuilder->createNamedParameter($uid, \PDO::PARAM_INT))
            );

        if ($unprocessed_only !== false) {
            $statement->andWhere(
                $queryBuilder->expr()->eq('exec_time', 0)
            );
        }

        if ($timed_only !== false) {
            $statement->andWhere(
                $queryBuilder->expr()->neq('scheduled', 0)
            );
        }

        if ($timestamp) {
            $statement->andWhere(
                $queryBuilder->expr()->eq('scheduled', $queryBuilder->createNamedParameter($timestamp, \PDO::PARAM_INT))
            );
        }

        // TODO: Currently it's not working if page doesn't exists. See tests
        $count = $statement
            ->execute()
            ->fetchColumn(0);

        if ($count !== false && $count > 0) {
            $isPageInQueue = true;
        }

        return $isPageInQueue;
    }

    /**
     * Method to check if a page is in the queue which is timed for a
     * date when it should be crawled
     */
    public function isPageInQueueTimed(int $uid, bool $show_unprocessed = true): bool
    {
        return $this->isPageInQueue($uid, $show_unprocessed);
    }

    public function getAvailableSets(): array
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $statement = $queryBuilder
            ->selectLiteral('count(*) as count_value')
            ->addSelect('set_id', 'scheduled')
            ->from($this->tableName)
            ->orderBy('scheduled', 'desc')
            ->groupBy('set_id', 'scheduled')
            ->execute();

        $rows = [];
        while ($row = $statement->fetch()) {
            $rows[] = $row;
        }

        return $rows;
    }

    public function findByQueueId(string $queueId): ?array
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $queueRec = $queryBuilder
            ->select('*')
            ->from($this->tableName)
            ->where(
                $queryBuilder->expr()->eq('qid', $queryBuilder->createNamedParameter($queueId))
            )
            ->execute()
            ->fetch();
        return is_array($queueRec) ? $queueRec : null;
    }

    public function cleanupQueue(): void
    {
        $extensionSettings = GeneralUtility::makeInstance(ExtensionConfigurationProvider::class)->getExtensionConfiguration();
        $purgeDays = (int) $extensionSettings['purgeQueueDays'];

        if ($purgeDays > 0) {
            $purgeDate = time() - 24 * 60 * 60 * $purgeDays;

            $queryBuilderDelete = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
            $del = $queryBuilderDelete
                ->delete($this->tableName)
                ->where(
                    'exec_time != 0 AND exec_time < ' . $purgeDate
                )->execute();

            if ($del === false) {
                $this->logger->info(
                    'Records could not be deleted.'
                );
            }
        }
    }

    /**
     * Cleans up entries that stayed for too long in the queue. These are default:
     * - processed entries that are over 1.5 days in age
     * - scheduled entries that are over 7 days old
     */
    public function cleanUpOldQueueEntries(): void
    {
        $extensionSettings = GeneralUtility::makeInstance(ExtensionConfigurationProvider::class)->getExtensionConfiguration();
        $processedAgeInSeconds = $extensionSettings['cleanUpProcessedAge'] * 86400; // 24*60*60 Seconds in 24 hours
        $scheduledAgeInSeconds = $extensionSettings['cleanUpScheduledAge'] * 86400;

        $now = time();
        $condition = '(exec_time<>0 AND exec_time<' . ($now - $processedAgeInSeconds) . ') OR scheduled<=' . ($now - $scheduledAgeInSeconds);

        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $del = $queryBuilder
            ->delete($this->tableName)
            ->where(
                $condition
            )->execute();

        if ($del === false) {
            $this->logger->info(
                'Records could not be deleted.'
            );
        }
    }

    public function fetchRecordsToBeCrawled(int $countInARun)
    {
        $queryBuilderSelect = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        return $queryBuilderSelect
            ->select('qid', 'scheduled', 'page_id', 'sitemap_priority')
            ->from($this->tableName)
            ->leftJoin(
                $this->tableName,
                'pages',
                'p',
                $queryBuilderSelect->expr()->eq('p.uid', $queryBuilderSelect->quoteIdentifier($this->tableName . '.page_id'))
            )
            ->where(
                $queryBuilderSelect->expr()->eq('exec_time', 0),
                $queryBuilderSelect->expr()->eq('process_scheduled', 0),
                $queryBuilderSelect->expr()->lte('scheduled', time())
            )
            ->orderBy('sitemap_priority', 'DESC')
            ->addOrderBy('scheduled')
            ->addOrderBy('qid')
            ->setMaxResults($countInARun)
            ->execute()
            ->fetchAll();
    }

    public function updateProcessIdAndSchedulerForQueueIds(array $quidList, string $processId)
    {
        $queryBuilderUpdate = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        return $queryBuilderUpdate
            ->update($this->tableName)
            ->where(
                $queryBuilderUpdate->expr()->in('qid', $quidList)
            )
            ->set('process_scheduled', time())
            ->set('process_id', $processId)
            ->execute();
    }

    public function unsetProcessScheduledAndProcessIdForQueueEntries(array $processIds): void
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $queryBuilder
            ->update($this->tableName)
            ->where(
                $queryBuilder->expr()->eq('exec_time', 0),
                $queryBuilder->expr()->in('process_id', $queryBuilder->createNamedParameter($processIds, Connection::PARAM_STR_ARRAY))
            )
            ->set('process_scheduled', 0)
            ->set('process_id', '')
            ->execute();
    }

    /**
     * This method is used to count if there are ANY unprocessed queue entries
     * of a given page_id and the configuration which matches a given hash.
     * If there if none, we can skip an inner detail check
     *
     * @param int $uid
     * @param string $configurationHash
     * @return boolean
     */
    public function noUnprocessedQueueEntriesForPageWithConfigurationHashExist($uid, $configurationHash): bool
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $noUnprocessedQueueEntriesFound = true;

        $result = $queryBuilder
            ->count('*')
            ->from($this->tableName)
            ->where(
                $queryBuilder->expr()->eq('page_id', (int) $uid),
                $queryBuilder->expr()->eq('configuration_hash', $queryBuilder->createNamedParameter($configurationHash)),
                $queryBuilder->expr()->eq('exec_time', 0)
            )
            ->execute()
            ->fetchColumn();

        if ($result) {
            $noUnprocessedQueueEntriesFound = false;
        }

        return $noUnprocessedQueueEntriesFound;
    }

    /**
     * Removes queue entries
     *
     * @param string $filter all, pending, finished
     */
    public function flushQueue(string $filter): void
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);

        switch (strtolower($filter)) {
            case 'all':
                // No where claus needed delete everything
                break;
            case 'pending':
                $queryBuilder->andWhere($queryBuilder->expr()->eq('exec_time', 0));
                break;
            case 'finished':
            default:
                $queryBuilder->andWhere($queryBuilder->expr()->gt('exec_time', 0));
                break;
        }

        $queryBuilder
            ->delete($this->tableName)
            ->execute();
    }

    /**
     * @param string $processId
     *
     * @return bool|string
     */
    public function countAllByProcessId($processId)
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);

        return $queryBuilder
            ->count('*')
            ->from($this->tableName)
            ->where(
                $queryBuilder->expr()->eq('process_id', $queryBuilder->createNamedParameter($processId, \PDO::PARAM_STR))
            )
            ->execute()
            ->fetchColumn(0);
    }

    /**
     * This internal helper method is used to create an instance of an entry object
     *
     * @param Process $process
     * @param string $orderByField first matching item will be returned as object
     * @param string $orderBySorting sorting direction
     */
    protected function getFirstOrLastObjectByProcess($process, $orderByField, $orderBySorting = 'ASC'): array
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $first = $queryBuilder
            ->select('*')
            ->from($this->tableName)
            ->where(
                $queryBuilder->expr()->eq('process_id_completed', $queryBuilder->createNamedParameter($process->getProcessId())),
                $queryBuilder->expr()->gt('exec_time', 0)
            )
            ->setMaxResults(1)
            ->addOrderBy($orderByField, $orderBySorting)
            ->execute()->fetch(0);

        return $first ?: [];
    }
}


1		<?php
2
3		declare(strict_types=1);
4
5		namespace AOE\Crawler\Domain\Repository;
6
7		/***************************************************************
8		* Copyright notice
9		*
10		* (c) 2020 AOE GmbH <[email protected]>
11		*
12		* All rights reserved
13		*
14		* This script is part of the TYPO3 project. The TYPO3 project is
15		* free software; you can redistribute it and/or modify
16		* it under the terms of the GNU General Public License as published by
17		* the Free Software Foundation; either version 3 of the License, or
18		* (at your option) any later version.
19		*
20		* The GNU General Public License can be found at
21		* http://www.gnu.org/copyleft/gpl.html.
22		*
23		* This script is distributed in the hope that it will be useful,
24		* but WITHOUT ANY WARRANTY; without even the implied warranty of
25		* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
26		* GNU General Public License for more details.
27		*
28		* This copyright notice MUST APPEAR in all copies of the script!
29		***************************************************************/
30
31		use AOE\Crawler\Configuration\ExtensionConfigurationProvider;
32		use AOE\Crawler\Domain\Model\Process;
33		use Psr\Log\LoggerAwareInterface;
34		use TYPO3\CMS\Core\Database\Connection;
35		use TYPO3\CMS\Core\Database\ConnectionPool;
36		use TYPO3\CMS\Core\Utility\GeneralUtility;
37		use TYPO3\CMS\Extbase\Object\ObjectManager;
38		use TYPO3\CMS\Extbase\Persistence\Repository;
39
40		/**
41		* Class QueueRepository
42		*
43		* @package AOE\Crawler\Domain\Repository
44		*/
45		class QueueRepository extends Repository implements LoggerAwareInterface
46		{
47		use \Psr\Log\LoggerAwareTrait;
48
49		/**
50		* @var string
51		*/
52		protected $tableName = 'tx_crawler_queue';
53
54	102	public function __construct()
55		{
56	102	$objectManager = GeneralUtility::makeInstance(ObjectManager::class);
57	102	parent::__construct($objectManager);
58	102	}
59
60	3	public function unsetQueueProcessId(string $processId): void
61		{
62	3	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
63		$queryBuilder
64	3	->update($this->tableName)
65	3	->where(
66	3	$queryBuilder->expr()->eq('process_id', $queryBuilder->createNamedParameter($processId))
67		)
68	3	->set('process_id', '')
69	3	->execute();
70	3	}
71
72		/**
73		* This method is used to find the youngest entry for a given process.
74		*/
75	1	public function findYoungestEntryForProcess(Process $process): array
76		{
77	1	return $this->getFirstOrLastObjectByProcess($process, 'exec_time');
78		}
79
80		/**
81		* This method is used to find the oldest entry for a given process.
82		*/
83	1	public function findOldestEntryForProcess(Process $process): array
84		{
85	1	return $this->getFirstOrLastObjectByProcess($process, 'exec_time', 'DESC');
86		}
87
88		/**
89		* Counts all executed items of a process.
90		*
91		* @param Process $process
92		*/
93	1	public function countExecutedItemsByProcess($process): int
94		{
95	1	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
96
97		return $queryBuilder
98	1	->count('*')
99	1	->from($this->tableName)
100	1	->where(
101	1	$queryBuilder->expr()->eq('process_id_completed', $queryBuilder->createNamedParameter($process->getProcessId())),
102	1	$queryBuilder->expr()->gt('exec_time', 0)
103		)
104	1	->execute()
105	1	->fetchColumn(0);
106		}
107
108		/**
109		* Counts items of a process which yet have not been processed/executed
110		*
111		* @param Process $process
112		*/
113	1	public function countNonExecutedItemsByProcess($process): int
114		{
115	1	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
116
117		return $queryBuilder
118	1	->count('*')
119	1	->from($this->tableName)
120	1	->where(
121	1	$queryBuilder->expr()->eq('process_id', $queryBuilder->createNamedParameter($process->getProcessId())),
122	1	$queryBuilder->expr()->eq('exec_time', 0)
123		)
124	1	->execute()
125	1	->fetchColumn(0);
126		}
127
128		/**
129		* get items which have not been processed yet
130		*/
131	5	public function getUnprocessedItems(): array
132		{
133	5	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
134
135		return $queryBuilder
136	5	->select('*')
137	5	->from($this->tableName)
138	5	->where(
139	5	$queryBuilder->expr()->eq('exec_time', 0)
140		)
141	5	->execute()->fetchAll();
142		}
143
144		/**
145		* Count items which have not been processed yet
146		*/
147	5	public function countUnprocessedItems(): int
148		{
149	5	return count($this->getUnprocessedItems());
150		}
151
152		/**
153		* This method can be used to count all queue entrys which are
154		* scheduled for now or a earlier date.
155		*/
156	2	public function countAllPendingItems(): int
157		{
158	2	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
159
160		return $queryBuilder
161	2	->count('*')
162	2	->from($this->tableName)
163	2	->where(
164	2	$queryBuilder->expr()->eq('process_scheduled', 0),
165	2	$queryBuilder->expr()->eq('exec_time', 0),
166	2	$queryBuilder->expr()->lte('scheduled', time())
167		)
168	2	->execute()
169	2	->fetchColumn(0);
170		}
171
172		/**
173		* This method can be used to count all queue entries which are
174		* scheduled for now or a earlier date and are assigned to a process.
175		*/
176	2	public function countAllAssignedPendingItems(): int
177		{
178	2	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
179
180		return $queryBuilder
181	2	->count('*')
182	2	->from($this->tableName)
183	2	->where(
184	2	$queryBuilder->expr()->neq('process_id', '""'),
185	2	$queryBuilder->expr()->eq('exec_time', 0),
186	2	$queryBuilder->expr()->lte('scheduled', time())
187		)
188	2	->execute()
189	2	->fetchColumn(0);
190		}
191
192		/**
193		* This method can be used to count all queue entrys which are
194		* scheduled for now or a earlier date and are not assigned to a process.
195		*/
196	2	public function countAllUnassignedPendingItems(): int
197		{
198	2	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
199
200		return $queryBuilder
201	2	->count('*')
202	2	->from($this->tableName)
203	2	->where(
204	2	$queryBuilder->expr()->eq('process_id', '""'),
205	2	$queryBuilder->expr()->eq('exec_time', 0),
206	2	$queryBuilder->expr()->lte('scheduled', time())
207		)
208	2	->execute()
209	2	->fetchColumn(0);
210		}
211
212		/**
213		* Count pending queue entries grouped by configuration key
214		*/
215	1	public function countPendingItemsGroupedByConfigurationKey(): array
216		{
217	1	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
218		$statement = $queryBuilder
219	1	->from($this->tableName)
220	1	->selectLiteral('count(*) as unprocessed', 'sum(process_id != \'\') as assignedButUnprocessed')
221	1	->addSelect('configuration')
222	1	->where(
223	1	$queryBuilder->expr()->eq('exec_time', 0),
224	1	$queryBuilder->expr()->lt('scheduled', time())
225		)
226	1	->groupBy('configuration')
227	1	->execute();
228
229	1	return $statement->fetchAll();
230		}
231
232		/**
233		* Get set id with unprocessed entries
234		*
235		* @return array array of set ids
236		*/
237	1	public function getSetIdWithUnprocessedEntries(): array
238		{
239	1	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
240		$statement = $queryBuilder
241	1	->select('set_id')
242	1	->from($this->tableName)
243	1	->where(
244	1	$queryBuilder->expr()->lt('scheduled', time()),
245	1	$queryBuilder->expr()->eq('exec_time', 0)
246		)
247	1	->addGroupBy('set_id')
248	1	->execute();
249
250	1	$setIds = [];
251	1	while ($row = $statement->fetch()) {
252	1	$setIds[] = intval($row['set_id']);
253		}
254
255	1	return $setIds;
256		}
257
258		/**
259		* Get total queue entries by configuration
260		*
261		* @return array totals by configuration (keys)
262		*/
263	1	public function getTotalQueueEntriesByConfiguration(array $setIds): array
264		{
265	1	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
266	1	$totals = [];
267	1	if (count($setIds) > 0) {
268		$statement = $queryBuilder
269	1	->from($this->tableName)
270	1	->selectLiteral('count(*) as c')
271	1	->addSelect('configuration')
272	1	->where(
273	1	$queryBuilder->expr()->in('set_id', implode(',', $setIds)),
274	1	$queryBuilder->expr()->lt('scheduled', time())
275		)
276	1	->groupBy('configuration')
277	1	->execute();
278
279	1	while ($row = $statement->fetch()) {
280	1	$totals[$row['configuration']] = $row['c'];
281		}
282		}
283
284	1	return $totals;
285		}
286
287		/**
288		* Get the timestamps of the last processed entries
289		*
290		* @param int $limit
291		*/
292	1	public function getLastProcessedEntriesTimestamps($limit = 100): array
293		{
294	1	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
295		$statement = $queryBuilder
296	1	->select('exec_time')
297	1	->from($this->tableName)
298	1	->addOrderBy('exec_time', 'desc')
299	1	->setMaxResults($limit)
300	1	->execute();
301
302	1	$rows = [];
303	1	while ($row = $statement->fetch()) {
304	1	$rows[] = $row['exec_time'];
305		}
306
307	1	return $rows;
308		}
309
310		/**
311		* Get the last processed entries
312		*
313		* @param int $limit
314		*/
315	1	public function getLastProcessedEntries($limit = 100): array
316		{
317	1	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
318		$statement = $queryBuilder
319	1	->from($this->tableName)
320	1	->select('*')
321	1	->orderBy('exec_time', 'desc')
322	1	->setMaxResults($limit)
323	1	->execute();
324
325	1	$rows = [];
326	1	while (($row = $statement->fetch()) !== false) {
327	1	$rows[] = $row;
328		}
329
330	1	return $rows;
331		}
332
333		/**
334		* Get performance statistics data
335		*
336		* @param int $start timestamp
337		* @param int $end timestamp
338		*
339		* @return array performance data
340		*/
341	1	public function getPerformanceData($start, $end): array
342		{
343	1	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
344		$statement = $queryBuilder
345	1	->from($this->tableName)
346	1	->selectLiteral('min(exec_time) as start', 'max(exec_time) as end', 'count(*) as urlcount')
347	1	->addSelect('process_id_completed')
348	1	->where(
349	1	$queryBuilder->expr()->neq('exec_time', 0),
350	1	$queryBuilder->expr()->gte('exec_time', $queryBuilder->createNamedParameter($start, \PDO::PARAM_INT)),
351	1	$queryBuilder->expr()->lte('exec_time', $queryBuilder->createNamedParameter($end, \PDO::PARAM_INT))
352		)
353	1	->groupBy('process_id_completed')
354	1	->execute();
355
356	1	$rows = [];
357	1	while ($row = $statement->fetch()) {
358	1	$rows[$row['process_id_completed']] = $row;
359		}
360
361	1	return $rows;
362		}
363
364		/**
365		* Determines if a page is queued
366		*/
367	5	public function isPageInQueue(int $uid, bool $unprocessed_only = true, bool $timed_only = false, int $timestamp = 0): bool
368		{
369	5	$isPageInQueue = false;
370
371	5	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
372		$statement = $queryBuilder
373	5	->from($this->tableName)
374	5	->count('*')
375	5	->where(
376	5	$queryBuilder->expr()->eq('page_id', $queryBuilder->createNamedParameter($uid, \PDO::PARAM_INT))
377		);
378
379	5	if ($unprocessed_only !== false) {
380	2	$statement->andWhere(
381	2	$queryBuilder->expr()->eq('exec_time', 0)
382		);
383		}
384
385	5	if ($timed_only !== false) {
386	1	$statement->andWhere(
387	1	$queryBuilder->expr()->neq('scheduled', 0)
388		);
389		}
390
391	5	if ($timestamp) {
392	1	$statement->andWhere(
393	1	$queryBuilder->expr()->eq('scheduled', $queryBuilder->createNamedParameter($timestamp, \PDO::PARAM_INT))
394		);
395		}
396
397		// TODO: Currently it's not working if page doesn't exists. See tests
398		$count = $statement
399	5	->execute()
400	5	->fetchColumn(0);
401
402	5	if ($count !== false && $count > 0) {
403	4	$isPageInQueue = true;
404		}
405
406	5	return $isPageInQueue;
407		}
408
409		/**
410		* Method to check if a page is in the queue which is timed for a
411		* date when it should be crawled
412		*/
413	1	public function isPageInQueueTimed(int $uid, bool $show_unprocessed = true): bool
414		{
415	1	return $this->isPageInQueue($uid, $show_unprocessed);
416		}
417
418	1	public function getAvailableSets(): array
419		{
420	1	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
421		$statement = $queryBuilder
422	1	->selectLiteral('count(*) as count_value')
423	1	->addSelect('set_id', 'scheduled')
424	1	->from($this->tableName)
425	1	->orderBy('scheduled', 'desc')
426	1	->groupBy('set_id', 'scheduled')
427	1	->execute();
428
429	1	$rows = [];
430	1	while ($row = $statement->fetch()) {
431	1	$rows[] = $row;
432		}
433
434	1	return $rows;
435		}
436
437	1	public function findByQueueId(string $queueId): ?array
438		{
439	1	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
440		$queueRec = $queryBuilder
441	1	->select('*')
442	1	->from($this->tableName)
443	1	->where(
444	1	$queryBuilder->expr()->eq('qid', $queryBuilder->createNamedParameter($queueId))
445		)
446	1	->execute()
447	1	->fetch();
448	1	return is_array($queueRec) ? $queueRec : null;
449		}
450
451	1	public function cleanupQueue(): void
452		{
453	1	$extensionSettings = GeneralUtility::makeInstance(ExtensionConfigurationProvider::class)->getExtensionConfiguration();
454	1	$purgeDays = (int) $extensionSettings['purgeQueueDays'];
455
456	1	if ($purgeDays > 0) {
457	1	$purgeDate = time() - 24 * 60 * 60 * $purgeDays;
458
459	1	$queryBuilderDelete = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
460		$del = $queryBuilderDelete
461	1	->delete($this->tableName)
462	1	->where(
463	1	'exec_time != 0 AND exec_time < ' . $purgeDate
464	1	)->execute();
465
466	1	if ($del === false) {
467		$this->logger->info(
468		'Records could not be deleted.'
469		);
470		}
471		}
472	1	}
473
474		/**
475		* Cleans up entries that stayed for too long in the queue. These are default:
476		* - processed entries that are over 1.5 days in age
477		* - scheduled entries that are over 7 days old
478		*/
479	1	public function cleanUpOldQueueEntries(): void
480		{
481	1	$extensionSettings = GeneralUtility::makeInstance(ExtensionConfigurationProvider::class)->getExtensionConfiguration();
482	1	$processedAgeInSeconds = $extensionSettings['cleanUpProcessedAge'] * 86400; // 246060 Seconds in 24 hours
483	1	$scheduledAgeInSeconds = $extensionSettings['cleanUpScheduledAge'] * 86400;
484
485	1	$now = time();
486	1	$condition = '(exec_time<>0 AND exec_time<' . ($now - $processedAgeInSeconds) . ') OR scheduled<=' . ($now - $scheduledAgeInSeconds);
487
488	1	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
489		$del = $queryBuilder
490	1	->delete($this->tableName)
491	1	->where(
492	1	$condition
493	1	)->execute();
494
495	1	if ($del === false) {
496		$this->logger->info(
497		'Records could not be deleted.'
498		);
499		}
500	1	}
501
502	2	public function fetchRecordsToBeCrawled(int $countInARun)
503		{
504	2	$queryBuilderSelect = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
505		return $queryBuilderSelect
506	2	->select('qid', 'scheduled', 'page_id', 'sitemap_priority')
507	2	->from($this->tableName)
508	2	->leftJoin(
509	2	$this->tableName,
510	2	'pages',
511	2	'p',
512	2	$queryBuilderSelect->expr()->eq('p.uid', $queryBuilderSelect->quoteIdentifier($this->tableName . '.page_id'))
513		)
514	2	->where(
515	2	$queryBuilderSelect->expr()->eq('exec_time', 0),
516	2	$queryBuilderSelect->expr()->eq('process_scheduled', 0),
517	2	$queryBuilderSelect->expr()->lte('scheduled', time())
518		)
519	2	->orderBy('sitemap_priority', 'DESC')
520	2	->addOrderBy('scheduled')
521	2	->addOrderBy('qid')
522	2	->setMaxResults($countInARun)
523	2	->execute()
524	2	->fetchAll();
525		}
526
527	1	public function updateProcessIdAndSchedulerForQueueIds(array $quidList, string $processId)
528		{
529	1	$queryBuilderUpdate = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
530		return $queryBuilderUpdate
531	1	->update($this->tableName)
532	1	->where(
533	1	$queryBuilderUpdate->expr()->in('qid', $quidList)
534		)
535	1	->set('process_scheduled', time())
536	1	->set('process_id', $processId)
537	1	->execute();
538		}
539
540	1	public function unsetProcessScheduledAndProcessIdForQueueEntries(array $processIds): void
541		{
542	1	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
543		$queryBuilder
544	1	->update($this->tableName)
545	1	->where(
546	1	$queryBuilder->expr()->eq('exec_time', 0),
547	1	$queryBuilder->expr()->in('process_id', $queryBuilder->createNamedParameter($processIds, Connection::PARAM_STR_ARRAY))
548		)
549	1	->set('process_scheduled', 0)
550	1	->set('process_id', '')
551	1	->execute();
552	1	}
553
554		/**
555		* This method is used to count if there are ANY unprocessed queue entries
556		* of a given page_id and the configuration which matches a given hash.
557		* If there if none, we can skip an inner detail check
558		*
559		* @param int $uid
560		* @param string $configurationHash
561		* @return boolean
562		*/
563	4	public function noUnprocessedQueueEntriesForPageWithConfigurationHashExist($uid, $configurationHash): bool
564		{
565	4	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
566	4	$noUnprocessedQueueEntriesFound = true;
567
568		$result = $queryBuilder
569	4	->count('*')
570	4	->from($this->tableName)
571	4	->where(
572	4	$queryBuilder->expr()->eq('page_id', (int) $uid),
573	4	$queryBuilder->expr()->eq('configuration_hash', $queryBuilder->createNamedParameter($configurationHash)),
574	4	$queryBuilder->expr()->eq('exec_time', 0)
575		)
576	4	->execute()
577	4	->fetchColumn();
578
579	4	if ($result) {
580	4	$noUnprocessedQueueEntriesFound = false;
581		}
582
583	4	return $noUnprocessedQueueEntriesFound;
584		}
585
586		/**
587		* Removes queue entries
588		*
589		* @param string $filter all, pending, finished
590		*/
591	6	public function flushQueue(string $filter): void
592		{
593	6	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
594
595	6	switch (strtolower($filter)) {
596	6	case 'all':
597		// No where claus needed delete everything
598	2	break;
599	4	case 'pending':
600	1	$queryBuilder->andWhere($queryBuilder->expr()->eq('exec_time', 0));
601	1	break;
602	3	case 'finished':
603		default:
604	3	$queryBuilder->andWhere($queryBuilder->expr()->gt('exec_time', 0));
605	3	break;
606		}
607
608		$queryBuilder
609	6	->delete($this->tableName)
610	6	->execute();
611	6	}
612
613		/**
614		* @param string $processId
615		*
616		* @return bool\|string
617		*/
618	1	public function countAllByProcessId($processId)
619		{
620	1	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
621
622		return $queryBuilder
623	1	->count('*')
624	1	->from($this->tableName)
625	1	->where(
626	1	$queryBuilder->expr()->eq('process_id', $queryBuilder->createNamedParameter($processId, \PDO::PARAM_STR))
627		)
628	1	->execute()
629	1	->fetchColumn(0);
630		}
631
632		/**
633		* This internal helper method is used to create an instance of an entry object
634		*
635		* @param Process $process
636		* @param string $orderByField first matching item will be returned as object
637		* @param string $orderBySorting sorting direction
638		*/
639	5	protected function getFirstOrLastObjectByProcess($process, $orderByField, $orderBySorting = 'ASC'): array
640		{
641	5	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
642		$first = $queryBuilder
643	5	->select('*')
644	5	->from($this->tableName)
645	5	->where(
646	5	$queryBuilder->expr()->eq('process_id_completed', $queryBuilder->createNamedParameter($process->getProcessId())),
647	5	$queryBuilder->expr()->gt('exec_time', 0)
648		)
649	5	->setMaxResults(1)
650	5	->addOrderBy($orderByField, $orderBySorting)
651	5	->execute()->fetch(0);
652
653	5	return $first ?: [];
654		}
655		}
656

AOEpeople / crawler

Pull Request — master (#621)

QueueRepository::findOldestEntryForProcess() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like