1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
|
<?php
namespace Digitigrade;
use Pheanstalk\Contract\JobIdInterface;
use Pheanstalk\Pheanstalk;
use Pheanstalk\Values\TubeName;
class JobQueue extends Singleton {
private Pheanstalk $conn;
private const PRIO_URGENT = 10;
private const PRIO_DELAYED = 20;
private const PRIO_RETRY = 30;
private const PRIO_NORMAL = 50;
private const PRIO_LAZY = 100;
private const TIMEOUT = 60; // seconds
private const RETRY_DELAY = 5; // seconds, to the power of the number of tries
public function __construct() {
$config = GlobalConfig::getInstance();
$this->conn = Pheanstalk::create($config->getQueueHost(), $config->getQueuePort());
$this->conn->useTube(new TubeName($config->getQueueChannel()));
}
public function subscribe() {
$this->conn->watch(new TubeName(GlobalConfig::getInstance()->getQueueChannel()));
}
private function submit(Job $job, int $priority, int $delay = 0): JobIdInterface {
return $this->conn->put(json_encode($job), $priority, $delay, self::TIMEOUT);
}
public function submitNormal(Job $job) {
$this->submit($job, self::PRIO_NORMAL);
}
public function submitUrgent(Job $job) {
$this->submit($job, self::PRIO_URGENT);
}
public function submitLazy(Job $job) {
$this->submit($job, self::PRIO_LAZY);
}
public function submitDelayed(Job $job, int $delaySecs) {
$this->submit($job, self::PRIO_NORMAL, $delaySecs);
}
public function submitDelayedUntil(Job $job, \DateTimeInterface $until) {
$delay = $until->getTimestamp() - (new \DateTimeImmutable())->getTimestamp();
if ($delay < 0)
throw new \InvalidArgumentException('"until" date is in the past!');
$this->submitDelayed($job, $delay);
}
public function runNext() {
$beanJob = $this->conn->reserve();
$job = Job::fromJson($beanJob->getData());
$job->remainingTries--;
$job->doneTries++;
$log = Logger::getInstance();
$log->info('Running job ' . $beanJob->getId());
try {
$job->run();
$this->conn->delete($beanJob);
$log->info('Job ' . $beanJob->getId() . ' completed successfully');
} catch (\Throwable $e) {
if ($job->remainingTries <= 0) {
$this->conn->delete($beanJob);
$log->error(
'Ran out of retries trying to run job ' . $beanJob->getId() . ': ' . $beanJob->getData()
. "\n" . $e->getMessage()
);
} else {
$this->conn->delete($beanJob);
$id = $this->submit($job, self::PRIO_RETRY, pow(self::RETRY_DELAY, $job->doneTries));
$log->warning(
'A job of type ' . $job::class . ' (id ' . $beanJob->getId() . ') failed. '
. 'Resubmitted as ' . $id->getId() . '; '
. $job->remainingTries . ' retries remaining'
);
}
}
}
}
|
