JobManager::getJob() - Code Metrics - Inspection of "fix a bug with old job" - mmucklo/DtcQueueBundle - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 8a247e...4109b7 )

by Matthew

created 2017-11-14 03:28 UTC

JobManager::getJob() A

↳ Parent: JobManager

Complexity

Conditions	3
Paths	4

Size

Total Lines	57
Code Lines	38

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	27
CRAP Score	3.1406

Importance

Changes

Metric	Value
dl	0
loc	57
ccs	27
cts	36
cp	0.75
rs	9.6818
c	0
b	0
f	0
cc	3
eloc	38
nc	4
nop	4
crap	3.1406

How to fix Long Method

<?php

namespace Dtc\QueueBundle\ORM;

use Doctrine\DBAL\LockMode;
use Doctrine\ORM\EntityManager;
use Doctrine\ORM\EntityRepository;
use Doctrine\ORM\QueryBuilder;
use Dtc\QueueBundle\Doctrine\BaseJobManager;
use Dtc\QueueBundle\Entity\Job;
use Dtc\QueueBundle\Model\BaseJob;
use Dtc\QueueBundle\Model\RetryableJob;
use Symfony\Component\Process\Exception\LogicException;

class JobManager extends BaseJobManager
{
    use CommonTrait;
    protected static $saveInsertCalled = null;
    protected static $resetInsertCalled = null;

    public function countJobsByStatus($objectName, $status, $workerName = null, $method = null)
    {
        /** @var EntityManager $objectManager */
        $objectManager = $this->getObjectManager();

        $qb = $objectManager
            ->createQueryBuilder()
            ->select('count(a.id)')
            ->from($objectName, 'a')
            ->where('a.status = :status');

        if (null !== $workerName) {
            $qb->andWhere('a.workerName = :workerName')
                ->setParameter(':workerName', $workerName);
        }

        if (null !== $method) {
            $qb->andWhere('a.method = :method')
                ->setParameter(':method', $workerName);
        }

        $count = $qb->setParameter(':status', $status)
            ->getQuery()->getSingleScalarResult();

        if (!$count) {
            return 0;
        }

        return $count;
    }

    /**
     * @param string|null $workerName
     * @param string|null $method
     *
     * @return int Count of jobs pruned
     */
    public function pruneErroneousJobs($workerName = null, $method = null)
    {
        /** @var EntityManager $objectManager */
        $objectManager = $this->getObjectManager();
        $qb = $objectManager->createQueryBuilder()->delete($this->getArchiveObjectName(), 'j');
        $qb->where('j.status = :status')
            ->setParameter(':status', BaseJob::STATUS_ERROR);

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

        return intval($query->execute());
    }

    protected function resetSaveOk($function)
    {
        $objectManager = $this->getObjectManager();
        $splObjectHash = spl_object_hash($objectManager);

        if ('save' === $function) {
            $compare = static::$resetInsertCalled;
        } else {
            $compare = static::$saveInsertCalled;
        }

        if ($splObjectHash === $compare) {
            // Insert SQL is cached...
            $msg = "Can't call save and reset within the same process cycle (or using the same EntityManager)";
            throw new LogicException($msg);
        }

        if ('save' === $function) {
            static::$saveInsertCalled = spl_object_hash($objectManager);
        } else {
            static::$resetInsertCalled = spl_object_hash($objectManager);
        }
    }

    /**
     * @param string $workerName
     * @param string $method
     */
    protected function addWorkerNameCriterion(QueryBuilder $queryBuilder, $workerName = null, $method = null)
    {
        if (null !== $workerName) {
            $queryBuilder->andWhere('j.workerName = :workerName')->setParameter(':workerName', $workerName);
        }

        if (null !== $method) {
            $queryBuilder->andWhere('j.method = :method')->setParameter(':method', $method);
        }
    }

    protected function updateExpired($workerName = null, $method = null)
    {
        /** @var EntityManager $objectManager */
        $objectManager = $this->getObjectManager();
        $qb = $objectManager->createQueryBuilder()->update($this->getObjectName(), 'j');
        $qb->set('j.status', ':newStatus');
        $qb->where('j.expiresAt <= :expiresAt')
            ->setParameter(':expiresAt', new \DateTime());
        $qb->andWhere('j.status = :status')
            ->setParameter(':status', BaseJob::STATUS_NEW)
            ->setParameter(':newStatus', Job::STATUS_EXPIRED);

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

        return intval($query->execute());
    }

    protected function getJobCurrentStatus(\Dtc\QueueBundle\Model\Job $job)
    {
        /** @var EntityManager $objectManager */
        $objectManager = $this->getObjectManager();
        $qb = $objectManager->createQueryBuilder()->select('j.status')->from($this->getObjectName(), 'j');
        $qb->where('j.id = :id')->setParameter(':id', $job->getId());

        return $qb->getQuery()->getSingleScalarResult();
    }

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

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

        $qb = $qb->select('count(j)')->from($this->getObjectName(), 'j');

        $where = 'where';
        if (null !== $workerName) {
            if (null !== $method) {
                $qb->where($qb->expr()->andX(
                    $qb->expr()->eq('j.workerName', ':workerName'),
                                                $qb->expr()->eq('j.method', ':method')
                ))
                    ->setParameter(':method', $method);
            } else {
                $qb->where('j.workerName = :workerName');
            }
            $qb->setParameter(':workerName', $workerName);
            $where = 'andWhere';
        } elseif (null !== $method) {
            $qb->where('j.method = :method')->setParameter(':method', $method);
            $where = 'andWhere';
        }

        $dateTime = new \DateTime();
        // Filter
        $qb
            ->$where($qb->expr()->orX(
                $qb->expr()->isNull('j.whenAt'),
                                        $qb->expr()->lte('j.whenAt', ':whenAt')
            ))
            ->andWhere($qb->expr()->orX(
                $qb->expr()->isNull('j.expiresAt'),
                $qb->expr()->gt('j.expiresAt', ':expiresAt')
            ))
            ->andWhere('j.locked is NULL')
            ->setParameter(':whenAt', $dateTime)
            ->setParameter(':expiresAt', $dateTime);

        $query = $qb->getQuery();

        return $query->getSingleScalarResult();
    }

    /**
     * For ORM it's prudent to wrap things in a transaction.
     *
     * @param $i
     * @param $count
     * @param array $stalledJobs
     * @param $countProcessed
     */
    protected function runStalledLoop($i, $count, array $stalledJobs, &$countProcessed)
    {
        /** @var EntityManager $objectManager */
        $objectManager = $this->getObjectManager();
        try {
            $objectManager->beginTransaction();
            parent::runStalledLoop($i, $count, $stalledJobs, $countProcessed);
            $objectManager->commit();
        } catch (\Exception $exception) {
            $objectManager->rollback();

            // Try again
            parent::runStalledLoop($i, $count, $stalledJobs, $countProcessed);
        }
    }

    /**
     * Get Jobs statuses.
     */
    public function getStatus()
    {
        $result = [];
        $this->getStatusByEntityName($this->getObjectName(), $result);
        $this->getStatusByEntityName($this->getArchiveObjectName(), $result);

        $finalResult = [];
        foreach ($result as $key => $item) {
            ksort($item);
            foreach ($item as $status => $count) {
                if (isset($finalResult[$key][$status])) {
                    $finalResult[$key][$status] += $count;
                } else {
                    $finalResult[$key][$status] = $count;
                }
            }
        }

        return $finalResult;
    }

    /**
     * @param string $entityName
     */
    protected function getStatusByEntityName($entityName, array &$result)
    {
        /** @var EntityManager $objectManager */
        $objectManager = $this->getObjectManager();
        $result1 = $objectManager->getRepository($entityName)->createQueryBuilder('j')->select('j.workerName, j.method, j.status, count(j) as c')
            ->groupBy('j.workerName, j.method, j.status')->getQuery()->getArrayResult();

        foreach ($result1 as $item) {
            $method = $item['workerName'].'->'.$item['method'].'()';
            if (!isset($result[$method])) {
                $result[$method] = [BaseJob::STATUS_NEW => 0,
                    BaseJob::STATUS_RUNNING => 0,
                    RetryableJob::STATUS_EXPIRED => 0,
                    RetryableJob::STATUS_MAX_ERROR => 0,
                    RetryableJob::STATUS_MAX_STALLED => 0,
                    RetryableJob::STATUS_MAX_RETRIES => 0,
                    BaseJob::STATUS_SUCCESS => 0,
                    BaseJob::STATUS_ERROR => 0, ];
            }
            $result[$method][$item['status']] += intval($item['c']);
        }
    }

    /**
     * Get the next job to run (can be filtered by workername and method name).
     *
     * @param string $workerName
     * @param string $methodName
     * @param bool   $prioritize
     *
     * @return Job|null
     */
    public function getJob($workerName = null, $methodName = null, $prioritize = true, $runId = null)
    {
        /** @var EntityManager $objectManager */
        $objectManager = $this->getObjectManager();

        $objectManager->beginTransaction();

        /** @var EntityRepository $repository */
        $repository = $this->getRepository();
        $qb = $repository->createQueryBuilder('j');
        $dateTime = new \DateTime();
        $qb
            ->select('j')
            ->where('j.status = :status')->setParameter(':status', BaseJob::STATUS_NEW)
            ->andWhere('j.locked is NULL')
            ->andWhere($qb->expr()->orX(
                $qb->expr()->isNull('j.whenAt'),
                        $qb->expr()->lte('j.whenAt', ':whenAt')
            ))
            ->andWhere($qb->expr()->orX(
                $qb->expr()->isNull('j.expiresAt'),
                        $qb->expr()->gt('j.expiresAt', ':expiresAt')
            ))
            ->setParameter(':whenAt', $dateTime)
            ->setParameter(':expiresAt', $dateTime);

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

        if ($prioritize) {
            $qb->add('orderBy', 'j.priority DESC, j.whenAt ASC');
function acceptsInteger($int) { }

$x = '123'; // string "123"

// Instead of
acceptsInteger($x);

// we recommend to use
acceptsInteger((integer) $x);
        } else {
            $qb->orderBy('j.whenAt', 'ASC');
        }
        $qb->setMaxResults(1);

        /** @var QueryBuilder $qb */
        $query = $qb->getQuery();
        $query->setLockMode(LockMode::PESSIMISTIC_WRITE);
        $jobs = $query->getResult();

        if ($jobs) {
            /** @var Job $job */
            $job = $jobs[0];
            $job->setLocked(true);
            $job->setLockedAt(new \DateTime());
            $job->setStatus(BaseJob::STATUS_RUNNING);
            $job->setRunId($runId);
            $objectManager->commit();
            $objectManager->flush();

            return $job;
        }

        $objectManager->rollback();

        return null;
    }

    /**
     * Tries to update the nearest job as a batch.
     *
     * @param \Dtc\QueueBundle\Model\Job $job
     *
     * @return mixed|null
     */
    public function updateNearestBatch(\Dtc\QueueBundle\Model\Job $job)
    {
        $oldJob = null;
        do {
            try {
                /** @var EntityManager $entityManager */
                $entityManager = $this->getObjectManager();
                $entityManager->beginTransaction();

                /** @var QueryBuilder $queryBuilder */
                $queryBuilder = $this->getRepository()->createQueryBuilder('j');
                $queryBuilder->select()
                    ->where('j.crcHash = :crcHash')
                    ->andWhere('j.status = :status')
                    ->setParameter(':status', BaseJob::STATUS_NEW)
                    ->setParameter(':crcHash', $job->getCrcHash())
                    ->orderBy('j.whenAt', 'ASC')
                    ->setMaxResults(1);
                $oldJob = $queryBuilder->getQuery()->getFirstResult();

                if (!$oldJob) {
                    return null;
                }

                $oldJob->setPriority(max($job->getPriority(), $oldJob->getPriority()));
                $oldJob->setWhenAt(min($job->getWhenAt(), $oldJob->getWhenAt()));

                $entityManager->persist($oldJob);
                $entityManager->commit();
                $this->flush();
            } catch (\Exception $exception) {
                $entityManager->rollback();
            }
        } while (null === $oldJob);

        return $oldJob;
    }
}


1		<?php
2
3		namespace Dtc\QueueBundle\ORM;
4
5		use Doctrine\DBAL\LockMode;
6		use Doctrine\ORM\EntityManager;
7		use Doctrine\ORM\EntityRepository;
8		use Doctrine\ORM\QueryBuilder;
9		use Dtc\QueueBundle\Doctrine\BaseJobManager;
10		use Dtc\QueueBundle\Entity\Job;
11		use Dtc\QueueBundle\Model\BaseJob;
12		use Dtc\QueueBundle\Model\RetryableJob;
13		use Symfony\Component\Process\Exception\LogicException;
14
15		class JobManager extends BaseJobManager
16		{
17		use CommonTrait;
18		protected static $saveInsertCalled = null;
19		protected static $resetInsertCalled = null;
20
21	3	public function countJobsByStatus($objectName, $status, $workerName = null, $method = null)
22		{
23		/** @var EntityManager $objectManager */
24	3	$objectManager = $this->getObjectManager();
25
26		$qb = $objectManager
27	3	->createQueryBuilder()
28	3	->select('count(a.id)')
29	3	->from($objectName, 'a')
30	3	->where('a.status = :status');
31
32	3	if (null !== $workerName) {
33	1	$qb->andWhere('a.workerName = :workerName')
34	1	->setParameter(':workerName', $workerName);
35		}
36
37	3	if (null !== $method) {
38	1	$qb->andWhere('a.method = :method')
39	1	->setParameter(':method', $workerName);
40		}
41
42	3	$count = $qb->setParameter(':status', $status)
43	3	->getQuery()->getSingleScalarResult();
44
45	3	if (!$count) {
46	1	return 0;
47		}
48
49	3	return $count;
50		}
51
52		/**
53		* @param string\|null $workerName
54		* @param string\|null $method
55		*
56		* @return int Count of jobs pruned
57		*/
58	1	public function pruneErroneousJobs($workerName = null, $method = null)
59		{
60		/** @var EntityManager $objectManager */
61	1	$objectManager = $this->getObjectManager();
62	1	$qb = $objectManager->createQueryBuilder()->delete($this->getArchiveObjectName(), 'j');
63	1	$qb->where('j.status = :status')
64	1	->setParameter(':status', BaseJob::STATUS_ERROR);
65
66	1	$this->addWorkerNameCriterion($qb, $workerName, $method);
67	1	$query = $qb->getQuery();
68
69	1	return intval($query->execute());
70		}
71
72	6	protected function resetSaveOk($function)
73		{
74	6	$objectManager = $this->getObjectManager();
75	6	$splObjectHash = spl_object_hash($objectManager);
76
77	6	if ('save' === $function) {
78		$compare = static::$resetInsertCalled;
79		} else {
80	6	$compare = static::$saveInsertCalled;
81		}
82
83	6	if ($splObjectHash === $compare) {
84		// Insert SQL is cached...
85		$msg = "Can't call save and reset within the same process cycle (or using the same EntityManager)";
86		throw new LogicException($msg);
87		}
88
89	6	if ('save' === $function) {
90		static::$saveInsertCalled = spl_object_hash($objectManager);
91		} else {
92	6	static::$resetInsertCalled = spl_object_hash($objectManager);
93		}
94	6	}
95
96		/**
97		* @param string $workerName
98		* @param string $method
99		*/
100	2	protected function addWorkerNameCriterion(QueryBuilder $queryBuilder, $workerName = null, $method = null)
101		{
102	2	if (null !== $workerName) {
103	1	$queryBuilder->andWhere('j.workerName = :workerName')->setParameter(':workerName', $workerName);
104		}
105
106	2	if (null !== $method) {
107	1	$queryBuilder->andWhere('j.method = :method')->setParameter(':method', $method);
108		}
109	2	}
110
111		protected function updateExpired($workerName = null, $method = null)
112		{
113		/** @var EntityManager $objectManager */
114		$objectManager = $this->getObjectManager();
115		$qb = $objectManager->createQueryBuilder()->update($this->getObjectName(), 'j');
116		$qb->set('j.status', ':newStatus');
117		$qb->where('j.expiresAt <= :expiresAt')
118		->setParameter(':expiresAt', new \DateTime());
119		$qb->andWhere('j.status = :status')
120		->setParameter(':status', BaseJob::STATUS_NEW)
121		->setParameter(':newStatus', Job::STATUS_EXPIRED);
122
123		$this->addWorkerNameCriterion($qb, $workerName, $method);
124		$query = $qb->getQuery();
125
126		return intval($query->execute());
127		}
128
129	1	protected function getJobCurrentStatus(\Dtc\QueueBundle\Model\Job $job)
130		{
131		/** @var EntityManager $objectManager */
132	1	$objectManager = $this->getObjectManager();
133	1	$qb = $objectManager->createQueryBuilder()->select('j.status')->from($this->getObjectName(), 'j');
134	1	$qb->where('j.id = :id')->setParameter(':id', $job->getId());
135
136	1	return $qb->getQuery()->getSingleScalarResult();
137		}
138
139		/**
140		* Removes archived jobs older than $olderThan.
141		*
142		* @param \DateTime $olderThan
143		*/
144		public function pruneArchivedJobs(\DateTime $olderThan)
145		{
146		return $this->removeOlderThan($this->getArchiveObjectName(),
147		'updatedAt',
148		$olderThan);
149		}
150
151	1	public function getJobCount($workerName = null, $method = null)
152		{
153		/** @var EntityManager $objectManager */
154	1	$objectManager = $this->getObjectManager();
155	1	$qb = $objectManager->createQueryBuilder();
156
157	1	$qb = $qb->select('count(j)')->from($this->getObjectName(), 'j');
158
159	1	$where = 'where';
160	1	if (null !== $workerName) {
161		if (null !== $method) {
162		$qb->where($qb->expr()->andX(
163		$qb->expr()->eq('j.workerName', ':workerName'),
164		$qb->expr()->eq('j.method', ':method')
165		))
166		->setParameter(':method', $method);
167		} else {
168		$qb->where('j.workerName = :workerName');
169		}
170		$qb->setParameter(':workerName', $workerName);
171		$where = 'andWhere';
172	1	} elseif (null !== $method) {
173		$qb->where('j.method = :method')->setParameter(':method', $method);
174		$where = 'andWhere';
175		}
176
177	1	$dateTime = new \DateTime();
178		// Filter
179		$qb
180	1	->$where($qb->expr()->orX(
181	1	$qb->expr()->isNull('j.whenAt'),
182	1	$qb->expr()->lte('j.whenAt', ':whenAt')
183		))
184	1	->andWhere($qb->expr()->orX(
185	1	$qb->expr()->isNull('j.expiresAt'),
186	1	$qb->expr()->gt('j.expiresAt', ':expiresAt')
187		))
188	1	->andWhere('j.locked is NULL')
189	1	->setParameter(':whenAt', $dateTime)
190	1	->setParameter(':expiresAt', $dateTime);
191
192	1	$query = $qb->getQuery();
193
194	1	return $query->getSingleScalarResult();
195		}
196
197		/**
198		* For ORM it's prudent to wrap things in a transaction.
199		*
200		* @param $i
201		* @param $count
202		* @param array $stalledJobs
203		* @param $countProcessed
204		*/
205	1	protected function runStalledLoop($i, $count, array $stalledJobs, &$countProcessed)
206		{
207		/** @var EntityManager $objectManager */
208	1	$objectManager = $this->getObjectManager();
209		try {
210	1	$objectManager->beginTransaction();
211	1	parent::runStalledLoop($i, $count, $stalledJobs, $countProcessed);
212	1	$objectManager->commit();
213		} catch (\Exception $exception) {
214		$objectManager->rollback();
215
216		// Try again
217		parent::runStalledLoop($i, $count, $stalledJobs, $countProcessed);
218		}
219	1	}
220
221		/**
222		* Get Jobs statuses.
223		*/
224	1	public function getStatus()
225		{
226	1	$result = [];
227	1	$this->getStatusByEntityName($this->getObjectName(), $result);
228	1	$this->getStatusByEntityName($this->getArchiveObjectName(), $result);
229
230	1	$finalResult = [];
231	1	foreach ($result as $key => $item) {
232		ksort($item);
233		foreach ($item as $status => $count) {
234		if (isset($finalResult[$key][$status])) {
235		$finalResult[$key][$status] += $count;
236		} else {
237		$finalResult[$key][$status] = $count;
238		}
239		}
240		}
241
242	1	return $finalResult;
243		}
244
245		/**
246		* @param string $entityName
247		*/
248	1	protected function getStatusByEntityName($entityName, array &$result)
249		{
250		/** @var EntityManager $objectManager */
251	1	$objectManager = $this->getObjectManager();
252	1	$result1 = $objectManager->getRepository($entityName)->createQueryBuilder('j')->select('j.workerName, j.method, j.status, count(j) as c')
253	1	->groupBy('j.workerName, j.method, j.status')->getQuery()->getArrayResult();
254
255	1	foreach ($result1 as $item) {
256		$method = $item['workerName'].'->'.$item['method'].'()';
257		if (!isset($result[$method])) {
258		$result[$method] = [BaseJob::STATUS_NEW => 0,
259		BaseJob::STATUS_RUNNING => 0,
260		RetryableJob::STATUS_EXPIRED => 0,
261		RetryableJob::STATUS_MAX_ERROR => 0,
262		RetryableJob::STATUS_MAX_STALLED => 0,
263		RetryableJob::STATUS_MAX_RETRIES => 0,
264		BaseJob::STATUS_SUCCESS => 0,
265		BaseJob::STATUS_ERROR => 0, ];
266		}
267		$result[$method][$item['status']] += intval($item['c']);
268		}
269	1	}
270
271		/**
272		* Get the next job to run (can be filtered by workername and method name).
273		*
274		* @param string $workerName
275		* @param string $methodName
276		* @param bool $prioritize
277		*
278		* @return Job\|null
279		*/
280	1	public function getJob($workerName = null, $methodName = null, $prioritize = true, $runId = null)
281		{
282		/** @var EntityManager $objectManager */
283	1	$objectManager = $this->getObjectManager();
284
285	1	$objectManager->beginTransaction();
286
287		/** @var EntityRepository $repository */
288	1	$repository = $this->getRepository();
289	1	$qb = $repository->createQueryBuilder('j');
290	1	$dateTime = new \DateTime();
291		$qb
292	1	->select('j')
293	1	->where('j.status = :status')->setParameter(':status', BaseJob::STATUS_NEW)
294	1	->andWhere('j.locked is NULL')
295	1	->andWhere($qb->expr()->orX(
296	1	$qb->expr()->isNull('j.whenAt'),
297	1	$qb->expr()->lte('j.whenAt', ':whenAt')
298		))
299	1	->andWhere($qb->expr()->orX(
300	1	$qb->expr()->isNull('j.expiresAt'),
301	1	$qb->expr()->gt('j.expiresAt', ':expiresAt')
302		))
303	1	->setParameter(':whenAt', $dateTime)
304	1	->setParameter(':expiresAt', $dateTime);
305
306	1	$this->addWorkerNameCriterion($qb, $workerName, $methodName);
307
308	1	if ($prioritize) {
309	1	$qb->add('orderBy', 'j.priority DESC, j.whenAt ASC');
		0 ignored issues – show Documentation introduced 2017-11-01 01:45 UTC by Report Bug Copy Issue Report `'j.priority DESC, j.whenAt ASC'` is of type `string`, but the function expects a `object<Doctrine\ORM\Query\Expr\Base>`. It seems like the type of the argument is not accepted by the function/method which you are calling. In some cases, in particular if PHP’s automatic type-juggling kicks in this might be fine. In other cases, however this might be a bug. We suggest to add an explicit type cast like in the following example: function acceptsInteger($int) { } $x = '123'; // string "123" // Instead of acceptsInteger($x); // we recommend to use acceptsInteger((integer) $x); Loading history...
310		} else {
311		$qb->orderBy('j.whenAt', 'ASC');
312		}
313	1	$qb->setMaxResults(1);
314
315		/** @var QueryBuilder $qb */
316	1	$query = $qb->getQuery();
317	1	$query->setLockMode(LockMode::PESSIMISTIC_WRITE);
318	1	$jobs = $query->getResult();
319
320	1	if ($jobs) {
321		/** @var Job $job */
322		$job = $jobs[0];
323		$job->setLocked(true);
324		$job->setLockedAt(new \DateTime());
325		$job->setStatus(BaseJob::STATUS_RUNNING);
326		$job->setRunId($runId);
327		$objectManager->commit();
328		$objectManager->flush();
329
330		return $job;
331		}
332
333	1	$objectManager->rollback();
334
335	1	return null;
336		}
337
338		/**
339		* Tries to update the nearest job as a batch.
340		*
341		* @param \Dtc\QueueBundle\Model\Job $job
342		*
343		* @return mixed\|null
344		*/
345	1	public function updateNearestBatch(\Dtc\QueueBundle\Model\Job $job)
346		{
347	1	$oldJob = null;
348		do {
349		try {
350		/** @var EntityManager $entityManager */
351	1	$entityManager = $this->getObjectManager();
352	1	$entityManager->beginTransaction();
353
354		/** @var QueryBuilder $queryBuilder */
355	1	$queryBuilder = $this->getRepository()->createQueryBuilder('j');
356	1	$queryBuilder->select()
357	1	->where('j.crcHash = :crcHash')
358	1	->andWhere('j.status = :status')
359	1	->setParameter(':status', BaseJob::STATUS_NEW)
360	1	->setParameter(':crcHash', $job->getCrcHash())
361	1	->orderBy('j.whenAt', 'ASC')
362	1	->setMaxResults(1);
363	1	$oldJob = $queryBuilder->getQuery()->getFirstResult();
364
365	1	if (!$oldJob) {
366	1	return null;
367		}
368
369		$oldJob->setPriority(max($job->getPriority(), $oldJob->getPriority()));
370		$oldJob->setWhenAt(min($job->getWhenAt(), $oldJob->getWhenAt()));
371
372		$entityManager->persist($oldJob);
373		$entityManager->commit();
374		$this->flush();
375		} catch (\Exception $exception) {
376		$entityManager->rollback();
377		}
378		} while (null === $oldJob);
379
380		return $oldJob;
381		}
382		}
383

mmucklo / DtcQueueBundle

Push — master ( 8a247e...4109b7 )

JobManager::getJob() A

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like