Completed
Push — master ( bf1f19...1204eb )
by Matthew
07:02
created

JobManager::pruneErroneousJobs()   A

Complexity

Conditions 1
Paths 1

Size

Total Lines 13
Code Lines 8

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 8
CRAP Score 1

Importance

Changes 0
Metric Value
dl 0
loc 13
ccs 8
cts 8
cp 1
rs 9.4285
c 0
b 0
f 0
cc 1
eloc 8
nc 1
nop 2
crap 1
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
14
class JobManager extends BaseJobManager
15
{
16
    use CommonTrait;
17
    protected static $saveInsertCalled = null;
18
    protected static $resetInsertCalled = null;
19
20 3
    public function countJobsByStatus($objectName, $status, $workerName = null, $method = null)
21
    {
22
        /** @var EntityManager $objectManager */
23 3
        $objectManager = $this->getObjectManager();
24
25
        $qb = $objectManager
26 3
            ->createQueryBuilder()
27 3
            ->select('count(a.id)')
28 3
            ->from($objectName, 'a')
29 3
            ->where('a.status = :status');
30
31 3
        if (null !== $workerName) {
32 1
            $qb->andWhere('a.workerName = :workerName')
33 1
                ->setParameter(':workerName', $workerName);
34
        }
35
36 3
        if (null !== $method) {
37 1
            $qb->andWhere('a.method = :method')
38 1
                ->setParameter(':method', $workerName);
39
        }
40
41 3
        $count = $qb->setParameter(':status', $status)
42 3
            ->getQuery()->getSingleScalarResult();
43
44 3
        if (!$count) {
45 1
            return 0;
46
        }
47
48 3
        return $count;
49
    }
50
51
    /**
52
     * @param string|null $workerName
53
     * @param string|null $method
54
     *
55
     * @return int Count of jobs pruned
56
     */
57 1
    public function pruneErroneousJobs($workerName = null, $method = null)
58
    {
59
        /** @var EntityManager $objectManager */
60 1
        $objectManager = $this->getObjectManager();
61 1
        $qb = $objectManager->createQueryBuilder()->delete($this->getArchiveObjectName(), 'j');
62 1
        $qb->where('j.status = :status')
63 1
            ->setParameter(':status', BaseJob::STATUS_ERROR);
64
65 1
        $this->addWorkerNameCriterion($qb, $workerName, $method);
66 1
        $query = $qb->getQuery();
67
68 1
        return intval($query->execute());
69
    }
70
71 14
    protected function resetSaveOk($function)
72
    {
73 14
        $objectManager = $this->getObjectManager();
74 14
        $splObjectHash = spl_object_hash($objectManager);
75
76 14
        if ('save' === $function) {
77
            $compare = static::$resetInsertCalled;
78
        } else {
79 14
            $compare = static::$saveInsertCalled;
80
        }
81
82 14
        if ($splObjectHash === $compare) {
83
            // Insert SQL is cached...
84
            $msg = "Can't call save and reset within the same process cycle (or using the same EntityManager)";
85
            throw new \Exception($msg);
86
        }
87
88 14
        if ('save' === $function) {
89
            static::$saveInsertCalled = spl_object_hash($objectManager);
90
        } else {
91 14
            static::$resetInsertCalled = spl_object_hash($objectManager);
92
        }
93 14
    }
94
95
    /**
96
     * @param string $workerName
97
     * @param string $method
98
     */
99 8
    protected function addWorkerNameCriterion(QueryBuilder $queryBuilder, $workerName = null, $method = null)
100
    {
101 8
        if (null !== $workerName) {
102 3
            $queryBuilder->andWhere('j.workerName = :workerName')->setParameter(':workerName', $workerName);
103
        }
104
105 8
        if (null !== $method) {
106 2
            $queryBuilder->andWhere('j.method = :method')->setParameter(':method', $method);
107
        }
108 8
    }
109
110 1
    protected function updateExpired($workerName = null, $method = null)
111
    {
112
        /** @var EntityManager $objectManager */
113 1
        $objectManager = $this->getObjectManager();
114 1
        $qb = $objectManager->createQueryBuilder()->update($this->getObjectName(), 'j');
115 1
        $qb->set('j.status', ':newStatus');
116 1
        $qb->where('j.expiresAt <= :expiresAt')
117 1
            ->setParameter(':expiresAt', new \DateTime());
118 1
        $qb->andWhere('j.status = :status')
119 1
            ->setParameter(':status', BaseJob::STATUS_NEW)
120 1
            ->setParameter(':newStatus', Job::STATUS_EXPIRED);
121
122 1
        $this->addWorkerNameCriterion($qb, $workerName, $method);
123 1
        $query = $qb->getQuery();
124
125 1
        return intval($query->execute());
126
    }
127
128
    /**
129
     * Removes archived jobs older than $olderThan.
130
     *
131
     * @param \DateTime $olderThan
132
     */
133 1
    public function pruneArchivedJobs(\DateTime $olderThan)
134
    {
135 1
        return $this->removeOlderThan($this->getArchiveObjectName(),
136 1
                'updatedAt',
137 1
                $olderThan);
138
    }
139
140 2
    public function getJobCount($workerName = null, $method = null)
141
    {
142
        /** @var EntityManager $objectManager */
143 2
        $objectManager = $this->getObjectManager();
144 2
        $qb = $objectManager->createQueryBuilder();
145
146 2
        $qb = $qb->select('count(j)')->from($this->getObjectName(), 'j');
147
148 2
        $where = 'where';
149 2
        if (null !== $workerName) {
150
            if (null !== $method) {
151
                $qb->where($qb->expr()->andX(
152
                    $qb->expr()->eq('j.workerName', ':workerName'),
153
                                                $qb->expr()->eq('j.method', ':method')
154
                ))
155
                    ->setParameter(':method', $method);
156
            } else {
157
                $qb->where('j.workerName = :workerName');
158
            }
159
            $qb->setParameter(':workerName', $workerName);
160
            $where = 'andWhere';
161 2
        } elseif (null !== $method) {
162
            $qb->where('j.method = :method')->setParameter(':method', $method);
163
            $where = 'andWhere';
164
        }
165
166 2
        $dateTime = new \DateTime();
167
        // Filter
168
        $qb
169 2
            ->$where($qb->expr()->orX(
170 2
                $qb->expr()->isNull('j.whenAt'),
171 2
                                        $qb->expr()->lte('j.whenAt', ':whenAt')
172
            ))
173 2
            ->andWhere($qb->expr()->orX(
174 2
                $qb->expr()->isNull('j.expiresAt'),
175 2
                $qb->expr()->gt('j.expiresAt', ':expiresAt')
176
            ))
177 2
            ->andWhere('j.locked is NULL')
178 2
            ->setParameter(':whenAt', $dateTime)
179 2
            ->setParameter(':expiresAt', $dateTime);
180
181 2
        $query = $qb->getQuery();
182
183 2
        return $query->getSingleScalarResult();
184
    }
185
186
    /**
187
     * Get Jobs statuses.
188
     */
189 2
    public function getStatus()
190
    {
191 2
        $result = [];
192 2
        $this->getStatusByEntityName($this->getObjectName(), $result);
193 2
        $this->getStatusByEntityName($this->getArchiveObjectName(), $result);
194
195 2
        $finalResult = [];
196 2
        foreach ($result as $key => $item) {
197 1
            ksort($item);
198 1
            foreach ($item as $status => $count) {
199 1
                if (isset($finalResult[$key][$status])) {
200
                    $finalResult[$key][$status] += $count;
201
                } else {
202 1
                    $finalResult[$key][$status] = $count;
203
                }
204
            }
205
        }
206
207 2
        return $finalResult;
208
    }
209
210
    /**
211
     * @param string $entityName
212
     */
213 2
    protected function getStatusByEntityName($entityName, array &$result)
214
    {
215
        /** @var EntityManager $objectManager */
216 2
        $objectManager = $this->getObjectManager();
217 2
        $result1 = $objectManager->getRepository($entityName)->createQueryBuilder('j')->select('j.workerName, j.method, j.status, count(j) as c')
218 2
            ->groupBy('j.workerName, j.method, j.status')->getQuery()->getArrayResult();
219
220 2
        foreach ($result1 as $item) {
221 1
            $method = $item['workerName'].'->'.$item['method'].'()';
222 1
            if (!isset($result[$method])) {
223 1
                $result[$method] = [BaseJob::STATUS_NEW => 0,
224
                    BaseJob::STATUS_RUNNING => 0,
225
                    RetryableJob::STATUS_EXPIRED => 0,
226
                    RetryableJob::STATUS_MAX_ERROR => 0,
227
                    RetryableJob::STATUS_MAX_STALLED => 0,
228
                    RetryableJob::STATUS_MAX_RETRIES => 0,
229
                    BaseJob::STATUS_SUCCESS => 0,
230
                    BaseJob::STATUS_ERROR => 0, ];
231
            }
232 1
            $result[$method][$item['status']] += intval($item['c']);
233
        }
234 2
    }
235
236
    /**
237
     * Get the next job to run (can be filtered by workername and method name).
238
     *
239
     * @param string $workerName
240
     * @param string $methodName
241
     * @param bool   $prioritize
242
     *
243
     * @return Job|null
244
     */
245 6
    public function getJob($workerName = null, $methodName = null, $prioritize = true, $runId = null)
246
    {
247 6
        $uniqid = uniqid(gethostname().'-'.getmypid(), true);
248 6
        $hash = hash('sha256', $uniqid);
249
250
        /** @var EntityManager $objectManager */
251 6
        $objectManager = $this->getObjectManager();
252
253 6
        $objectManager->beginTransaction();
254
255
        /** @var EntityRepository $repository */
256 6
        $repository = $this->getRepository();
257 6
        $qb = $repository->createQueryBuilder('j');
258 6
        $dateTime = new \DateTime();
259
        $qb
260 6
            ->select('j')
261 6
            ->where('j.status = :status')->setParameter(':status', BaseJob::STATUS_NEW)
262 6
            ->andWhere('j.locked is NULL')
263 6
            ->andWhere($qb->expr()->orX(
264 6
                $qb->expr()->isNull('j.whenAt'),
265 6
                        $qb->expr()->lte('j.whenAt', ':whenAt')
266
            ))
267 6
            ->andWhere($qb->expr()->orX(
268 6
                $qb->expr()->isNull('j.expiresAt'),
269 6
                        $qb->expr()->gt('j.expiresAt', ':expiresAt')
270
            ))
271 6
            ->setParameter(':whenAt', $dateTime)
272 6
            ->setParameter(':expiresAt', $dateTime);
273
274 6
        $this->addWorkerNameCriterion($qb, $workerName, $methodName);
275
276 6
        if ($prioritize) {
277 6
            $qb->add('orderBy', 'j.priority DESC, j.whenAt ASC');
0 ignored issues
show
Documentation introduced by
'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...
278
        } else {
279
            $qb->orderBy('j.whenAt', 'ASC');
280
        }
281 6
        $qb->setMaxResults(1);
282
283
        /** @var QueryBuilder $qb */
284 6
        $query = $qb->getQuery();
285 6
        $query->setLockMode(LockMode::PESSIMISTIC_WRITE);
286 6
        $jobs = $query->getResult();
287
288 6
        if ($jobs) {
289
            /** @var Job $job */
290 5
            $job = $jobs[0];
291 5
            if (!$job) {
292
                throw new \Exception("No job found for $hash, even though last result was count ".count($jobs));
293
            }
294 5
            $job->setLocked(true);
295 5
            $job->setLockedAt(new \DateTime());
296 5
            $job->setStatus(BaseJob::STATUS_RUNNING);
297 5
            $job->setRunId($runId);
298 5
            $objectManager->commit();
299 5
            $objectManager->flush();
300
301 5
            return $job;
302
        }
303
304 3
        $objectManager->rollback();
305
306 3
        return null;
307
    }
308
309
    /**
310
     * Tries to update the nearest job as a batch.
311
     *
312
     * @param \Dtc\QueueBundle\Model\Job $job
313
     *
314
     * @return mixed|null
315
     */
316 1
    public function updateNearestBatch(\Dtc\QueueBundle\Model\Job $job)
317
    {
318 1
        $oldJob = null;
319
        do {
320
            try {
321
                /** @var EntityManager $entityManager */
322 1
                $entityManager = $this->getObjectManager();
323 1
                $entityManager->beginTransaction();
324
325
                /** @var QueryBuilder $queryBuilder */
326 1
                $queryBuilder = $this->getRepository()->createQueryBuilder('j');
327 1
                $queryBuilder->select()
328 1
                    ->where('j.crcHash = :crcHash')
329 1
                    ->andWhere('j.status = :status')
330 1
                    ->setParameter(':status', BaseJob::STATUS_NEW)
331 1
                    ->setParameter(':crcHash', $job->getCrcHash())
332 1
                    ->orderBy('j.whenAt', 'ASC')
333 1
                    ->setMaxResults(1);
334 1
                $oldJob = $queryBuilder->getQuery()->getSingleResult();
335
336 1
                if (!$oldJob) {
337
                    return null;
338
                }
339
340 1
                $oldJob->setPriority(max($job->getPriority(), $oldJob->getPriority()));
341 1
                $oldJob->setWhenAt(min($job->getWhenAt(), $oldJob->getWhenAt()));
342
343 1
                $entityManager->persist($oldJob);
344 1
                $entityManager->commit();
345 1
                $this->flush();
346
            } catch (\Exception $exception) {
347
                $entityManager->rollback();
348
            }
349 1
        } while (null === $oldJob);
350
351 1
        return $oldJob;
352
    }
353
}
354