QueueRepository::cleanUpOldQueueEntries() - Code Metrics - Inspection of "[TASK] Added cleanUpOldQueueEntries to QueueReposi..." - AOEpeople/crawler - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — cleanup/crawlercontroller ( 49e992 )

by Tomas Norre

created 2020-04-30 19:06 UTC

QueueRepository::cleanUpOldQueueEntries() A

↳ Parent: QueueRepository

Complexity

Conditions	2
Paths	2

Size

Total Lines	19
Code Lines	14

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	12
CRAP Score	2.0625

Importance

Changes

Metric	Value
cc	2
eloc	14
c	0
b	0
f	0
nc	2
nop	0
dl	0
loc	19
ccs	12
cts	16
cp	0.75
crap	2.0625
rs	9.7998

<?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 AOE\Crawler\Event\EventDispatcher;
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\Core\Utility\MathUtility;

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

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

    public function __construct()
    {
        // Left empty intentional
    }

    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
    {
        // TODO: Looks like the check is obsolete as PHP doesn't allow other than ints for the $uid, PHP 7.2 Strict Mode
        if (! MathUtility::canBeInterpretedAsInteger($uid)) {
            throw new \InvalidArgumentException('Invalid parameter type', 1468931945);
        }

        $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);
        $rows = $queryBuilderSelect
            ->select('qid', 'scheduled')
            ->from($this->tableName)
            ->where(
                $queryBuilderSelect->expr()->eq('exec_time', 0),
                $queryBuilderSelect->expr()->eq('process_scheduled', 0),
                $queryBuilderSelect->expr()->lte('scheduled', time())
            )
            ->orderBy('scheduled')
            ->addOrderBy('qid')
            ->setMaxResults($countInARun)
            ->execute()
            ->fetchAll();
        return $rows;
    }

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

    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();
    }

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

AOEpeople / crawler

Push — cleanup/crawlercontroller ( 49e992 )

QueueRepository::cleanUpOldQueueEntries() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like