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
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
<?php
namespace Jenssegers\Mongodb\Queue;
use Carbon\Carbon;
use Illuminate\Queue\DatabaseQueue;
use Jenssegers\Mongodb\Connection;
use MongoDB\Operation\FindOneAndUpdate;
class MongoQueue extends DatabaseQueue
{
/**
* The expiration time of a job.
* @var int|null
*/
protected $retryAfter = 60;
/**
* The connection name for the queue.
* @var string
*/
protected $connectionName;
/**
* @inheritdoc
*/
public function __construct(Connection $database, $table, $default = 'default', $retryAfter = 60)
{
parent::__construct($database, $table, $default, $retryAfter);
$this->retryAfter = $retryAfter;
}
/**
* @inheritdoc
*/
public function pop($queue = null)
{
$queue = $this->getQueue($queue);
if ($this->retryAfter !== null) {
$this->releaseJobsThatHaveBeenReservedTooLong($queue);
}
if ($job = $this->getNextAvailableJobAndReserve($queue)) {
return new MongoJob(
$this->container, $this, $job, $this->connectionName, $queue
);
}
}
/**
* Get the next available job for the queue and mark it as reserved.
* When using multiple daemon queue listeners to process jobs there
* is a possibility that multiple processes can end up reading the
* same record before one has flagged it as reserved.
* This race condition can result in random jobs being run more then
* once. To solve this we use findOneAndUpdate to lock the next jobs
* record while flagging it as reserved at the same time.
* @param string|null $queue
* @return \StdClass|null
*/
protected function getNextAvailableJobAndReserve($queue)
{
$job = $this->database->getCollection($this->table)->findOneAndUpdate(
[
'queue' => $this->getQueue($queue),
'reserved' => ['$ne' => 1],
'available_at' => ['$lte' => Carbon::now()->getTimestamp()],
],
[
'$set' => [
'reserved' => 1,
'reserved_at' => Carbon::now()->getTimestamp(),
],
'$inc' => [
'attempts' => 1,
],
],
[
'returnDocument' => FindOneAndUpdate::RETURN_DOCUMENT_AFTER,
'sort' => ['available_at' => 1],
]
);
if ($job) {
$job->id = $job->_id;
}
return $job;
}
/**
* Release the jobs that have been reserved for too long.
* @param string $queue
* @return void
*/
protected function releaseJobsThatHaveBeenReservedTooLong($queue)
{
$expiration = Carbon::now()->subSeconds($this->retryAfter)->getTimestamp();
$reserved = $this->database->collection($this->table)
->where('queue', $this->getQueue($queue))
->whereNotNull('reserved_at')
->where('reserved_at', '<=', $expiration)
->get();
foreach ($reserved as $job) {
$this->releaseJob($job['_id'], $job['attempts']);
}
}
/**
* Release the given job ID from reservation.
* @param string $id
* @param int $attempts
* @return void
*/
protected function releaseJob($id, $attempts)
{
$this->database->table($this->table)->where('_id', $id)->update([
'reserved' => 0,
'reserved_at' => null,
'attempts' => $attempts,
]);
}
/**
* @inheritdoc
*/
public function deleteReserved($queue, $id)
{
$this->database->collection($this->table)->where('_id', $id)->delete();
}
}