JobManager::addWorkerNameCriterion() - Code Metrics - mmucklo/DtcQueueBundle - Measure and Improve Code Quality continuously with Scrutinizer

JobManager::addWorkerNameCriterion() A
last analyzed 2025-06-22 22:56 UTC

↳ Parent: JobManager

Complexity

Conditions	3
Paths	4

Size

Total Lines	8
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	5
CRAP Score	3

Importance

Changes

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

<?php

namespace Dtc\QueueBundle\ODM;

use Doctrine\MongoDB\Exception\ResultException;
filter:
    dependency_paths: ["lib/*"]
use Doctrine\MongoDB\Query\Builder;
filter:
    dependency_paths: ["lib/*"]
use Doctrine\ODM\MongoDB\DocumentManager;
filter:
    dependency_paths: ["lib/*"]
use Dtc\QueueBundle\Doctrine\DoctrineJobManager;
use Dtc\QueueBundle\Document\Job;
use Dtc\QueueBundle\Model\BaseJob;
use Dtc\QueueBundle\Util\Util;
use MongoDB\DeleteResult;
filter:
    dependency_paths: ["lib/*"]
use MongoDB\UpdateResult;
filter:
    dependency_paths: ["lib/*"]

class JobManager extends DoctrineJobManager
{
    use CommonTrait;
    public const REDUCE_FUNCTION = 'function(k, vals) {
            var result = {};
            for (var index in vals) {
                var val = vals[index];
                for (var i in val) {
                    if (result.hasOwnProperty(i)) {
                        result[i] += val[i];
                    }
                    else {
                        result[i] = val[i];
                    }
                }
            }
            return result;
        }';

    public function countJobsByStatus($objectName, $status, $workerName = null, $method = null)
    {
        /** @var DocumentManager $objectManager */
        $objectManager = $this->getObjectManager();
        $qb = $objectManager->createQueryBuilder($objectName);
        $qb
            ->find()
            ->field('status')->equals($status);

        $this->addWorkerNameCriterion($qb, $workerName, $method);
        $method = 'count';
        if (method_exists($qb, 'count')) {
            $method = 'execute';
            $qb->count();
        }
        $query = $qb->getQuery();

        return $this->runQuery($query, $method, [], 0);
    }

    /**
     * @param string|null $workerName
     * @param string|null $method
     */
    public function pruneExceptionJobs($workerName = null, $method = null)
    {
        /** @var DocumentManager $objectManager */
        $objectManager = $this->getObjectManager();
        $qb = $objectManager->createQueryBuilder($this->getJobArchiveClass());
        $qb = $qb->remove();
        $qb->field('status')->equals(BaseJob::STATUS_EXCEPTION);
        $this->addWorkerNameCriterion($qb, $workerName, $method);

        $query = $qb->getQuery();
        $result = $this->runQuery($query, 'execute');
        if ($result instanceof DeleteResult) {
            return $result->getDeletedCount();
        } elseif (isset($result['n'])) {
            return $result['n'];
        }

        return 0;
    }

    /**
     * @param $builder
     * @param string|null $workerName
     * @param string|null $method
     */
    protected function addWorkerNameCriterion($builder, $workerName = null, $method = null)
    {
        if (null !== $workerName) {
            $builder->field('workerName')->equals($workerName);
        }

        if (null !== $method) {
            $builder->field('method')->equals($method);
        }
    }

    /**
     * @param null $workerName

     * @param null $method

     *
     * @return int
     */
    protected function updateExpired($workerName = null, $method = null)
    {
        /** @var DocumentManager $objectManager */
        $objectManager = $this->getObjectManager();
        $qb = $objectManager->createQueryBuilder($this->getJobClass());
        $qb = $qb->updateMany();
        $qb->field('expiresAt')->lte(Util::getMicrotimeDateTime());
        $qb->field('status')->equals(BaseJob::STATUS_NEW);
        $this->addWorkerNameCriterion($qb, $workerName, $method);
        $qb->field('status')->set(\Dtc\QueueBundle\Model\Job::STATUS_EXPIRED);
        $query = $qb->getQuery();
        $result = $this->runQuery($query, 'execute');
        if ($result instanceof UpdateResult) {
            return $result->getModifiedCount();
        } elseif (isset($result['n'])) {
            return $result['n'];
        }

        return 0;
    }

    /**
     * Removes archived jobs older than $olderThan.
     *
     * @return int
     */
    public function pruneArchivedJobs(\DateTime $olderThan)
    {
        return $this->removeOlderThan($this->getJobArchiveClass(), 'updatedAt', $olderThan);
    }

    public function getWaitingJobCount($workerName = null, $method = null)
    {
        /** @var DocumentManager $objectManager */
        $objectManager = $this->getObjectManager();
        $builder = $objectManager->createQueryBuilder($this->getJobClass());
        $builder
            ->find();

        $this->addWorkerNameCriterion($builder, $workerName, $method);
        $this->addStandardPredicates($builder);
        $method = 'count';
        if (method_exists($builder, 'count')) {
            $builder->count();
            $method = 'execute';
        }
        $query = $builder->getQuery();

        return $this->runQuery($query, $method, [true], 0);
    }

    protected function getStatusByDocumentMapReduce($builder, $documentName)
    {
        $reduceFunc = self::REDUCE_FUNCTION;
        $mapFunc = "function() {
            var result = {};
            result[this.status] = 1;
            var key = this.worker_name + '->' + this.method + '()';
            emit(key, result);
            }";
        $builder->map($mapFunc)
            ->reduce($reduceFunc);
        $query = $builder->getQuery();
        $results = $this->runQuery($query, 'execute', [], []);
        $allStatus = static::getAllStatuses();

        $status = [];

        foreach ($results as $info) {
            $status[$info['_id']] = $info['value'] + $allStatus;
        }

        return $status;
    }

    /**
     * Get Status Jobs.
     *
     * @param string $documentName
     *
     * @return array
     */
    protected function getStatusByDocument($documentName)
    {
        // Run a map reduce function get worker and status break down
        /** @var DocumentManager $objectManager */
        $objectManager = $this->getObjectManager();
        $builder = $objectManager->createQueryBuilder($documentName);
        if (method_exists($builder, 'map')) {
            return $this->getStatusByDocumentMapReduce($builder, $documentName);
        }

        $aggregation = $objectManager->createAggregationBuilder($documentName);
        $aggregation->group()
            ->field('id')->expression(
                $aggregation->expr()
                                                ->field('worker_name')
                                                ->expression('$worker_name')
                                                ->field('method')
                                                ->expression('$method')
                                                ->field('status')
                                                ->expression('$status')
            )
            ->field('value')
            ->sum(1);
        $results = $this->runQuery($aggregation, 'execute', [], []);
        $allStatus = static::getAllStatuses();

        $status = [];

        foreach ($results as $info) {
            $key = $info['_id']['worker_name'].'->'.$info['_id']['method'].'()';
            if (!isset($status[$key])) {
                $status[$key] = $allStatus;
            }
            $status[$key][$info['_id']['status']] = $info['value'];
        }

        return $status;
    }

    public function getStatus(): array
    {
        $result = $this->getStatusByDocument($this->getJobClass());
        $status2 = $this->getStatusByDocument($this->getJobArchiveClass());
        foreach ($status2 as $key => $value) {
            foreach ($value as $k => $v) {
                if (isset($result[$key][$k])) {
                    $result[$key][$k] += $v;
                } else {
                    $result[$key][$k] = $v;
                }
            }
        }

        $finalResult = [];
        foreach ($result as $key => $item) {
            ksort($item);
            $finalResult[$key] = $item;
        }

        return $finalResult;
    }

    /**
     * Get the next job to run (can be filtered by workername and method name).
     *
     * @param string      $workerName
     * @param string      $methodName
     * @param bool        $prioritize
     * @param string|null $runId
     *
     * @return \Dtc\QueueBundle\Model\Job
     */
    public function getJob($workerName = null, $methodName = null, $prioritize = true, $runId = null)
    {
        $builder = $this->getJobQueryBuilder($workerName, $methodName, $prioritize);
        $builder
            ->findAndUpdate()
            ->returnNew();

        $date = Util::getMicrotimeDateTime();
        // Update
        $builder
            ->field('startedAt')->set($date)
            ->field('status')->set(BaseJob::STATUS_RUNNING);
        if (null !== $runId) {
            $builder->field('runId')->set($runId);
        }

        $query = $builder->getQuery();

        $job = $this->runQuery($query, 'execute');

        return $job;
    }

    protected function runQuery($query, $method, array $arguments = [], $resultIfNamespaceError = null)
    {
        try {
            $result = call_user_func_array([$query, $method], $arguments);
        } catch (ResultException $resultException) {
            if (false === strpos($resultException->getMessage(), 'namespace does not exist') && false === strpos($resultException->getMessage(), 'ns doesn\'t exist')) {
                throw $resultException;
            }
            $result = $resultIfNamespaceError;
        }

        return $result;
    }

    /**
     * @param string|null $workerName
     * @param string|null $methodName
     * @param bool        $prioritize
     *
     * @return Builder
     */
    public function getJobQueryBuilder($workerName = null, $methodName = null, $prioritize = true)
    {
        /** @var DocumentManager $objectManager */
        $objectManager = $this->getObjectManager();
        $builder = $objectManager->createQueryBuilder($this->getJobClass());

        $this->addWorkerNameCriterion($builder, $workerName, $methodName);
        if ($prioritize) {
            $builder->sort([
                'priority' => 'desc',
                'whenAt' => 'asc',
            ]);
        } else {
            $builder->sort('whenAt', 'asc');
        }

        // Filter
        $this->addStandardPredicates($builder);

        return $builder;
    }

    protected function updateNearestBatch(\Dtc\QueueBundle\Model\Job $job)
    {
        /** @var DocumentManager $objectManager */
        $objectManager = $this->getObjectManager();
        $builder = $objectManager->createQueryBuilder($this->getJobClass());
        $builder->find();

        $builder->sort('whenAt', 'asc');
        $builder->field('status')->equals(BaseJob::STATUS_NEW)
            ->field('crcHash')->equals($job->getCrcHash());
        $oldJob = $this->runQuery($builder->getQuery(), 'getSingleResult');

        if (!$oldJob) {
            return null;
        }

        // Update priority or whenAt
        //  This makes sure if someone else is updating at the same time
        //  that we don't trounce their changes.
        $builder = $objectManager->createQueryBuilder($this->getJobClass());
        $builder->findAndUpdate();
        $builder->field('_id')->equals($oldJob->getId());
        $builder->field('priority')->lt($job->getPriority());
        $builder->field('priority')->set($job->getPriority());
        $this->runQuery($builder->getQuery(), 'execute');

        $builder = $objectManager->createQueryBuilder($this->getJobClass());
        $builder->findAndUpdate();
        $builder->field('_id')->equals($oldJob->getId());
        $builder->field('whenAt')->gt($job->getWhenAt());
        $builder->field('whenAt')->set($job->getWhenAt());
        $this->runQuery($builder->getQuery(), 'execute');

        if ($job->getWhenAt() < $oldJob->getWhenAt()) {
            $oldJob->setWhenAt($job->getWhenAt());
        }
        if ($job->getPriority() > $oldJob->getPriority()) {
            $oldJob->setPriority($job->getPriority());
        }

        return $oldJob;
    }

    /**
     * @param mixed $builder
     */
    protected function addStandardPredicates($builder)
    {
        $date = Util::getMicrotimeDateTime();
        $builder
            ->addAnd(
                $builder->expr()->addOr($builder->expr()->field('whenAt')->equals(null), $builder->expr()->field('whenAt')->lte($date)),
                $builder->expr()->addOr($builder->expr()->field('expiresAt')->equals(null), $builder->expr()->field('expiresAt')->gt($date))
            )
            ->field('status')->equals(BaseJob::STATUS_NEW);
    }

    public function getWorkersAndMethods()
    {
        /** @var DocumentManager $documentManager */
        $documentManager = $this->getObjectManager();

        if (!method_exists($documentManager, 'createAggregationBuilder')) {
            return [];
        }

        $aggregationBuilder = $documentManager->createAggregationBuilder($this->getJobClass());

        $this->addStandardPredicates($aggregationBuilder->match());

        $aggregationBuilder->group()
            ->field('id')
            ->expression(
                $aggregationBuilder->expr()
                ->field('workerName')->expression('$workerName')
                ->field('method')->expression('$method')
            );
        $results = $aggregationBuilder->execute()->toArray();

        if (!$results) {
            return [];
        }

        $workersMethods = [];
        foreach ($results as $result) {
            if (isset($result['_id'])) {
                $workersMethods[$result['_id']['worker_name']][] = $result['_id']['method'];
            }
        }

        return $workersMethods;
    }

    /**
     * @param string $workerName
     * @param string $methodName
     */
    public function countLiveJobs($workerName = null, $methodName = null)
    {
        /** @var DocumentManager $objectManager */
        $objectManager = $this->getObjectManager();
        $builder = $objectManager->createQueryBuilder($this->getJobClass());

        $this->addWorkerNameCriterion($builder, $workerName, $methodName);
        // Filter
        $this->addStandardPredicates($builder);
        $method = 'count';
        if (method_exists($builder, 'count')) {
            $builder->count();
            $method = 'execute';
        }

        return $this->runQuery($builder->getQuery(), $method, [], 0);
    }

    /**
     * @param string $workerName
     * @param string $methodName
     */
    public function archiveAllJobs($workerName = null, $methodName = null, callable $progressCallback = null)
    {
        /** @var DocumentManager $documentManager */
        $documentManager = $this->getObjectManager();
        $count = 0;
        $builder = $this->getJobQueryBuilder($workerName, $methodName, true);
        $builder
            ->findAndUpdate()
            ->returnNew();

        $builder->field('status')->set(Job::STATUS_ARCHIVE);
        $query = $builder->getQuery();
        do {
            $job = $this->runQuery($query, 'execute');
            if ($job) {
                $documentManager->remove($job);
                ++$count;

                if (0 == $count % 10) {
                    $this->flush();
                    $this->updateProgress($progressCallback, $count);
                }
            }
        } while ($job);
        $this->flush();
        $this->updateProgress($progressCallback, $count);
    }
}


1		<?php
2
3		namespace Dtc\QueueBundle\ODM;
4
5		use Doctrine\MongoDB\Exception\ResultException;
		0 ignored issues – show Bug introduced 2019-12-20 07:38 UTC by Report Bug Copy Issue Report The type `Doctrine\MongoDB\Exception\ResultException` was not found. Maybe you did not declare it correctly or list all dependencies? The issue could also be caused by a filter entry in the build configuration. If the path has been excluded in your configuration, e.g. `excluded_paths: ["lib/"]`, you can move it to the dependency path list as follows: filter: dependency_paths: ["lib/"] For further information see https://scrutinizer-ci.com/docs/tools/php/php-scrutinizer/#list-dependency-paths Loading history...
6		use Doctrine\MongoDB\Query\Builder;
		0 ignored issues – show Bug introduced 2019-12-20 07:38 UTC by Report Bug Copy Issue Report The type `Doctrine\MongoDB\Query\Builder` was not found. Maybe you did not declare it correctly or list all dependencies? The issue could also be caused by a filter entry in the build configuration. If the path has been excluded in your configuration, e.g. `excluded_paths: ["lib/"]`, you can move it to the dependency path list as follows: filter: dependency_paths: ["lib/"] For further information see https://scrutinizer-ci.com/docs/tools/php/php-scrutinizer/#list-dependency-paths Loading history...
7		use Doctrine\ODM\MongoDB\DocumentManager;
		0 ignored issues – show Bug introduced 2020-08-09 07:45 UTC by Report Bug Copy Issue Report The type `Doctrine\ODM\MongoDB\DocumentManager` was not found. Maybe you did not declare it correctly or list all dependencies? The issue could also be caused by a filter entry in the build configuration. If the path has been excluded in your configuration, e.g. `excluded_paths: ["lib/"]`, you can move it to the dependency path list as follows: filter: dependency_paths: ["lib/"] For further information see https://scrutinizer-ci.com/docs/tools/php/php-scrutinizer/#list-dependency-paths Loading history...
8		use Dtc\QueueBundle\Doctrine\DoctrineJobManager;
9		use Dtc\QueueBundle\Document\Job;
10		use Dtc\QueueBundle\Model\BaseJob;
11		use Dtc\QueueBundle\Util\Util;
12		use MongoDB\DeleteResult;
		0 ignored issues – show Bug introduced 2020-08-09 07:45 UTC by Report Bug Copy Issue Report The type `MongoDB\DeleteResult` was not found. Maybe you did not declare it correctly or list all dependencies? The issue could also be caused by a filter entry in the build configuration. If the path has been excluded in your configuration, e.g. `excluded_paths: ["lib/"]`, you can move it to the dependency path list as follows: filter: dependency_paths: ["lib/"] For further information see https://scrutinizer-ci.com/docs/tools/php/php-scrutinizer/#list-dependency-paths Loading history...
13		use MongoDB\UpdateResult;
		0 ignored issues – show Bug introduced 2020-08-09 07:45 UTC by Report Bug Copy Issue Report The type `MongoDB\UpdateResult` was not found. Maybe you did not declare it correctly or list all dependencies? The issue could also be caused by a filter entry in the build configuration. If the path has been excluded in your configuration, e.g. `excluded_paths: ["lib/"]`, you can move it to the dependency path list as follows: filter: dependency_paths: ["lib/"] For further information see https://scrutinizer-ci.com/docs/tools/php/php-scrutinizer/#list-dependency-paths Loading history...
14
15		class JobManager extends DoctrineJobManager
16		{
17	1	use CommonTrait;
18		public const REDUCE_FUNCTION = 'function(k, vals) {
19		var result = {};
20		for (var index in vals) {
21		var val = vals[index];
22		for (var i in val) {
23		if (result.hasOwnProperty(i)) {
24		result[i] += val[i];
25		}
26		else {
27		result[i] = val[i];
28		}
29		}
30		}
31		return result;
32		}';
33
34	3	public function countJobsByStatus($objectName, $status, $workerName = null, $method = null)
35		{
36		/** @var DocumentManager $objectManager */
37	3	$objectManager = $this->getObjectManager();
38	3	$qb = $objectManager->createQueryBuilder($objectName);
39		$qb
40	3	->find()
41	3	->field('status')->equals($status);
42
43	3	$this->addWorkerNameCriterion($qb, $workerName, $method);
44	3	$method = 'count';
45	3	if (method_exists($qb, 'count')) {
46	3	$method = 'execute';
47	3	$qb->count();
48		}
49	3	$query = $qb->getQuery();
50
51	3	return $this->runQuery($query, $method, [], 0);
52		}
53
54		/**
55		* @param string\|null $workerName
56		* @param string\|null $method
57		*/
58	1	public function pruneExceptionJobs($workerName = null, $method = null)
59		{
60		/** @var DocumentManager $objectManager */
61	1	$objectManager = $this->getObjectManager();
62	1	$qb = $objectManager->createQueryBuilder($this->getJobArchiveClass());
63	1	$qb = $qb->remove();
64	1	$qb->field('status')->equals(BaseJob::STATUS_EXCEPTION);
65	1	$this->addWorkerNameCriterion($qb, $workerName, $method);
66
67	1	$query = $qb->getQuery();
68	1	$result = $this->runQuery($query, 'execute');
69	1	if ($result instanceof DeleteResult) {
70	1	return $result->getDeletedCount();
71		} elseif (isset($result['n'])) {
72		return $result['n'];
73		}
74
75		return 0;
76		}
77
78		/**
79		* @param $builder
80		* @param string\|null $workerName
81		* @param string\|null $method
82		*/
83	20	protected function addWorkerNameCriterion($builder, $workerName = null, $method = null)
84		{
85	20	if (null !== $workerName) {
86	6	$builder->field('workerName')->equals($workerName);
87		}
88
89	20	if (null !== $method) {
90	5	$builder->field('method')->equals($method);
91		}
92	20	}
93
94		/**
95		* @param null $workerName
		0 ignored issues – show Documentation Bug introduced 2017-12-07 08:03 UTC by Report Bug Copy Issue Report Are you sure the doc-type for parameter `$workerName` is correct as it would always require `null` to be passed? Loading history...
96		* @param null $method
		0 ignored issues – show Documentation Bug introduced 2018-05-24 16:25 UTC by Report Bug Copy Issue Report Are you sure the doc-type for parameter `$method` is correct as it would always require `null` to be passed? Loading history...
97		*
98		* @return int
99		*/
100	2	protected function updateExpired($workerName = null, $method = null)
101		{
102		/** @var DocumentManager $objectManager */
103	2	$objectManager = $this->getObjectManager();
104	2	$qb = $objectManager->createQueryBuilder($this->getJobClass());
105	2	$qb = $qb->updateMany();
106	2	$qb->field('expiresAt')->lte(Util::getMicrotimeDateTime());
107	2	$qb->field('status')->equals(BaseJob::STATUS_NEW);
108	2	$this->addWorkerNameCriterion($qb, $workerName, $method);
109	2	$qb->field('status')->set(\Dtc\QueueBundle\Model\Job::STATUS_EXPIRED);
110	2	$query = $qb->getQuery();
111	2	$result = $this->runQuery($query, 'execute');
112	2	if ($result instanceof UpdateResult) {
113	2	return $result->getModifiedCount();
114		} elseif (isset($result['n'])) {
115		return $result['n'];
116		}
117
118		return 0;
119		}
120
121		/**
122		* Removes archived jobs older than $olderThan.
123		*
124		* @return int
125		*/
126	1	public function pruneArchivedJobs(\DateTime $olderThan)
127		{
128	1	return $this->removeOlderThan($this->getJobArchiveClass(), 'updatedAt', $olderThan);
129		}
130
131	2	public function getWaitingJobCount($workerName = null, $method = null)
132		{
133		/** @var DocumentManager $objectManager */
134	2	$objectManager = $this->getObjectManager();
135	2	$builder = $objectManager->createQueryBuilder($this->getJobClass());
136		$builder
137	2	->find();
138
139	2	$this->addWorkerNameCriterion($builder, $workerName, $method);
140	2	$this->addStandardPredicates($builder);
141	2	$method = 'count';
142	2	if (method_exists($builder, 'count')) {
143	2	$builder->count();
144	2	$method = 'execute';
145		}
146	2	$query = $builder->getQuery();
147
148	2	return $this->runQuery($query, $method, [true], 0);
149		}
150
151		protected function getStatusByDocumentMapReduce($builder, $documentName)
152		{
153		$reduceFunc = self::REDUCE_FUNCTION;
154		$mapFunc = "function() {
155		var result = {};
156		result[this.status] = 1;
157		var key = this.worker_name + '->' + this.method + '()';
158		emit(key, result);
159		}";
160		$builder->map($mapFunc)
161		->reduce($reduceFunc);
162		$query = $builder->getQuery();
163		$results = $this->runQuery($query, 'execute', [], []);
164		$allStatus = static::getAllStatuses();
165
166		$status = [];
167
168		foreach ($results as $info) {
169		$status[$info['_id']] = $info['value'] + $allStatus;
170		}
171
172		return $status;
173		}
174
175		/**
176		* Get Status Jobs.
177		*
178		* @param string $documentName
179		*
180		* @return array
181		*/
182	3	protected function getStatusByDocument($documentName)
183		{
184		// Run a map reduce function get worker and status break down
185		/** @var DocumentManager $objectManager */
186	3	$objectManager = $this->getObjectManager();
187	3	$builder = $objectManager->createQueryBuilder($documentName);
188	3	if (method_exists($builder, 'map')) {
189		return $this->getStatusByDocumentMapReduce($builder, $documentName);
190		}
191
192	3	$aggregation = $objectManager->createAggregationBuilder($documentName);
193	3	$aggregation->group()
194	3	->field('id')->expression(
195	3	$aggregation->expr()
196	3	->field('worker_name')
197	3	->expression('$worker_name')
198	3	->field('method')
199	3	->expression('$method')
200	3	->field('status')
201	3	->expression('$status')
202		)
203	3	->field('value')
204	3	->sum(1);
205	3	$results = $this->runQuery($aggregation, 'execute', [], []);
206	3	$allStatus = static::getAllStatuses();
207
208	3	$status = [];
209
210	3	foreach ($results as $info) {
211	1	$key = $info['_id']['worker_name'].'->'.$info['_id']['method'].'()';
212	1	if (!isset($status[$key])) {
213	1	$status[$key] = $allStatus;
214		}
215	1	$status[$key][$info['_id']['status']] = $info['value'];
216		}
217
218	3	return $status;
219		}
220
221	3	public function getStatus(): array
222		{
223	3	$result = $this->getStatusByDocument($this->getJobClass());
224	3	$status2 = $this->getStatusByDocument($this->getJobArchiveClass());
225	3	foreach ($status2 as $key => $value) {
226	1	foreach ($value as $k => $v) {
227	1	if (isset($result[$key][$k])) {
228	1	$result[$key][$k] += $v;
229		} else {
230		$result[$key][$k] = $v;
231		}
232		}
233		}
234
235	3	$finalResult = [];
236	3	foreach ($result as $key => $item) {
237	1	ksort($item);
238	1	$finalResult[$key] = $item;
239		}
240
241	3	return $finalResult;
242		}
243
244		/**
245		* Get the next job to run (can be filtered by workername and method name).
246		*
247		* @param string $workerName
248		* @param string $methodName
249		* @param bool $prioritize
250		* @param string\|null $runId
251		*
252		* @return \Dtc\QueueBundle\Model\Job
253		*/
254	14	public function getJob($workerName = null, $methodName = null, $prioritize = true, $runId = null)
255		{
256	14	$builder = $this->getJobQueryBuilder($workerName, $methodName, $prioritize);
257		$builder
258	14	->findAndUpdate()
259	14	->returnNew();
260
261	14	$date = Util::getMicrotimeDateTime();
262		// Update
263		$builder
264	14	->field('startedAt')->set($date)
265	14	->field('status')->set(BaseJob::STATUS_RUNNING);
266	14	if (null !== $runId) {
267	1	$builder->field('runId')->set($runId);
268		}
269
270	14	$query = $builder->getQuery();
271
272	14	$job = $this->runQuery($query, 'execute');
273
274	14	return $job;
275		}
276
277	23	protected function runQuery($query, $method, array $arguments = [], $resultIfNamespaceError = null)
278		{
279		try {
280	23	$result = call_user_func_array([$query, $method], $arguments);
281		} catch (ResultException $resultException) {
282		if (false === strpos($resultException->getMessage(), 'namespace does not exist') && false === strpos($resultException->getMessage(), 'ns doesn\'t exist')) {
283		throw $resultException;
284		}
285		$result = $resultIfNamespaceError;
286		}
287
288	23	return $result;
289		}
290
291		/**
292		* @param string\|null $workerName
293		* @param string\|null $methodName
294		* @param bool $prioritize
295		*
296		* @return Builder
297		*/
298	14	public function getJobQueryBuilder($workerName = null, $methodName = null, $prioritize = true)
299		{
300		/** @var DocumentManager $objectManager */
301	14	$objectManager = $this->getObjectManager();
302	14	$builder = $objectManager->createQueryBuilder($this->getJobClass());
303
304	14	$this->addWorkerNameCriterion($builder, $workerName, $methodName);
305	14	if ($prioritize) {
306	14	$builder->sort([
307	14	'priority' => 'desc',
308		'whenAt' => 'asc',
309		]);
310		} else {
311	1	$builder->sort('whenAt', 'asc');
312		}
313
314		// Filter
315	14	$this->addStandardPredicates($builder);
316
317	14	return $builder;
318		}
319
320	1	protected function updateNearestBatch(\Dtc\QueueBundle\Model\Job $job)
321		{
322		/** @var DocumentManager $objectManager */
323	1	$objectManager = $this->getObjectManager();
324	1	$builder = $objectManager->createQueryBuilder($this->getJobClass());
325	1	$builder->find();
326
327	1	$builder->sort('whenAt', 'asc');
328	1	$builder->field('status')->equals(BaseJob::STATUS_NEW)
329	1	->field('crcHash')->equals($job->getCrcHash());
330	1	$oldJob = $this->runQuery($builder->getQuery(), 'getSingleResult');
331
332	1	if (!$oldJob) {
333		return null;
334		}
335
336		// Update priority or whenAt
337		// This makes sure if someone else is updating at the same time
338		// that we don't trounce their changes.
339	1	$builder = $objectManager->createQueryBuilder($this->getJobClass());
340	1	$builder->findAndUpdate();
341	1	$builder->field('_id')->equals($oldJob->getId());
342	1	$builder->field('priority')->lt($job->getPriority());
343	1	$builder->field('priority')->set($job->getPriority());
344	1	$this->runQuery($builder->getQuery(), 'execute');
345
346	1	$builder = $objectManager->createQueryBuilder($this->getJobClass());
347	1	$builder->findAndUpdate();
348	1	$builder->field('_id')->equals($oldJob->getId());
349	1	$builder->field('whenAt')->gt($job->getWhenAt());
350	1	$builder->field('whenAt')->set($job->getWhenAt());
351	1	$this->runQuery($builder->getQuery(), 'execute');
352
353	1	if ($job->getWhenAt() < $oldJob->getWhenAt()) {
354	1	$oldJob->setWhenAt($job->getWhenAt());
355		}
356	1	if ($job->getPriority() > $oldJob->getPriority()) {
357	1	$oldJob->setPriority($job->getPriority());
358		}
359
360	1	return $oldJob;
361		}
362
363		/**
364		* @param mixed $builder
365		*/
366	16	protected function addStandardPredicates($builder)
367		{
368	16	$date = Util::getMicrotimeDateTime();
369		$builder
370	16	->addAnd(
371	16	$builder->expr()->addOr($builder->expr()->field('whenAt')->equals(null), $builder->expr()->field('whenAt')->lte($date)),
372	16	$builder->expr()->addOr($builder->expr()->field('expiresAt')->equals(null), $builder->expr()->field('expiresAt')->gt($date))
373		)
374	16	->field('status')->equals(BaseJob::STATUS_NEW);
375	16	}
376
377	1	public function getWorkersAndMethods()
378		{
379		/** @var DocumentManager $documentManager */
380	1	$documentManager = $this->getObjectManager();
381
382	1	if (!method_exists($documentManager, 'createAggregationBuilder')) {
383		return [];
384		}
385
386	1	$aggregationBuilder = $documentManager->createAggregationBuilder($this->getJobClass());
387
388	1	$this->addStandardPredicates($aggregationBuilder->match());
389
390	1	$aggregationBuilder->group()
391	1	->field('id')
392	1	->expression(
393	1	$aggregationBuilder->expr()
394	1	->field('workerName')->expression('$workerName')
395	1	->field('method')->expression('$method')
396		);
397	1	$results = $aggregationBuilder->execute()->toArray();
398
399	1	if (!$results) {
400	1	return [];
401		}
402
403		$workersMethods = [];
404		foreach ($results as $result) {
405		if (isset($result['_id'])) {
406		$workersMethods[$result['_id']['worker_name']][] = $result['_id']['method'];
407		}
408		}
409
410		return $workersMethods;
411		}
412
413		/**
414		* @param string $workerName
415		* @param string $methodName
416		*/
417	2	public function countLiveJobs($workerName = null, $methodName = null)
418		{
419		/** @var DocumentManager $objectManager */
420	2	$objectManager = $this->getObjectManager();
421	2	$builder = $objectManager->createQueryBuilder($this->getJobClass());
422
423	2	$this->addWorkerNameCriterion($builder, $workerName, $methodName);
424		// Filter
425	2	$this->addStandardPredicates($builder);
426	2	$method = 'count';
427	2	if (method_exists($builder, 'count')) {
428	2	$builder->count();
429	2	$method = 'execute';
430		}
431
432	2	return $this->runQuery($builder->getQuery(), $method, [], 0);
433		}
434
435		/**
436		* @param string $workerName
437		* @param string $methodName
438		*/
439	1	public function archiveAllJobs($workerName = null, $methodName = null, callable $progressCallback = null)
440		{
441		/** @var DocumentManager $documentManager */
442	1	$documentManager = $this->getObjectManager();
443	1	$count = 0;
444	1	$builder = $this->getJobQueryBuilder($workerName, $methodName, true);
445		$builder
446	1	->findAndUpdate()
447	1	->returnNew();
448
449	1	$builder->field('status')->set(Job::STATUS_ARCHIVE);
450	1	$query = $builder->getQuery();
451		do {
452	1	$job = $this->runQuery($query, 'execute');
453	1	if ($job) {
454	1	$documentManager->remove($job);
455	1	++$count;
456
457	1	if (0 == $count % 10) {
458		$this->flush();
459		$this->updateProgress($progressCallback, $count);
460		}
461		}
462	1	} while ($job);
463	1	$this->flush();
464	1	$this->updateProgress($progressCallback, $count);
465	1	}
466		}
467

mmucklo / DtcQueueBundle

JobManager::addWorkerNameCriterion() A last analyzed 2025-06-22 22:56 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

JobManager::addWorkerNameCriterion() A
last analyzed 2025-06-22 22:56 UTC