JobManager::pruneExceptionJobs() - Code Metrics - Inspection of "4.0 release" - mmucklo/DtcQueueBundle - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#30)

by Matthew

created 2018-01-07 02:22 UTC

JobManager::pruneExceptionJobs() A

↳ Parent: JobManager

Complexity

Conditions	2
Paths	2

Size

Total Lines	17
Code Lines	11

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	6

Importance

Changes

Metric	Value
dl	0
loc	17
ccs	0
cts	11
cp	0
rs	9.4285
c	0
b	0
f	0
cc	2
eloc	11
nc	2
nop	2
crap	6

<?php

namespace Dtc\QueueBundle\ODM;

use Doctrine\MongoDB\Query\Builder;
use Dtc\QueueBundle\Doctrine\DoctrineJobManager;
use Doctrine\ODM\MongoDB\DocumentManager;
use Dtc\QueueBundle\Document\Job;
use Dtc\QueueBundle\Model\BaseJob;
use Dtc\QueueBundle\Model\RetryableJob;
use Dtc\QueueBundle\Model\StallableJob;

class JobManager extends DoctrineJobManager
{
    use CommonTrait;
    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);
        $query = $qb->getQuery();

        return $query->count();
    }

    /**
     * @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 = $query->execute();
        if (isset($result['n'])) {
            return $result['n'];
        }

        return 0;
    }

    /**
     * @param Builder     $builder
     * @param string|null $workerName
     * @param string|null $method
     */
    protected function addWorkerNameCriterion(Builder $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(new \DateTime());
        $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 = $query->execute();
        if (isset($result['n'])) {
            return $result['n'];
        }

        return 0;
    }

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

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

        $this->addWorkerNameCriterion($qb, $workerName, $method);

        // Filter
        $date = new \DateTime();
        $qb
            ->addAnd(
                $qb->expr()->addOr($qb->expr()->field('expiresAt')->equals(null), $qb->expr()->field('expiresAt')->gt($date))
            );
        $query = $qb->getQuery();

        return $query->count(true);
    }

    /**
     * Get Status Jobs.
     *
     * @param string $documentName
     *
     * @return array
     */
    protected function getStatusByDocument($documentName)
    {
        // Run a map reduce function get worker and status break down
        $mapFunc = "function() {
            var result = {};
            result[this.status] = 1;
            var key = this.worker_name + '->' + this.method + '()';
            emit(key, result);
        }";
        $reduceFunc = self::REDUCE_FUNCTION;
        /** @var DocumentManager $objectManager */
        $objectManager = $this->getObjectManager();
        $qb = $objectManager->createQueryBuilder($documentName);
        $qb->map($mapFunc)
            ->reduce($reduceFunc);
        $query = $qb->getQuery();
        $results = $query->execute();

        $allStatus = array(
            BaseJob::STATUS_NEW => 0,
            BaseJob::STATUS_RUNNING => 0,
            BaseJob::STATUS_SUCCESS => 0,
            BaseJob::STATUS_FAILURE => 0,
            BaseJob::STATUS_EXCEPTION => 0,
            StallableJob::STATUS_STALLED => 0,
            \Dtc\QueueBundle\Model\Job::STATUS_EXPIRED => 0,
            RetryableJob::STATUS_MAX_FAILURES => 0,
            RetryableJob::STATUS_MAX_EXCEPTIONS => 0,
            RetryableJob::STATUS_MAX_RETRIES => 0,
            StallableJob::STATUS_MAX_STALLS => 0,
        );

        $status = [];

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

        return $status;
    }

    public function getStatus()
    {
        $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 = new \DateTime();
        // Update
        $builder
            ->field('startedAt')->set($date)
            ->field('status')->set(BaseJob::STATUS_RUNNING)
            ->field('runId')->set($runId);

        $query = $builder->getQuery();

        $job = $query->execute();

        return $job;
    }

    /**
     * @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 = $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());
        $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());
        $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 = new \DateTime();
        $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);

        return $builder->getQuery()->count();
    }

    /**
     * @param string   $workerName
     * @param string   $methodName
     * @param \Closure $progressCallback
     */
    public function archiveAllJobs($workerName = null, $methodName = null, $progressCallback)
    {
        /** @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 = $query->execute();
            if ($job) {
                $documentManager->remove($job);
                ++$count;

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


1		<?php
2
3		namespace Dtc\QueueBundle\ODM;
4
5		use Doctrine\MongoDB\Query\Builder;
6		use Dtc\QueueBundle\Doctrine\DoctrineJobManager;
7		use Doctrine\ODM\MongoDB\DocumentManager;
8		use Dtc\QueueBundle\Document\Job;
9		use Dtc\QueueBundle\Model\BaseJob;
10		use Dtc\QueueBundle\Model\RetryableJob;
11		use Dtc\QueueBundle\Model\StallableJob;
12
13		class JobManager extends DoctrineJobManager
14		{
15		use CommonTrait;
16		const REDUCE_FUNCTION = 'function(k, vals) {
17		var result = {};
18		for (var index in vals) {
19		var val = vals[index];
20		for (var i in val) {
21		if (result.hasOwnProperty(i)) {
22		result[i] += val[i];
23		}
24		else {
25		result[i] = val[i];
26		}
27		}
28		}
29		return result;
30		}';
31
32		public function countJobsByStatus($objectName, $status, $workerName = null, $method = null)
33		{
34		/** @var DocumentManager $objectManager */
35		$objectManager = $this->getObjectManager();
36		$qb = $objectManager->createQueryBuilder($objectName);
37		$qb
38		->find()
39		->field('status')->equals($status);
40
41		$this->addWorkerNameCriterion($qb, $workerName, $method);
42		$query = $qb->getQuery();
43
44		return $query->count();
45		}
46
47		/**
48		* @param string\|null $workerName
49		* @param string\|null $method
50		*/
51		public function pruneExceptionJobs($workerName = null, $method = null)
52		{
53		/** @var DocumentManager $objectManager */
54		$objectManager = $this->getObjectManager();
55		$qb = $objectManager->createQueryBuilder($this->getJobArchiveClass());
56		$qb = $qb->remove();
57		$qb->field('status')->equals(BaseJob::STATUS_EXCEPTION);
58		$this->addWorkerNameCriterion($qb, $workerName, $method);
59
60		$query = $qb->getQuery();
61		$result = $query->execute();
62		if (isset($result['n'])) {
63		return $result['n'];
64		}
65
66		return 0;
67		}
68
69		/**
70		* @param Builder $builder
71		* @param string\|null $workerName
72		* @param string\|null $method
73		*/
74	2	protected function addWorkerNameCriterion(Builder $builder, $workerName = null, $method = null)
75		{
76	2	if (null !== $workerName) {
77		$builder->field('workerName')->equals($workerName);
78		}
79
80	2	if (null !== $method) {
81		$builder->field('method')->equals($method);
82		}
83	2	}
84
85		/**
86		* @param null $workerName
87		* @param null $method
88		*
89		* @return int
90		*/
91	1	protected function updateExpired($workerName = null, $method = null)
92		{
93		/** @var DocumentManager $objectManager */
94	1	$objectManager = $this->getObjectManager();
95	1	$qb = $objectManager->createQueryBuilder($this->getJobClass());
96	1	$qb = $qb->updateMany();
97	1	$qb->field('expiresAt')->lte(new \DateTime());
98	1	$qb->field('status')->equals(BaseJob::STATUS_NEW);
99	1	$this->addWorkerNameCriterion($qb, $workerName, $method);
100	1	$qb->field('status')->set(\Dtc\QueueBundle\Model\Job::STATUS_EXPIRED);
101	1	$query = $qb->getQuery();
102	1	$result = $query->execute();
103	1	if (isset($result['n'])) {
104	1	return $result['n'];
105		}
106
107		return 0;
108		}
109
110		/**
111		* Removes archived jobs older than $olderThan.
112		*
113		* @param \DateTime $olderThan
114		*
115		* @return int
116		*/
117		public function pruneArchivedJobs(\DateTime $olderThan)
118		{
119		return $this->removeOlderThan($this->getJobArchiveClass(), 'updatedAt', $olderThan);
120		}
121
122	1	public function getJobCount($workerName = null, $method = null)
123		{
124		/** @var DocumentManager $objectManager */
125	1	$objectManager = $this->getObjectManager();
126	1	$qb = $objectManager->createQueryBuilder($this->getJobClass());
127		$qb
128	1	->find();
129
130	1	$this->addWorkerNameCriterion($qb, $workerName, $method);
131
132		// Filter
133	1	$date = new \DateTime();
134		$qb
135	1	->addAnd(
136	1	$qb->expr()->addOr($qb->expr()->field('expiresAt')->equals(null), $qb->expr()->field('expiresAt')->gt($date))
137	1	);
138	1	$query = $qb->getQuery();
139
140	1	return $query->count(true);
141		}
142
143		/**
144		* Get Status Jobs.
145		*
146		* @param string $documentName
147		*
148		* @return array
149		*/
150	1	protected function getStatusByDocument($documentName)
151		{
152		// Run a map reduce function get worker and status break down
153		$mapFunc = "function() {
154		var result = {};
155		result[this.status] = 1;
156		var key = this.worker_name + '->' + this.method + '()';
157		emit(key, result);
158	1	}";
159	1	$reduceFunc = self::REDUCE_FUNCTION;
160		/** @var DocumentManager $objectManager */
161	1	$objectManager = $this->getObjectManager();
162	1	$qb = $objectManager->createQueryBuilder($documentName);
163	1	$qb->map($mapFunc)
164	1	->reduce($reduceFunc);
165	1	$query = $qb->getQuery();
166	1	$results = $query->execute();
167
168		$allStatus = array(
169	1	BaseJob::STATUS_NEW => 0,
170	1	BaseJob::STATUS_RUNNING => 0,
171	1	BaseJob::STATUS_SUCCESS => 0,
172	1	BaseJob::STATUS_FAILURE => 0,
173	1	BaseJob::STATUS_EXCEPTION => 0,
174	1	StallableJob::STATUS_STALLED => 0,
175	1	\Dtc\QueueBundle\Model\Job::STATUS_EXPIRED => 0,
176	1	RetryableJob::STATUS_MAX_FAILURES => 0,
177	1	RetryableJob::STATUS_MAX_EXCEPTIONS => 0,
178	1	RetryableJob::STATUS_MAX_RETRIES => 0,
179	1	StallableJob::STATUS_MAX_STALLS => 0,
180	1	);
181
182	1	$status = [];
183
184	1	foreach ($results as $info) {
185		$status[$info['_id']] = $info['value'] + $allStatus;
186	1	}
187
188	1	return $status;
189		}
190
191	1	public function getStatus()
192		{
193	1	$result = $this->getStatusByDocument($this->getJobClass());
194	1	$status2 = $this->getStatusByDocument($this->getJobArchiveClass());
195	1	foreach ($status2 as $key => $value) {
196		foreach ($value as $k => $v) {
197		if (isset($result[$key][$k])) {
198		$result[$key][$k] += $v;
199		} else {
200		$result[$key][$k] = $v;
201		}
202		}
203	1	}
204
205	1	$finalResult = [];
206	1	foreach ($result as $key => $item) {
207		ksort($item);
208		$finalResult[$key] = $item;
209	1	}
210
211	1	return $finalResult;
212		}
213
214		/**
215		* Get the next job to run (can be filtered by workername and method name).
216		*
217		* @param string $workerName
218		* @param string $methodName
219		* @param bool $prioritize
220		* @param string\|null $runId
221		*
222		* @return \Dtc\QueueBundle\Model\Job
223		*/
224		public function getJob($workerName = null, $methodName = null, $prioritize = true, $runId = null)
225		{
226		$builder = $this->getJobQueryBuilder($workerName, $methodName, $prioritize);
227		$builder
228		->findAndUpdate()
229		->returnNew();
230
231		$date = new \DateTime();
232		// Update
233		$builder
234		->field('startedAt')->set($date)
235		->field('status')->set(BaseJob::STATUS_RUNNING)
236		->field('runId')->set($runId);
237
238		$query = $builder->getQuery();
239
240		$job = $query->execute();
241
242		return $job;
243		}
244
245		/**
246		* @param string\|null $workerName
247		* @param string\|null $methodName
248		* @param bool $prioritize
249		*
250		* @return Builder
251		*/
252		public function getJobQueryBuilder($workerName = null, $methodName = null, $prioritize = true)
253		{
254		/** @var DocumentManager $objectManager */
255		$objectManager = $this->getObjectManager();
256		$builder = $objectManager->createQueryBuilder($this->getJobClass());
257
258		$this->addWorkerNameCriterion($builder, $workerName, $methodName);
259		if ($prioritize) {
260		$builder->sort([
261		'priority' => 'desc',
262		'whenAt' => 'asc',
263		]);
264		} else {
265		$builder->sort('whenAt', 'asc');
266		}
267
268		// Filter
269		$this->addStandardPredicates($builder);
270
271		return $builder;
272		}
273
274		protected function updateNearestBatch(\Dtc\QueueBundle\Model\Job $job)
275		{
276		/** @var DocumentManager $objectManager */
277		$objectManager = $this->getObjectManager();
278		$builder = $objectManager->createQueryBuilder($this->getJobClass());
279		$builder->find();
280
281		$builder->sort('whenAt', 'asc');
282		$builder->field('status')->equals(BaseJob::STATUS_NEW)
283		->field('crcHash')->equals($job->getCrcHash());
284		$oldJob = $builder->getQuery()->getSingleResult();
285
286		if (!$oldJob) {
287		return null;
288		}
289
290		// Update priority or whenAt
291		// This makes sure if someone else is updating at the same time
292		// that we don't trounce their changes.
293		$builder = $objectManager->createQueryBuilder($this->getJobClass());
294		$builder->findAndUpdate();
295		$builder->field('_id')->equals($oldJob->getId());
296		$builder->field('priority')->lt($job->getPriority());
297		$builder->field('priority')->set($job->getPriority());
298		$builder->getQuery()->execute();
299
300		$builder = $objectManager->createQueryBuilder($this->getJobClass());
301		$builder->findAndUpdate();
302		$builder->field('_id')->equals($oldJob->getId());
303		$builder->field('whenAt')->gt($job->getWhenAt());
304		$builder->field('whenAt')->set($job->getWhenAt());
305		$builder->getQuery()->execute();
306
307		if ($job->getWhenAt() < $oldJob->getWhenAt()) {
308		$oldJob->setWhenAt($job->getWhenAt());
309		}
310		if ($job->getPriority() > $oldJob->getPriority()) {
311		$oldJob->setPriority($job->getPriority());
312		}
313
314		return $oldJob;
315		}
316
317		/**
318		* @param mixed $builder
319		*/
320		protected function addStandardPredicates($builder)
321		{
322		$date = new \DateTime();
323		$builder
324		->addAnd(
325		$builder->expr()->addOr($builder->expr()->field('whenAt')->equals(null), $builder->expr()->field('whenAt')->lte($date)),
326		$builder->expr()->addOr($builder->expr()->field('expiresAt')->equals(null), $builder->expr()->field('expiresAt')->gt($date))
327		)
328		->field('status')->equals(BaseJob::STATUS_NEW);
329		}
330
331		public function getWorkersAndMethods()
332		{
333		/** @var DocumentManager $documentManager */
334		$documentManager = $this->getObjectManager();
335
336		if (!method_exists($documentManager, 'createAggregationBuilder')) {
337		return [];
338		}
339
340		$aggregationBuilder = $documentManager->createAggregationBuilder($this->getJobClass());
341
342		$this->addStandardPredicates($aggregationBuilder->match());
343
344		$aggregationBuilder->group()
345		->field('id')
346		->expression(
347		$aggregationBuilder->expr()
348		->field('workerName')->expression('$workerName')
349		->field('method')->expression('$method')
350		);
351		$results = $aggregationBuilder->execute()->toArray();
352
353		if (!$results) {
354		return [];
355		}
356
357		$workersMethods = [];
358		foreach ($results as $result) {
359		if (isset($result['_id'])) {
360		$workersMethods[$result['_id']['worker_name']][] = $result['_id']['method'];
361		}
362		}
363
364		return $workersMethods;
365		}
366
367		/**
368		* @param string $workerName
369		* @param string $methodName
370		*/
371		public function countLiveJobs($workerName = null, $methodName = null)
372		{
373		/** @var DocumentManager $objectManager */
374		$objectManager = $this->getObjectManager();
375		$builder = $objectManager->createQueryBuilder($this->getJobClass());
376
377		$this->addWorkerNameCriterion($builder, $workerName, $methodName);
378		// Filter
379		$this->addStandardPredicates($builder);
380
381		return $builder->getQuery()->count();
382		}
383
384		/**
385		* @param string $workerName
386		* @param string $methodName
387		* @param \Closure $progressCallback
388		*/
389	1	public function archiveAllJobs($workerName = null, $methodName = null, $progressCallback)
390		{
391		/** @var DocumentManager $documentManager */
392	1	$documentManager = $this->getObjectManager();
393		$count = 0;
394		$builder = $this->getJobQueryBuilder($workerName, $methodName, true);
395		$builder
396		->findAndUpdate()
397		->returnNew();
398
399		$builder->field('status')->set(Job::STATUS_ARCHIVE);
400		$query = $builder->getQuery();
401		do {
402		$job = $query->execute();
403		if ($job) {
404		$documentManager->remove($job);
405		++$count;
406
407		if (0 == $count % 10) {
408		$this->flush();
409		$progressCallback($count);
410		}
411		}
412		} while ($job);
413		$this->flush();
414		$progressCallback($count);
415		}
416		}
417

mmucklo / DtcQueueBundle

Pull Request — master (#30)

JobManager::pruneExceptionJobs() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like