-
Notifications
You must be signed in to change notification settings - Fork 762
/
Job.php
executable file
·289 lines (254 loc) · 6.71 KB
/
Job.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
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
<?php
/**
* Resque job.
*
* @package Resque/Job
* @author Chris Boulton <chris@bigcommerce.com>
* @license http://www.opensource.org/licenses/mit-license.php
*/
class Resque_Job
{
/**
* @var string The name of the queue that this job belongs to.
*/
public $queue;
/**
* @var Resque_Worker Instance of the Resque worker running this job.
*/
public $worker;
/**
* @var array Array containing details of the job.
*/
public $payload;
/**
* @var object|Resque_JobInterface Instance of the class performing work for this job.
*/
private $instance;
/**
* @var Resque_Job_FactoryInterface
*/
private $jobFactory;
/**
* Instantiate a new instance of a job.
*
* @param string $queue The queue that the job belongs to.
* @param array $payload array containing details of the job.
*/
public function __construct($queue, $payload)
{
$this->queue = $queue;
$this->payload = $payload;
}
/**
* Create a new job and save it to the specified queue.
*
* @param string $queue The name of the queue to place the job in.
* @param string $class The name of the class that contains the code to execute the job.
* @param array $args Any optional arguments that should be passed when the job is executed.
* @param boolean $monitor Set to true to be able to monitor the status of a job.
* @param string $id Unique identifier for tracking the job. Generated if not supplied.
*
* @return string
* @throws \InvalidArgumentException
*/
public static function create($queue, $class, $args = null, $monitor = false, $id = null)
{
if (is_null($id)) {
$id = Resque::generateJobId();
}
if($args !== null && !is_array($args)) {
throw new InvalidArgumentException(
'Supplied $args must be an array.'
);
}
Resque::push($queue, array(
'class' => $class,
'args' => array($args),
'id' => $id,
'queue_time' => microtime(true),
));
if($monitor) {
Resque_Job_Status::create($id);
}
return $id;
}
/**
* Find the next available job from the specified queue and return an
* instance of Resque_Job for it.
*
* @param string $queue The name of the queue to check for a job in.
* @return false|object Null when there aren't any waiting jobs, instance of Resque_Job when a job was found.
*/
public static function reserve($queue)
{
$payload = Resque::pop($queue);
if(!is_array($payload)) {
return false;
}
return new Resque_Job($queue, $payload);
}
/**
* Find the next available job from the specified queues using blocking list pop
* and return an instance of Resque_Job for it.
*
* @param array $queues
* @param int $timeout
* @return false|object Null when there aren't any waiting jobs, instance of Resque_Job when a job was found.
*/
public static function reserveBlocking(array $queues, $timeout = null)
{
$item = Resque::blpop($queues, $timeout);
if(!is_array($item)) {
return false;
}
return new Resque_Job($item['queue'], $item['payload']);
}
/**
* Update the status of the current job.
*
* @param int $status Status constant from Resque_Job_Status indicating the current status of a job.
*/
public function updateStatus($status)
{
if(empty($this->payload['id'])) {
return;
}
$statusInstance = new Resque_Job_Status($this->payload['id']);
$statusInstance->update($status);
}
/**
* Return the status of the current job.
*
* @return int The status of the job as one of the Resque_Job_Status constants.
*/
public function getStatus()
{
$status = new Resque_Job_Status($this->payload['id']);
return $status->get();
}
/**
* Get the arguments supplied to this job.
*
* @return array Array of arguments.
*/
public function getArguments()
{
if (!isset($this->payload['args'])) {
return array();
}
return $this->payload['args'][0];
}
/**
* Get the instantiated object for this job that will be performing work.
* @return Resque_JobInterface Instance of the object that this job belongs to.
* @throws Resque_Exception
*/
public function getInstance()
{
if (!is_null($this->instance)) {
return $this->instance;
}
$this->instance = $this->getJobFactory()->create($this->payload['class'], $this->getArguments(), $this->queue);
$this->instance->job = $this;
return $this->instance;
}
/**
* Actually execute a job by calling the perform method on the class
* associated with the job with the supplied arguments.
*
* @return bool
* @throws Resque_Exception When the job's class could not be found or it does not contain a perform method.
*/
public function perform()
{
try {
Resque_Event::trigger('beforePerform', $this);
$instance = $this->getInstance();
if(method_exists($instance, 'setUp')) {
$instance->setUp();
}
$instance->perform();
if(method_exists($instance, 'tearDown')) {
$instance->tearDown();
}
Resque_Event::trigger('afterPerform', $this);
}
// beforePerform/setUp have said don't perform this job. Return.
catch(Resque_Job_DontPerform $e) {
return false;
}
return true;
}
/**
* Mark the current job as having failed.
*
* @param $exception
*/
public function fail($exception)
{
Resque_Event::trigger('onFailure', array(
'exception' => $exception,
'job' => $this,
));
$this->updateStatus(Resque_Job_Status::STATUS_FAILED);
Resque_Failure::create(
$this->payload,
$exception,
$this->worker,
$this->queue
);
Resque_Stat::incr('failed');
Resque_Stat::incr('failed:' . $this->worker);
}
/**
* Re-queue the current job.
* @return string
*/
public function recreate()
{
$status = new Resque_Job_Status($this->payload['id']);
$monitor = false;
if($status->isTracking()) {
$monitor = true;
}
return self::create($this->queue, $this->payload['class'], $this->getArguments(), $monitor);
}
/**
* Generate a string representation used to describe the current job.
*
* @return string The string representation of the job.
*/
public function __toString()
{
$name = array(
'Job{' . $this->queue .'}'
);
if(!empty($this->payload['id'])) {
$name[] = 'ID: ' . $this->payload['id'];
}
$name[] = $this->payload['class'];
if(!empty($this->payload['args'])) {
$name[] = json_encode($this->payload['args']);
}
return '(' . implode(' | ', $name) . ')';
}
/**
* @param Resque_Job_FactoryInterface $jobFactory
* @return Resque_Job
*/
public function setJobFactory(Resque_Job_FactoryInterface $jobFactory)
{
$this->jobFactory = $jobFactory;
return $this;
}
/**
* @return Resque_Job_FactoryInterface
*/
public function getJobFactory()
{
if ($this->jobFactory === null) {
$this->jobFactory = new Resque_Job_Factory();
}
return $this->jobFactory;
}
}