Skip to content

Commit a041fb5

Browse files
taylorotwelltaylorotwell
committed
Refactor database and beanstalk queues.
1 parent 101791e commit a041fb5

File tree

6 files changed

+168
-129
lines changed

6 files changed

+168
-129
lines changed

‎src/Illuminate/Queue/CalculatesDelays.php

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,19 @@ protected function secondsUntil($delay)
2020
: (int) $delay;
2121
}
2222

23+
/**
24+
* Get the "available at" UNIX timestamp.
25+
*
26+
* @param \DateTimeInterface|int $delay
27+
* @return int
28+
*/
29+
protected function availableAt($delay = 0)
30+
{
31+
return $delay instanceof DateTimeInterface
32+
? $delay->getTimestamp()
33+
: Carbon::now()->addSeconds($delay)->getTimestamp();
34+
}
35+
2336
/**
2437
* Get the current system time as a UNIX timestamp.
2538
*

‎src/Illuminate/Queue/DatabaseQueue.php

Lines changed: 69 additions & 76 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
use Carbon\Carbon;
77
use Illuminate\Database\Connection;
88
use Illuminate\Queue\Jobs\DatabaseJob;
9+
use Illuminate\Queue\Jobs\DatabaseJobRecord;
910
use Illuminate\Contracts\Queue\Queue as QueueContract;
1011

1112
class DatabaseQueue extends Queue implements QueueContract
@@ -36,23 +37,23 @@ class DatabaseQueue extends Queue implements QueueContract
3637
*
3738
* @var int|null
3839
*/
39-
protected $expire = 60;
40+
protected $retryAfter = 60;
4041

4142
/**
4243
* Create a new database queue instance.
4344
*
4445
* @param \Illuminate\Database\Connection $database
4546
* @param string $table
4647
* @param string $default
47-
* @param int $expire
48+
* @param int $retryAfter
4849
* @return void
4950
*/
50-
public function __construct(Connection $database, $table, $default = 'default', $expire = 60)
51+
public function __construct(Connection $database, $table, $default = 'default', $retryAfter = 60)
5152
{
5253
$this->table = $table;
53-
$this->expire = $expire;
5454
$this->default = $default;
5555
$this->database = $database;
56+
$this->retryAfter = $retryAfter;
5657
}
5758

5859
/**
@@ -78,7 +79,7 @@ public function size($queue = null)
7879
*/
7980
public function push($job, $data = '', $queue = null)
8081
{
81-
return $this->pushToDatabase(0, $queue, $this->createPayload($job, $data));
82+
return $this->pushToDatabase($queue, $this->createPayload($job, $data));
8283
}
8384

8485
/**
@@ -91,7 +92,7 @@ public function push($job, $data = '', $queue = null)
9192
*/
9293
public function pushRaw($payload, $queue = null, array $options = [])
9394
{
94-
return $this->pushToDatabase(0, $queue, $payload);
95+
return $this->pushToDatabase($queue, $payload);
9596
}
9697

9798
/**
@@ -105,7 +106,7 @@ public function pushRaw($payload, $queue = null, array $options = [])
105106
*/
106107
public function later($delay, $job, $data = '', $queue = null)
107108
{
108-
return $this->pushToDatabase($delay, $queue, $this->createPayload($job, $data));
109+
return $this->pushToDatabase($queue, $this->createPayload($job, $data), $delay);
109110
}
110111

111112
/**
@@ -120,46 +121,63 @@ public function bulk($jobs, $data = '', $queue = null)
120121
{
121122
$queue = $this->getQueue($queue);
122123

123-
$availableAt = $this->getAvailableAt(0);
124+
$availableAt = $this->availableAt();
124125

125-
$records = array_map(function ($job) use ($queue, $data, $availableAt) {
126-
return $this->buildDatabaseRecord(
127-
$queue, $this->createPayload($job, $data), $availableAt
128-
);
129-
}, (array) $jobs);
130-
131-
return $this->database->table($this->table)->insert($records);
126+
return $this->database->table($this->table)->insert(collect((array) $jobs)->map(
127+
function ($job) use ($queue, $data, $availableAt) {
128+
return $this->buildDatabaseRecord($queue, $this->createPayload($job, $data), $availableAt);
129+
}
130+
)->all());
132131
}
133132

134133
/**
135134
* Release a reserved job back onto the queue.
136135
*
137136
* @param string $queue
138-
* @param \StdClass $job
137+
* @param \Illuminate\Queue\Jobs\DatabaseJobRecord $job
139138
* @param int $delay
140139
* @return mixed
141140
*/
142141
public function release($queue, $job, $delay)
143142
{
144-
return $this->pushToDatabase($delay, $queue, $job->payload, $job->attempts);
143+
return $this->pushToDatabase($queue, $job->payload, $delay, $job->attempts);
145144
}
146145

147146
/**
148147
* Push a raw payload to the database with a given delay.
149148
*
150-
* @param \DateTime|int $delay
151149
* @param string|null $queue
152150
* @param string $payload
151+
* @param \DateTime|int $delay
153152
* @param int $attempts
154153
* @return mixed
155154
*/
156-
protected function pushToDatabase($delay, $queue, $payload, $attempts = 0)
155+
protected function pushToDatabase($queue, $payload, $delay = 0, $attempts = 0)
157156
{
158-
$attributes = $this->buildDatabaseRecord(
159-
$this->getQueue($queue), $payload, $this->getAvailableAt($delay), $attempts
160-
);
157+
return $this->database->table($this->table)->insertGetId($this->buildDatabaseRecord(
158+
$this->getQueue($queue), $payload, $this->availableAt($delay), $attempts
159+
));
160+
}
161161

162-
return $this->database->table($this->table)->insertGetId($attributes);
162+
/**
163+
* Create an array to insert for the given job.
164+
*
165+
* @param string|null $queue
166+
* @param string $payload
167+
* @param int $availableAt
168+
* @param int $attempts
169+
* @return array
170+
*/
171+
protected function buildDatabaseRecord($queue, $payload, $availableAt, $attempts = 0)
172+
{
173+
return [
174+
'queue' => $queue,
175+
'payload' => $payload,
176+
'attempts' => $attempts,
177+
'reserved_at' => null,
178+
'available_at' => $availableAt,
179+
'created_at' => $this->currentTime(),
180+
];
163181
}
164182

165183
/**
@@ -175,13 +193,7 @@ public function pop($queue = null)
175193
$this->database->beginTransaction();
176194

177195
if ($job = $this->getNextAvailableJob($queue)) {
178-
$job = $this->markJobAsReserved($job);
179-
180-
$this->database->commit();
181-
182-
return new DatabaseJob(
183-
$this->container, $this, $job, $queue
184-
);
196+
return $this->marshalJob($queue, $job);
185197
}
186198

187199
$this->database->commit();
@@ -191,7 +203,7 @@ public function pop($queue = null)
191203
* Get the next available job for the queue.
192204
*
193205
* @param string|null $queue
194-
* @return \StdClass|null
206+
* @return \Illuminate\Queue\Jobs\DatabaseJobRecord|null
195207
*/
196208
protected function getNextAvailableJob($queue)
197209
{
@@ -205,7 +217,7 @@ protected function getNextAvailableJob($queue)
205217
->orderBy('id', 'asc')
206218
->first();
207219

208-
return $job ? (object) $job : null;
220+
return $job ? new DatabaseJobRecord((object) $job) : null;
209221
}
210222

211223
/**
@@ -217,8 +229,8 @@ protected function getNextAvailableJob($queue)
217229
protected function isAvailable($query)
218230
{
219231
$query->where(function ($query) {
220-
$query->whereNull('reserved_at');
221-
$query->where('available_at', '<=', $this->currentTime());
232+
$query->whereNull('reserved_at')
233+
->where('available_at', '<=', $this->currentTime());
222234
});
223235
}
224236

@@ -230,27 +242,42 @@ protected function isAvailable($query)
230242
*/
231243
protected function isReservedButExpired($query)
232244
{
233-
$expiration = Carbon::now()->subSeconds($this->expire)->getTimestamp();
245+
$expiration = Carbon::now()->subSeconds($this->retryAfter)->getTimestamp();
234246

235247
$query->orWhere(function ($query) use ($expiration) {
236248
$query->where('reserved_at', '<=', $expiration);
237249
});
238250
}
239251

252+
/**
253+
* Marshal the reserved job into a DatabaseJob instance.
254+
*
255+
* @param string $queue
256+
* @param \Illuminate\Queue\Jobs\DatabaseJobRecord $job
257+
* @return \Illuminate\Queue\Jobs\DatabaseJob
258+
*/
259+
protected function marshalJob($queue, $job)
260+
{
261+
$job = $this->markJobAsReserved($job);
262+
263+
$this->database->commit();
264+
265+
return new DatabaseJob(
266+
$this->container, $this, $job, $queue
267+
);
268+
}
269+
240270
/**
241271
* Mark the given job ID as reserved.
242272
*
243-
* @param \stdClass $job
244-
* @return \stdClass
273+
* @param \Illuminate\Queue\Jobs\DatabaseJobRecord $job
274+
* @return \Illuminate\Queue\Jobs\DatabaseJobRecord
245275
*/
246276
protected function markJobAsReserved($job)
247277
{
248-
$job->attempts = $job->attempts + 1;
249-
$job->reserved_at = $this->currentTime();
250-
251278
$this->database->table($this->table)->where('id', $job->id)->update([
252-
'reserved_at' => $job->reserved_at,
253-
'attempts' => $job->attempts,
279+
'reserved_at' => $job->touch(),
280+
'attempts' => $job->increment(),
254281
]);
255282

256283
return $job;
@@ -274,40 +301,6 @@ public function deleteReserved($queue, $id)
274301
$this->database->commit();
275302
}
276303

277-
/**
278-
* Get the "available at" UNIX timestamp.
279-
*
280-
* @param \DateTime|int $delay
281-
* @return int
282-
*/
283-
protected function getAvailableAt($delay)
284-
{
285-
$availableAt = $delay instanceof DateTime ? $delay : Carbon::now()->addSeconds($delay);
286-
287-
return $availableAt->getTimestamp();
288-
}
289-
290-
/**
291-
* Create an array to insert for the given job.
292-
*
293-
* @param string|null $queue
294-
* @param string $payload
295-
* @param int $availableAt
296-
* @param int $attempts
297-
* @return array
298-
*/
299-
protected function buildDatabaseRecord($queue, $payload, $availableAt, $attempts = 0)
300-
{
301-
return [
302-
'queue' => $queue,
303-
'attempts' => $attempts,
304-
'reserved_at' => null,
305-
'available_at' => $availableAt,
306-
'created_at' => $this->currentTime(),
307-
'payload' => $payload,
308-
];
309-
}
310-
311304
/**
312305
* Get the queue or return the default.
313306
*

‎src/Illuminate/Queue/Jobs/BeanstalkdJob.php

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -40,16 +40,6 @@ public function __construct(Container $container, Pheanstalk $pheanstalk, Pheans
4040
$this->pheanstalk = $pheanstalk;
4141
}
4242

43-
/**
44-
* Get the raw body string for the job.
45-
*
46-
* @return string
47-
*/
48-
public function getRawBody()
49-
{
50-
return $this->job->getData();
51-
}
52-
5343
/**
5444
* Release the job back into the queue.
5545
*
@@ -111,6 +101,16 @@ public function getJobId()
111101
return $this->job->getId();
112102
}
113103

104+
/**
105+
* Get the raw body string for the job.
106+
*
107+
* @return string
108+
*/
109+
public function getRawBody()
110+
{
111+
return $this->job->getData();
112+
}
113+
114114
/**
115115
* Get the underlying Pheanstalk instance.
116116
*

0 commit comments

Comments
 (0)