forked from mongodb/laravel-mongodb
-
Notifications
You must be signed in to change notification settings - Fork 37
/
MongoQueue.php
131 lines (117 loc) · 3.77 KB
/
MongoQueue.php
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
<?php namespace Moloquent\Queue;
use Carbon\Carbon;
use Illuminate\Queue\DatabaseQueue;
use Illuminate\Queue\Jobs\DatabaseJob;
use MongoDB\Operation\FindOneAndUpdate;
class MongoQueue extends DatabaseQueue
{
/**
* Pop the next job off of the queue.
*
* @param string $queue
* @return \Illuminate\Contracts\Queue\Job|null
*/
public function pop($queue = null)
{
$queue = $this->getQueue($queue);
if (!is_null($this->expire)) {
$this->releaseJobsThatHaveBeenReservedTooLong($queue);
}
if ($job = $this->getNextAvailableJobAndReserve($queue)) {
return new DatabaseJob(
$this->container, $this, $job, $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' => 0,
'available_at' => ['$lte' => $this->getTime()],
],
[
'$set' => [
'reserved' => 1,
'reserved_at' => $this->getTime(),
],
],
[
'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->expire)->getTimestamp();
$now = time();
$reserved = $this->database->collection($this->table)
->where('queue', $this->getQueue($queue))
->where(function ($query) use ($expiration, $now) {
// Check for available jobs
$query->where(function ($query) use ($now) {
$query->whereNull('reserved_at');
$query->where('available_at', '<=', $now);
});
// Check for jobs that are reserved but have expired
$query->orWhere('reserved_at', '<=', $expiration);
})->get();
foreach ($reserved as $job) {
$attempts = $job['attempts'] + 1;
$this->releaseJob($job['_id'], $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,
]);
}
/**
* Delete a reserved job from the queue.
*
* @param string $queue
* @param string $id
* @return void
*/
public function deleteReserved($queue, $id)
{
$this->database->collection($this->table)->where('_id', $id)->delete();
}
}