QueueRepository::countUnprocessedItems() - Code Metrics - Inspection of "Merge branch 'master' into deprecations" - AOEpeople/crawler - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — deprecations ( 3cb403...87297a )

by Tomas Norre

created 2020-04-27 14:56 UTC

QueueRepository::countUnprocessedItems() 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.037

Importance

Changes

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

<?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.'
                );
            }
        }
    }

    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 $where SQL related filter for the entries which should be removed
     */
    public function flushQueue($where = ''): void
    {
        $realWhere = strlen((string) $where) > 0 ? $where : '1=1';
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);

        if (EventDispatcher::getInstance()->hasObserver('queueEntryFlush')) {
            $groups = $queryBuilder
                ->select('DISTINCT set_id')
                ->from($this->tableName)
                ->where($realWhere)
                ->execute()
                ->fetchAll();
            if (is_array($groups)) {
                foreach ($groups as $group) {
                    $subSet = $queryBuilder
                        ->select('uid', 'set_id')
                        ->from($this->tableName)
                        ->where(
                            $realWhere,
                            $queryBuilder->expr()->eq('set_id', $group['set_id'])
                        )
                        ->execute()
                        ->fetchAll();
                    EventDispatcher::getInstance()->post('queueEntryFlush', $group['set_id'], $subSet);
                }
            }
        }

        $queryBuilder
            ->delete($this->tableName)
            ->where($realWhere)
            ->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	88	public function __construct()
55		{
56		// Left empty intentional
57	88	}
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	1	public function fetchRecordsToBeCrawled(int $countInARun)
479		{
480	1	$queryBuilderSelect = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
481		$rows = $queryBuilderSelect
482	1	->select('qid', 'scheduled')
483	1	->from($this->tableName)
484	1	->where(
485	1	$queryBuilderSelect->expr()->eq('exec_time', 0),
486	1	$queryBuilderSelect->expr()->eq('process_scheduled', 0),
487	1	$queryBuilderSelect->expr()->lte('scheduled', time())
488		)
489	1	->orderBy('scheduled')
490	1	->addOrderBy('qid')
491	1	->setMaxResults($countInARun)
492	1	->execute()
493	1	->fetchAll();
494	1	return $rows;
495		}
496
497	1	public function updateProcessIdAndSchedulerForQueueIds(array $quidList, string $processId)
498		{
499	1	$queryBuilderUpdate = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
500		$numberOfAffectedRows = $queryBuilderUpdate
501	1	->update($this->tableName)
502	1	->where(
503	1	$queryBuilderUpdate->expr()->in('qid', $quidList)
504		)
505	1	->set('process_scheduled', time())
506	1	->set('process_id', $processId)
507	1	->execute();
508	1	return $numberOfAffectedRows;
509		}
510
511	1	public function unsetProcessScheduledAndProcessIdForQueueEntries(array $processIds): void
512		{
513	1	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
514		$queryBuilder
515	1	->update($this->tableName)
516	1	->where(
517	1	$queryBuilder->expr()->eq('exec_time', 0),
518	1	$queryBuilder->expr()->in('process_id', $queryBuilder->createNamedParameter($processIds, Connection::PARAM_STR_ARRAY))
519		)
520	1	->set('process_scheduled', 0)
521	1	->set('process_id', '')
522	1	->execute();
523	1	}
524
525		/**
526		* This method is used to count if there are ANY unprocessed queue entries
527		* of a given page_id and the configuration which matches a given hash.
528		* If there if none, we can skip an inner detail check
529		*
530		* @param int $uid
531		* @param string $configurationHash
532		* @return boolean
533		*/
534	2	public function noUnprocessedQueueEntriesForPageWithConfigurationHashExist($uid, $configurationHash): bool
535		{
536	2	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
537	2	$noUnprocessedQueueEntriesFound = true;
538
539		$result = $queryBuilder
540	2	->count('*')
541	2	->from($this->tableName)
542	2	->where(
543	2	$queryBuilder->expr()->eq('page_id', (int) $uid),
544	2	$queryBuilder->expr()->eq('configuration_hash', $queryBuilder->createNamedParameter($configurationHash)),
545	2	$queryBuilder->expr()->eq('exec_time', 0)
546		)
547	2	->execute()
548	2	->fetchColumn();
549
550	2	if ($result) {
551	2	$noUnprocessedQueueEntriesFound = false;
552		}
553
554	2	return $noUnprocessedQueueEntriesFound;
555		}
556
557		/**
558		* Removes queue entries
559		*
560		* @param string $where SQL related filter for the entries which should be removed
561		*/
562	11	public function flushQueue($where = ''): void
563		{
564	11	$realWhere = strlen((string) $where) > 0 ? $where : '1=1';
565	11	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
566
567	11	if (EventDispatcher::getInstance()->hasObserver('queueEntryFlush')) {
568		$groups = $queryBuilder
569		->select('DISTINCT set_id')
570		->from($this->tableName)
571		->where($realWhere)
572		->execute()
573		->fetchAll();
574		if (is_array($groups)) {
575		foreach ($groups as $group) {
576		$subSet = $queryBuilder
577		->select('uid', 'set_id')
578		->from($this->tableName)
579		->where(
580		$realWhere,
581		$queryBuilder->expr()->eq('set_id', $group['set_id'])
582		)
583		->execute()
584		->fetchAll();
585		EventDispatcher::getInstance()->post('queueEntryFlush', $group['set_id'], $subSet);
586		}
587		}
588		}
589
590		$queryBuilder
591	11	->delete($this->tableName)
592	11	->where($realWhere)
593	11	->execute();
594	11	}
595
596		/**
597		* This internal helper method is used to create an instance of an entry object
598		*
599		* @param Process $process
600		* @param string $orderByField first matching item will be returned as object
601		* @param string $orderBySorting sorting direction
602		*/
603	5	protected function getFirstOrLastObjectByProcess($process, $orderByField, $orderBySorting = 'ASC'): array
604		{
605	5	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
606		$first = $queryBuilder
607	5	->select('*')
608	5	->from($this->tableName)
609	5	->where(
610	5	$queryBuilder->expr()->eq('process_id_completed', $queryBuilder->createNamedParameter($process->getProcessId())),
611	5	$queryBuilder->expr()->gt('exec_time', 0)
612		)
613	5	->setMaxResults(1)
614	5	->addOrderBy($orderByField, $orderBySorting)
615	5	->execute()->fetch(0);
616
617	5	return $first ?: [];
618		}
619		}
620

AOEpeople / crawler

Push — deprecations ( 3cb403...87297a )

QueueRepository::countUnprocessedItems() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like