DoctrineJobManager::resetJobsByCriterion() - 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-15 00:47 UTC

DoctrineJobManager::resetJobsByCriterion() B

↳ Parent: DoctrineJobManager

Complexity

Conditions	3
Paths	4

Size

Total Lines	28
Code Lines	22

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	19
CRAP Score	3

Importance

Changes

Metric	Value
dl	0
loc	28
ccs	19
cts	19
cp	1
rs	8.8571
c	0
b	0
f	0
cc	3
eloc	22
nc	4
nop	3
crap	3

<?php

namespace Dtc\QueueBundle\Doctrine;

use Doctrine\Common\Persistence\ObjectManager;
use Doctrine\Common\Persistence\ObjectRepository;
use Doctrine\ODM\MongoDB\DocumentRepository;
use Doctrine\ORM\EntityRepository;
use Dtc\QueueBundle\Manager\ArchivableJobManager;
use Dtc\QueueBundle\Model\BaseJob;
use Dtc\QueueBundle\Model\RetryableJob;
use Dtc\QueueBundle\Model\Job;
use Dtc\QueueBundle\Model\JobTiming;
use Dtc\QueueBundle\Manager\JobTimingManager;
use Dtc\QueueBundle\Model\StallableJob;
use Dtc\QueueBundle\Model\Run;
use Dtc\QueueBundle\Manager\RunManager;
use Dtc\QueueBundle\Util\Util;

abstract class DoctrineJobManager extends ArchivableJobManager
{
    /** Number of jobs to prune / reset / gather at a time */
    const FETCH_COUNT = 100;

    /** Number of seconds before a job is considered stalled if the runner is no longer active */
    const STALLED_SECONDS = 1800;

    /**
     * @var ObjectManager
     */
    protected $objectManager;

    /**
     * DoctrineJobManager constructor.
     *
     * @param RunManager       $runManager
     * @param JobTimingManager $jobTimingManager
     * @param ObjectManager    $objectManager
     * @param $jobClass
     * @param $jobArchiveClass
     */
    public function __construct(
        RunManager $runManager,
        JobTimingManager $jobTimingManager,
        ObjectManager $objectManager,
        $jobClass,
        $jobArchiveClass
    ) {
        $this->objectManager = $objectManager;
        parent::__construct($runManager, $jobTimingManager, $jobClass, $jobArchiveClass);
    }

    /**
     * @return ObjectManager
     */
    public function getObjectManager()
    {
        return $this->objectManager;
    }

    /**
     * @return ObjectRepository
     */
    public function getRepository()
    {
        return $this->getObjectManager()->getRepository($this->getJobClass());
    }

    /**
     * @param string $objectName
     */
    abstract protected function countJobsByStatus($objectName, $status, $workerName = null, $method = null);

    public function resetExceptionJobs($workerName = null, $method = null)
    {
        $count = $this->countJobsByStatus($this->getJobArchiveClass(), Job::STATUS_EXCEPTION, $workerName, $method);

        $criterion = ['status' => Job::STATUS_EXCEPTION];
        $this->addWorkerNameMethod($criterion, $workerName, $method);

        $countProcessed = 0;
        for ($i = 0; $i < $count; $i += static::FETCH_COUNT) {
            $countProcessed += $this->resetJobsByCriterion(
                $criterion,
                static::FETCH_COUNT,
                $i
            );
        }

        return $countProcessed;
    }

    /**
     * Sets the status to Job::STATUS_EXPIRED for those jobs that are expired.
     *
     * @param null $workerName
     * @param null $method
     *
     * @return mixed
     */
    abstract protected function updateExpired($workerName = null, $method = null);

    protected function addWorkerNameMethod(array &$criterion, $workerName = null, $method = null)
    {
        if (null !== $workerName) {
            $criterion['workerName'] = $workerName;
        }
        if (null !== $method) {
            $criterion['method'] = $method;
        }
    }

    public function pruneExpiredJobs($workerName = null, $method = null)
    {
        $count = $this->updateExpired($workerName, $method);
        $criterion = ['status' => Job::STATUS_EXPIRED];
        $this->addWorkerNameMethod($criterion, $workerName, $method);
        $objectManager = $this->getObjectManager();
        $repository = $this->getRepository();
        $finalCount = 0;
        for ($i = 0; $i < $count; $i += static::FETCH_COUNT) {
            $expiredJobs = $repository->findBy($criterion, null, static::FETCH_COUNT, $i);
            $innerCount = 0;
            if (!empty($expiredJobs)) {
                foreach ($expiredJobs as $expiredJob) {
                    /* @var Job $expiredJob */
                    $expiredJob->setStatus(Job::STATUS_EXPIRED);
                    $objectManager->remove($expiredJob);
                    ++$finalCount;
                    ++$innerCount;
                }
            }
            $this->flush();
            for ($j = 0; $j < $innerCount; ++$j) {
                $this->jobTiminigManager->recordTiming(JobTiming::STATUS_FINISHED_EXPIRED);
            }
        }

        return $finalCount;
    }

    protected function flush()
    {
        $this->getObjectManager()->flush();
    }

    protected function getStalledJobs($workerName = null, $method = null)
    {
        $count = $this->countJobsByStatus($this->getJobClass(), Job::STATUS_RUNNING, $workerName, $method);

        $criterion = ['status' => BaseJob::STATUS_RUNNING];
        $this->addWorkerNameMethod($criterion, $workerName, $method);

        $runningJobs = $this->findRunningJobs($criterion, $count);

        return $this->extractStalledJobs($runningJobs);
    }

    protected function findRunningJobs($criterion, $count)
    {
        $repository = $this->getRepository();
        $runningJobsById = [];

        for ($i = 0; $i < $count; $i += static::FETCH_COUNT) {
            $runningJobs = $repository->findBy($criterion, null, static::FETCH_COUNT, $i);
            if (!empty($runningJobs)) {
                foreach ($runningJobs as $job) {
                    /** @var StallableJob $job */
                    if (null !== $runId = $job->getRunId()) {
                        $runningJobsById[$runId][] = $job;
                    }
                }
            }
        }

        return $runningJobsById;
    }

    /**
     * @param $runId
     * @param array $jobs
     * @param array $stalledJobs
     */
    protected function extractStalledLiveRuns($runId, array $jobs, array &$stalledJobs)
    {
        $objectManager = $this->getObjectManager();
        $runRepository = $objectManager->getRepository($this->getRunManager()->getRunClass());
        if ($run = $runRepository->find($runId)) {
            foreach ($jobs as $job) {
                if ($run->getCurrentJobId() == $job->getId()) {
                    continue;
                }
                $stalledJobs[] = $job;
            }
        }
    }

    /**
     * @param array $runningJobsById
     *
     * @return array
     */
    protected function extractStalledJobs(array $runningJobsById)
    {
        $stalledJobs = [];
        foreach (array_keys($runningJobsById) as $runId) {
            $this->extractStalledLiveRuns($runId, $runningJobsById[$runId], $stalledJobs);
            $this->extractStalledJobsRunArchive($runningJobsById, $stalledJobs, $runId);
        }

        return $stalledJobs;
    }

    protected function extractStalledJobsRunArchive(array $runningJobsById, array &$stalledJobs, $runId)
    {
        $runManager = $this->getRunManager();
        if (!method_exists($runManager, 'getObjectManager')) {
            return;
        }
        if (!method_exists($runManager, 'getRunArchiveClass')) {
            return;
        }

        /** @var EntityRepository|DocumentRepository $runArchiveRepository */
        $runArchiveRepository = $runManager->getObjectManager()->getRepository($runManager->getRunArchiveClass());
        /** @var Run $run */
        if ($run = $runArchiveRepository->find($runId)) {
            if ($endTime = $run->getEndedAt()) {
                // Did it end over an hour ago
                if ((time() - $endTime->getTimestamp()) > static::STALLED_SECONDS) {
                    $stalledJobs = array_merge($stalledJobs, $runningJobsById[$runId]);
                }
            }
        }
    }

    protected function runStalledLoop($i, $count, array $stalledJobs)
    {
        $resetCount = 0;
        for ($j = $i, $max = $i + static::FETCH_COUNT; $j < $max && $j < $count; ++$j) {
            /* StallableJob $job */
            $job = $stalledJobs[$j];
            $job->setStatus(StallableJob::STATUS_STALLED);
            if ($this->saveHistory($job)) {
                ++$resetCount;
            }
        }

        return $resetCount;
    }

    public function resetStalledJobs($workerName = null, $method = null)
    {
        $stalledJobs = $this->getStalledJobs($workerName, $method);

        $countReset = 0;
        for ($i = 0, $count = count($stalledJobs); $i < $count; $i += static::FETCH_COUNT) {
            $resetCount = $this->runStalledLoop($i, $count, $stalledJobs);
            for ($j = $i, $max = $i + static::FETCH_COUNT; $j < $max && $j < $count; ++$j) {
                $this->jobTiminigManager->recordTiming(JobTiming::STATUS_FINISHED_STALLED);
            }
            $countReset += $resetCount;
            $this->flush();
        }

        return $countReset;
    }

    /**
     * @param string $workerName
     * @param string $method
     */
    public function pruneStalledJobs($workerName = null, $method = null)
    {
        $stalledJobs = $this->getStalledJobs($workerName, $method);

        $countProcessed = 0;
        for ($i = 0, $count = count($stalledJobs); $i < $count; $i += static::FETCH_COUNT) {
            for ($j = $i, $max = $i + static::FETCH_COUNT; $j < $max && $j < $count; ++$j) {
                /** @var StallableJob $job */
                $job = $stalledJobs[$j];
                $job->setStatus(StallableJob::STATUS_STALLED);
                $job->setStalls(intval($job->getStalls()) + 1);
                $this->deleteJob($job);
                ++$countProcessed;
            }
            $this->flush();
        }

        return $countProcessed;
    }

    public function deleteJob(\Dtc\QueueBundle\Model\Job $job)
    {
        $objectManager = $this->getObjectManager();
        $objectManager->remove($job);
        $objectManager->flush();
    }

    protected function stallableSaveHistory(StallableJob $job, $retry)
    {
        if (!$retry) {
            $this->deleteJob($job);
        }

        return $retry;
    }

    protected function stallableSave(StallableJob $job)
    {
        // Generate crc hash for the job
        $hashValues = array(get_class($job), $job->getMethod(), $job->getWorkerName(), $job->getArgs());
        $crcHash = hash('sha256', serialize($hashValues));
        $job->setCrcHash($crcHash);
        $objectManager = $this->getObjectManager();

        if (true === $job->getBatch()) {
            $oldJob = $this->updateNearestBatch($job);
            if ($oldJob) {
                return $oldJob;
            }
        }

        // Just save a new job
        $this->resetSaveOk(__FUNCTION__);
        $objectManager->persist($job);
        $objectManager->flush();

        return $job;
    }

    abstract protected function updateNearestBatch(Job $job);

    /**
     * @param string $objectName
     */
    abstract protected function stopIdGenerator($objectName);

    abstract protected function restoreIdGenerator($objectName);

    /**
     * @param array $criterion
     * @param int   $limit
     * @param int   $offset
     */
    private function resetJobsByCriterion(
        array $criterion,
        $limit,
        $offset
    ) {
        $objectManager = $this->getObjectManager();
        $this->resetSaveOk(__FUNCTION__);
        $objectName = $this->getJobClass();
        $archiveObjectName = $this->getJobArchiveClass();
        $jobRepository = $objectManager->getRepository($objectName);
        $jobArchiveRepository = $objectManager->getRepository($archiveObjectName);
        $className = $jobRepository->getClassName();
        $metadata = $objectManager->getClassMetadata($className);
        $this->stopIdGenerator($objectName);
        $identifierData = $metadata->getIdentifier();
        $idColumn = isset($identifierData[0]) ? $identifierData[0] : 'id';
        $results = $jobArchiveRepository->findBy($criterion, [$idColumn => 'ASC'], $limit, $offset);
        $countProcessed = 0;

        foreach ($results as $jobArchive) {
            $countProcessed += $this->resetArchiveJob($jobArchive);
        }
        $objectManager->flush();

        $this->restoreIdGenerator($objectName);

        return $countProcessed;
    }

    protected function resetSaveOk($function)
    {
    }

    /**
     * @param null $workerName
     * @param null $methodName
     * @param bool $prioritize
     */
    abstract public function getJobQueryBuilder($workerName = null, $methodName = null, $prioritize = true);

    /**
     * @param StallableJob $jobArchive
     * @param $className
     *
     * @return int Number of jobs reset
     */
    protected function resetArchiveJob(StallableJob $jobArchive)
    {
        $objectManager = $this->getObjectManager();
        if ($this->updateMaxStatus($jobArchive, StallableJob::STATUS_MAX_RETRIES, $jobArchive->getMaxRetries(), $jobArchive->getRetries())) {

            $objectManager->persist($jobArchive);

            return 0;
        }

        /** @var StallableJob $job */
        $className = $this->getJobClass();
        $newJob = new $className();
        Util::copy($jobArchive, $newJob);
        $this->resetJob($newJob);
        $objectManager->remove($jobArchive);
        $this->flush();

        return 1;
    }

    protected function resetJob(RetryableJob $job)
    {
        if (!$job instanceof StallableJob) {
            throw new \InvalidArgumentException('$job should be instance of '.StallableJob::class);
        }
        $job->setStatus(BaseJob::STATUS_NEW);
        $job->setMessage(null);
        $job->setFinishedAt(null);
        $job->setStartedAt(null);
        $job->setElapsed(null);
        $job->setRetries($job->getRetries() + 1);
        $this->getObjectManager()->persist($job);
        $this->flush();

        return true;
    }

    abstract public function getWorkersAndMethods();

    abstract public function countLiveJobs($workerName = null, $methodName = null);

    abstract public function archiveAllJobs($workerName = null, $methodName = null, $progressCallback);
}


1		<?php
2
3		namespace Dtc\QueueBundle\Doctrine;
4
5		use Doctrine\Common\Persistence\ObjectManager;
6		use Doctrine\Common\Persistence\ObjectRepository;
7		use Doctrine\ODM\MongoDB\DocumentRepository;
8		use Doctrine\ORM\EntityRepository;
9		use Dtc\QueueBundle\Manager\ArchivableJobManager;
10		use Dtc\QueueBundle\Model\BaseJob;
11		use Dtc\QueueBundle\Model\RetryableJob;
12		use Dtc\QueueBundle\Model\Job;
13		use Dtc\QueueBundle\Model\JobTiming;
14		use Dtc\QueueBundle\Manager\JobTimingManager;
15		use Dtc\QueueBundle\Model\StallableJob;
16		use Dtc\QueueBundle\Model\Run;
17		use Dtc\QueueBundle\Manager\RunManager;
18		use Dtc\QueueBundle\Util\Util;
19
20		abstract class DoctrineJobManager extends ArchivableJobManager
21		{
22		/** Number of jobs to prune / reset / gather at a time */
23		const FETCH_COUNT = 100;
24
25		/** Number of seconds before a job is considered stalled if the runner is no longer active */
26		const STALLED_SECONDS = 1800;
27
28		/**
29		* @var ObjectManager
30		*/
31		protected $objectManager;
32
33		/**
34		* DoctrineJobManager constructor.
35		*
36		* @param RunManager $runManager
37		* @param JobTimingManager $jobTimingManager
38		* @param ObjectManager $objectManager
39		* @param $jobClass
40		* @param $jobArchiveClass
41		*/
42	20	public function __construct(
43		RunManager $runManager,
44		JobTimingManager $jobTimingManager,
45		ObjectManager $objectManager,
46		$jobClass,
47		$jobArchiveClass
48		) {
49	20	$this->objectManager = $objectManager;
50	20	parent::__construct($runManager, $jobTimingManager, $jobClass, $jobArchiveClass);
51	20	}
52
53		/**
54		* @return ObjectManager
55		*/
56	57	public function getObjectManager()
57		{
58	57	return $this->objectManager;
59		}
60
61		/**
62		* @return ObjectRepository
63		*/
64	27	public function getRepository()
65		{
66	27	return $this->getObjectManager()->getRepository($this->getJobClass());
67		}
68
69		/**
70		* @param string $objectName
71		*/
72		abstract protected function countJobsByStatus($objectName, $status, $workerName = null, $method = null);
73
74	2	public function resetExceptionJobs($workerName = null, $method = null)
75		{
76	2	$count = $this->countJobsByStatus($this->getJobArchiveClass(), Job::STATUS_EXCEPTION, $workerName, $method);
77
78	2	$criterion = ['status' => Job::STATUS_EXCEPTION];
79	2	$this->addWorkerNameMethod($criterion, $workerName, $method);
80
81	2	$countProcessed = 0;
82	2	for ($i = 0; $i < $count; $i += static::FETCH_COUNT) {
83	2	$countProcessed += $this->resetJobsByCriterion(
84	2	$criterion,
85	2	static::FETCH_COUNT,
86	2	$i
87		);
88		}
89
90	2	return $countProcessed;
91		}
92
93		/**
94		* Sets the status to Job::STATUS_EXPIRED for those jobs that are expired.
95		*
96		* @param null $workerName
97		* @param null $method
98		*
99		* @return mixed
100		*/
101		abstract protected function updateExpired($workerName = null, $method = null);
102
103	9	protected function addWorkerNameMethod(array &$criterion, $workerName = null, $method = null)
104		{
105	9	if (null !== $workerName) {
106	4	$criterion['workerName'] = $workerName;
107		}
108	9	if (null !== $method) {
109	4	$criterion['method'] = $method;
110		}
111	9	}
112
113	3	public function pruneExpiredJobs($workerName = null, $method = null)
114		{
115	3	$count = $this->updateExpired($workerName, $method);
116	3	$criterion = ['status' => Job::STATUS_EXPIRED];
117	3	$this->addWorkerNameMethod($criterion, $workerName, $method);
118	3	$objectManager = $this->getObjectManager();
119	3	$repository = $this->getRepository();
120	3	$finalCount = 0;
121	3	for ($i = 0; $i < $count; $i += static::FETCH_COUNT) {
122	3	$expiredJobs = $repository->findBy($criterion, null, static::FETCH_COUNT, $i);
123	3	$innerCount = 0;
124	3	if (!empty($expiredJobs)) {
125	3	foreach ($expiredJobs as $expiredJob) {
126		/* @var Job $expiredJob */
127	3	$expiredJob->setStatus(Job::STATUS_EXPIRED);
128	3	$objectManager->remove($expiredJob);
129	3	++$finalCount;
130	3	++$innerCount;
131		}
132		}
133	3	$this->flush();
134	3	for ($j = 0; $j < $innerCount; ++$j) {
135	3	$this->jobTiminigManager->recordTiming(JobTiming::STATUS_FINISHED_EXPIRED);
136		}
137		}
138
139	3	return $finalCount;
140		}
141
142	15	protected function flush()
143		{
144	15	$this->getObjectManager()->flush();
145	15	}
146
147	4	protected function getStalledJobs($workerName = null, $method = null)
148		{
149	4	$count = $this->countJobsByStatus($this->getJobClass(), Job::STATUS_RUNNING, $workerName, $method);
150
151	4	$criterion = ['status' => BaseJob::STATUS_RUNNING];
152	4	$this->addWorkerNameMethod($criterion, $workerName, $method);
153
154	4	$runningJobs = $this->findRunningJobs($criterion, $count);
155
156	4	return $this->extractStalledJobs($runningJobs);
157		}
158
159	4	protected function findRunningJobs($criterion, $count)
160		{
161	4	$repository = $this->getRepository();
162	4	$runningJobsById = [];
163
164	4	for ($i = 0; $i < $count; $i += static::FETCH_COUNT) {
165	4	$runningJobs = $repository->findBy($criterion, null, static::FETCH_COUNT, $i);
166	4	if (!empty($runningJobs)) {
167	4	foreach ($runningJobs as $job) {
168		/** @var StallableJob $job */
169	4	if (null !== $runId = $job->getRunId()) {
170	4	$runningJobsById[$runId][] = $job;
171		}
172		}
173		}
174		}
175
176	4	return $runningJobsById;
177		}
178
179		/**
180		* @param $runId
181		* @param array $jobs
182		* @param array $stalledJobs
183		*/
184	4	protected function extractStalledLiveRuns($runId, array $jobs, array &$stalledJobs)
185		{
186	4	$objectManager = $this->getObjectManager();
187	4	$runRepository = $objectManager->getRepository($this->getRunManager()->getRunClass());
188	4	if ($run = $runRepository->find($runId)) {
189	2	foreach ($jobs as $job) {
190	2	if ($run->getCurrentJobId() == $job->getId()) {
191	2	continue;
192		}
193	2	$stalledJobs[] = $job;
194		}
195		}
196	4	}
197
198		/**
199		* @param array $runningJobsById
200		*
201		* @return array
202		*/
203	4	protected function extractStalledJobs(array $runningJobsById)
204		{
205	4	$stalledJobs = [];
206	4	foreach (array_keys($runningJobsById) as $runId) {
207	4	$this->extractStalledLiveRuns($runId, $runningJobsById[$runId], $stalledJobs);
208	4	$this->extractStalledJobsRunArchive($runningJobsById, $stalledJobs, $runId);
209		}
210
211	4	return $stalledJobs;
212		}
213
214	4	protected function extractStalledJobsRunArchive(array $runningJobsById, array &$stalledJobs, $runId)
215		{
216	4	$runManager = $this->getRunManager();
217	4	if (!method_exists($runManager, 'getObjectManager')) {
218		return;
219		}
220	4	if (!method_exists($runManager, 'getRunArchiveClass')) {
221		return;
222		}
223
224		/** @var EntityRepository\|DocumentRepository $runArchiveRepository */
225	4	$runArchiveRepository = $runManager->getObjectManager()->getRepository($runManager->getRunArchiveClass());
226		/** @var Run $run */
227	4	if ($run = $runArchiveRepository->find($runId)) {
228	4	if ($endTime = $run->getEndedAt()) {
229		// Did it end over an hour ago
230	4	if ((time() - $endTime->getTimestamp()) > static::STALLED_SECONDS) {
231	4	$stalledJobs = array_merge($stalledJobs, $runningJobsById[$runId]);
232		}
233		}
234		}
235	4	}
236
237	2	protected function runStalledLoop($i, $count, array $stalledJobs)
238		{
239	2	$resetCount = 0;
240	2	for ($j = $i, $max = $i + static::FETCH_COUNT; $j < $max && $j < $count; ++$j) {
241		/* StallableJob $job */
242	2	$job = $stalledJobs[$j];
243	2	$job->setStatus(StallableJob::STATUS_STALLED);
244	2	if ($this->saveHistory($job)) {
245	2	++$resetCount;
246		}
247		}
248
249	2	return $resetCount;
250		}
251
252	2	public function resetStalledJobs($workerName = null, $method = null)
253		{
254	2	$stalledJobs = $this->getStalledJobs($workerName, $method);
255
256	2	$countReset = 0;
257	2	for ($i = 0, $count = count($stalledJobs); $i < $count; $i += static::FETCH_COUNT) {
258	2	$resetCount = $this->runStalledLoop($i, $count, $stalledJobs);
259	2	for ($j = $i, $max = $i + static::FETCH_COUNT; $j < $max && $j < $count; ++$j) {
260	2	$this->jobTiminigManager->recordTiming(JobTiming::STATUS_FINISHED_STALLED);
261		}
262	2	$countReset += $resetCount;
263	2	$this->flush();
264		}
265
266	2	return $countReset;
267		}
268
269		/**
270		* @param string $workerName
271		* @param string $method
272		*/
273	2	public function pruneStalledJobs($workerName = null, $method = null)
274		{
275	2	$stalledJobs = $this->getStalledJobs($workerName, $method);
276
277	2	$countProcessed = 0;
278	2	for ($i = 0, $count = count($stalledJobs); $i < $count; $i += static::FETCH_COUNT) {
279	2	for ($j = $i, $max = $i + static::FETCH_COUNT; $j < $max && $j < $count; ++$j) {
280		/** @var StallableJob $job */
281	2	$job = $stalledJobs[$j];
282	2	$job->setStatus(StallableJob::STATUS_STALLED);
283	2	$job->setStalls(intval($job->getStalls()) + 1);
284	2	$this->deleteJob($job);
285	2	++$countProcessed;
286		}
287	2	$this->flush();
288		}
289
290	2	return $countProcessed;
291		}
292
293	20	public function deleteJob(\Dtc\QueueBundle\Model\Job $job)
294		{
295	20	$objectManager = $this->getObjectManager();
296	20	$objectManager->remove($job);
297	20	$objectManager->flush();
298	20	}
299
300	8	protected function stallableSaveHistory(StallableJob $job, $retry)
301		{
302	8	if (!$retry) {
303	8	$this->deleteJob($job);
304		}
305
306	8	return $retry;
307		}
308
309	43	protected function stallableSave(StallableJob $job)
310		{
311		// Generate crc hash for the job
312	43	$hashValues = array(get_class($job), $job->getMethod(), $job->getWorkerName(), $job->getArgs());
313	43	$crcHash = hash('sha256', serialize($hashValues));
314	43	$job->setCrcHash($crcHash);
315	43	$objectManager = $this->getObjectManager();
316
317	43	if (true === $job->getBatch()) {
318	2	$oldJob = $this->updateNearestBatch($job);
319	2	if ($oldJob) {
320	2	return $oldJob;
321		}
322		}
323
324		// Just save a new job
325	43	$this->resetSaveOk(__FUNCTION__);
326	43	$objectManager->persist($job);
327	43	$objectManager->flush();
328
329	43	return $job;
330		}
331
332		abstract protected function updateNearestBatch(Job $job);
333
334		/**
335		* @param string $objectName
336		*/
337		abstract protected function stopIdGenerator($objectName);
338
339		abstract protected function restoreIdGenerator($objectName);
340
341		/**
342		* @param array $criterion
343		* @param int $limit
344		* @param int $offset
345		*/
346	2	private function resetJobsByCriterion(
347		array $criterion,
348		$limit,
349		$offset
350		) {
351	2	$objectManager = $this->getObjectManager();
352	2	$this->resetSaveOk(__FUNCTION__);
353	2	$objectName = $this->getJobClass();
354	2	$archiveObjectName = $this->getJobArchiveClass();
355	2	$jobRepository = $objectManager->getRepository($objectName);
356	2	$jobArchiveRepository = $objectManager->getRepository($archiveObjectName);
357	2	$className = $jobRepository->getClassName();
358	2	$metadata = $objectManager->getClassMetadata($className);
359	2	$this->stopIdGenerator($objectName);
360	2	$identifierData = $metadata->getIdentifier();
361	2	$idColumn = isset($identifierData[0]) ? $identifierData[0] : 'id';
362	2	$results = $jobArchiveRepository->findBy($criterion, [$idColumn => 'ASC'], $limit, $offset);
363	2	$countProcessed = 0;
364
365	2	foreach ($results as $jobArchive) {
366	2	$countProcessed += $this->resetArchiveJob($jobArchive);
367		}
368	2	$objectManager->flush();
369
370	2	$this->restoreIdGenerator($objectName);
371
372	2	return $countProcessed;
373		}
374
375	22	protected function resetSaveOk($function)
376		{
377	22	}
378
379		/**
380		* @param null $workerName
381		* @param null $methodName
382		* @param bool $prioritize
383		*/
384		abstract public function getJobQueryBuilder($workerName = null, $methodName = null, $prioritize = true);
385
386		/**
387		* @param StallableJob $jobArchive
388		* @param $className
389		*
390		* @return int Number of jobs reset
391		*/
392	2	protected function resetArchiveJob(StallableJob $jobArchive)
393		{
394	2	$objectManager = $this->getObjectManager();
395	2	if ($this->updateMaxStatus($jobArchive, StallableJob::STATUS_MAX_RETRIES, $jobArchive->getMaxRetries(), $jobArchive->getRetries())) {
		0 ignored issues – show Bug introduced 2018-01-07 01:22 UTC by Report Bug Copy Issue Report It seems like `$jobArchive->getMaxRetries()` targeting `Dtc\QueueBundle\Model\Re...bleJob::getMaxRetries()` can also be of type `integer`; however, `Dtc\QueueBundle\Manager\...ager::updateMaxStatus()` does only seem to accept `null`, maybe add an additional type check? This check looks at variables that are passed out again to other methods. If the outgoing method call has stricter type requirements than the method itself, an issue is raised. An additional type check may prevent trouble. Loading history...
396	2	$objectManager->persist($jobArchive);
397
398	2	return 0;
399		}
400
401		/** @var StallableJob $job */
402	2	$className = $this->getJobClass();
403	2	$newJob = new $className();
404	2	Util::copy($jobArchive, $newJob);
405	2	$this->resetJob($newJob);
406	2	$objectManager->remove($jobArchive);
407	2	$this->flush();
408
409	2	return 1;
410		}
411
412	8	protected function resetJob(RetryableJob $job)
413		{
414	8	if (!$job instanceof StallableJob) {
415		throw new \InvalidArgumentException('$job should be instance of '.StallableJob::class);
416		}
417	8	$job->setStatus(BaseJob::STATUS_NEW);
418	8	$job->setMessage(null);
419	8	$job->setFinishedAt(null);
420	8	$job->setStartedAt(null);
421	8	$job->setElapsed(null);
422	8	$job->setRetries($job->getRetries() + 1);
423	8	$this->getObjectManager()->persist($job);
424	8	$this->flush();
425
426	8	return true;
427		}
428
429		abstract public function getWorkersAndMethods();
430
431		abstract public function countLiveJobs($workerName = null, $methodName = null);
432
433		abstract public function archiveAllJobs($workerName = null, $methodName = null, $progressCallback);
434		}
435

mmucklo / DtcQueueBundle

Pull Request — master (#30)

DoctrineJobManager::resetJobsByCriterion() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like