1
|
|
|
<?php |
2
|
|
|
|
3
|
|
|
namespace Dtc\QueueBundle\Tests\RabbitMQ; |
4
|
|
|
|
5
|
|
|
use Dtc\QueueBundle\Exception\UnsupportedException; |
6
|
|
|
use Dtc\QueueBundle\Model\BaseJob; |
7
|
|
|
use Dtc\QueueBundle\Model\Job; |
8
|
|
|
use Dtc\QueueBundle\Model\JobTiming; |
9
|
|
|
use Dtc\QueueBundle\Manager\JobTimingManager; |
10
|
|
|
use Dtc\QueueBundle\Model\Run; |
11
|
|
|
use Dtc\QueueBundle\Manager\RunManager; |
12
|
|
|
use Dtc\QueueBundle\RabbitMQ\JobManager; |
13
|
|
|
use Dtc\QueueBundle\Tests\FibonacciWorker; |
14
|
|
|
use Dtc\QueueBundle\Tests\Manager\AutoRetryTrait; |
15
|
|
|
use Dtc\QueueBundle\Tests\Manager\BaseJobManagerTest; |
16
|
|
|
use Dtc\QueueBundle\Tests\Manager\PriorityTestTrait; |
17
|
|
|
use Dtc\QueueBundle\Tests\Manager\RetryableTrait; |
18
|
|
|
use PhpAmqpLib\Connection\AMQPStreamConnection; |
19
|
|
|
|
20
|
|
|
/** |
21
|
|
|
* @author David |
22
|
|
|
* |
23
|
|
|
* This test requires local beanstalkd running |
24
|
|
|
*/ |
25
|
|
|
class JobManagerTest extends BaseJobManagerTest |
26
|
|
|
{ |
27
|
|
|
use PriorityTestTrait; |
28
|
|
|
use AutoRetryTrait; |
29
|
|
|
use RetryableTrait; |
30
|
|
|
public static $connection; |
31
|
|
|
|
32
|
|
|
public static function setUpBeforeClass() |
33
|
|
|
{ |
34
|
|
|
$host = getenv('RABBIT_MQ_HOST'); |
35
|
|
|
$port = 5672; |
36
|
|
|
$user = 'guest'; |
37
|
|
|
$pass = 'guest'; |
38
|
|
|
$vhost = '/'; |
39
|
|
|
$jobTimingClass = JobTiming::class; |
40
|
|
|
$runClass = Run::class; |
41
|
|
|
self::$connection = new AMQPStreamConnection($host, $port, $user, $pass, $vhost); |
42
|
|
|
|
43
|
|
|
self::$jobTimingManager = new JobTimingManager($jobTimingClass, false); |
44
|
|
|
self::$runManager = new RunManager($runClass); |
45
|
|
|
self::$jobManager = new JobManager(self::$runManager, self::$jobTimingManager, \Dtc\QueueBundle\RabbitMQ\Job::class); |
46
|
|
|
self::$jobManager->setAMQPConnection(self::$connection); |
47
|
|
|
self::$jobManager->setMaxPriority(255); |
48
|
|
|
self::$jobManager->setQueueArgs('dtc_queue', false, true, false, false); |
49
|
|
|
self::$jobManager->setExchangeArgs('dtc_queue_exchange', 'direct', false, true, false); |
50
|
|
|
$channel = self::$connection->channel(); |
51
|
|
|
$channel->queue_delete('dtc_queue'); |
52
|
|
|
$channel->close(); |
53
|
|
|
self::$worker = new FibonacciWorker(); |
54
|
|
|
self::$jobManager->setupChannel(); |
55
|
|
|
$channel = self::$jobManager->getChannel(); |
56
|
|
|
$drained = self::drainQueue($channel); |
57
|
|
|
|
58
|
|
|
if ($drained) { |
59
|
|
|
echo "\nRabbitMQ - drained $drained prior to start of test"; |
60
|
|
|
} |
61
|
|
|
parent::setUpBeforeClass(); |
62
|
|
|
} |
63
|
|
|
|
64
|
|
View Code Duplication |
public function testConstructor() |
|
|
|
|
65
|
|
|
{ |
66
|
|
|
$test = null; |
67
|
|
|
try { |
68
|
|
|
$test = new JobManager(self::$runManager, self::$jobTimingManager, Job::class); |
69
|
|
|
} catch (\Exception $exception) { |
70
|
|
|
self::fail("shouldn't get here"); |
71
|
|
|
} |
72
|
|
|
self::assertNotNull($test); |
73
|
|
|
} |
74
|
|
|
|
75
|
|
|
public function testSetupChannel() |
76
|
|
|
{ |
77
|
|
|
$jobManager = new JobManager(self::$runManager, self::$jobTimingManager, Job::class); |
78
|
|
|
$failed = false; |
79
|
|
|
try { |
80
|
|
|
$jobManager->setupChannel(); |
81
|
|
|
$failed = true; |
82
|
|
|
} catch (\Exception $exception) { |
83
|
|
|
self::assertTrue(true); |
84
|
|
|
} |
85
|
|
|
self::assertFalse($failed); |
86
|
|
|
|
87
|
|
|
try { |
88
|
|
|
$jobManager->setQueueArgs('dtc_queue', false, true, false, false); |
89
|
|
|
$failed = true; |
90
|
|
|
} catch (\Exception $exception) { |
91
|
|
|
self::assertTrue(true); |
92
|
|
|
} |
93
|
|
|
self::assertFalse($failed); |
94
|
|
|
|
95
|
|
|
$jobManager->setMaxPriority('asdf'); |
96
|
|
|
try { |
97
|
|
|
$jobManager->setQueueArgs('dtc_queue', false, true, false, false); |
98
|
|
|
$failed = true; |
99
|
|
|
} catch (\Exception $exception) { |
100
|
|
|
self::assertTrue(true); |
101
|
|
|
} |
102
|
|
|
self::assertFalse($failed); |
103
|
|
|
|
104
|
|
|
$jobManager->setMaxPriority(255); |
105
|
|
|
$jobManager->setQueueArgs('dtc_queue', false, true, false, false); |
106
|
|
|
try { |
107
|
|
|
$jobManager->setupChannel(); |
108
|
|
|
$failed = true; |
109
|
|
|
} catch (\Exception $exception) { |
110
|
|
|
self::assertTrue(true); |
111
|
|
|
} |
112
|
|
|
self::assertFalse($failed); |
113
|
|
|
$jobManager->setExchangeArgs('dtc_queue_exchange', 'direct', false, true, false); |
114
|
|
|
$jobManager->setAMQPConnection(self::$connection); |
115
|
|
|
$jobManager->setupChannel(); |
116
|
|
|
self::assertTrue(true); |
117
|
|
|
} |
118
|
|
|
|
119
|
|
|
/** |
120
|
|
|
* RabbitMQ has no ability to delete specific messages off the queue. |
121
|
|
|
*/ |
122
|
|
|
public function testDeleteJob() |
123
|
|
|
{ |
124
|
|
|
$job = new self::$jobClass(self::$worker, false, null); |
125
|
|
|
$job->fibonacci(1); |
126
|
|
|
self::assertNotNull($job->getId(), 'Job id should be generated'); |
127
|
|
|
|
128
|
|
|
try { |
129
|
|
|
self::$jobManager->deleteJob($job); |
130
|
|
|
$this->fail('expected exception'); |
131
|
|
|
} catch (\Exception $e) { |
132
|
|
|
self::assertTrue(true); |
133
|
|
|
} |
134
|
|
|
self::$jobManager->getJob(); |
135
|
|
|
} |
136
|
|
|
|
137
|
|
View Code Duplication |
public function testGetJobByWorker() |
|
|
|
|
138
|
|
|
{ |
139
|
|
|
$failed = false; |
140
|
|
|
try { |
141
|
|
|
self::$jobManager->getJob(self::$worker->getName()); |
142
|
|
|
$failed = true; |
143
|
|
|
} catch (\Exception $exception) { |
144
|
|
|
self::assertTrue(true); |
145
|
|
|
} |
146
|
|
|
self::assertFalse($failed); |
147
|
|
|
} |
148
|
|
|
|
149
|
|
View Code Duplication |
public function testExpiredJob() |
|
|
|
|
150
|
|
|
{ |
151
|
|
|
$job = new self::$jobClass(self::$worker, false, null); |
152
|
|
|
$time = time() - 1; |
153
|
|
|
$job->setExpiresAt(new \DateTime("@$time"))->fibonacci(1); |
154
|
|
|
self::assertNotNull($job->getId(), 'Job id should be generated'); |
155
|
|
|
|
156
|
|
|
$jobInQueue = self::$jobManager->getJob(); |
157
|
|
|
self::assertNull($jobInQueue, 'The job should have been dropped...'); |
158
|
|
|
|
159
|
|
|
$job = new self::$jobClass(self::$worker, false, null); |
160
|
|
|
$time = time() - 1; |
161
|
|
|
$job->setExpiresAt(new \DateTime("@$time"))->fibonacci(1); |
162
|
|
|
|
163
|
|
|
$job = new self::$jobClass(self::$worker, false, null); |
164
|
|
|
$time = time() - 1; |
165
|
|
|
$job->setExpiresAt(new \DateTime("@$time"))->fibonacci(5); |
166
|
|
|
|
167
|
|
|
$job = new self::$jobClass(self::$worker, false, null); |
168
|
|
|
$time = time() - 1; |
169
|
|
|
$job->setExpiresAt(new \DateTime("@$time"))->fibonacci(2); |
170
|
|
|
|
171
|
|
|
$job = new self::$jobClass(self::$worker, false, null); |
172
|
|
|
$job->fibonacci(1); |
173
|
|
|
$jobInQueue = self::$jobManager->getJob(); |
174
|
|
|
self::assertNotNull($jobInQueue, 'There should be a job.'); |
175
|
|
|
self::assertEquals( |
176
|
|
|
$job->getId(), |
177
|
|
|
$jobInQueue->getId(), |
178
|
|
|
'Job id returned by manager should be the same' |
179
|
|
|
); |
180
|
|
|
} |
181
|
|
|
|
182
|
|
View Code Duplication |
public function testSaveJob() |
|
|
|
|
183
|
|
|
{ |
184
|
|
|
// Make sure a job proper type |
185
|
|
|
$failed = false; |
186
|
|
|
try { |
187
|
|
|
$job = new Job(); |
188
|
|
|
self::$jobManager->save($job); |
189
|
|
|
$failed = true; |
190
|
|
|
} catch (\Exception $exception) { |
191
|
|
|
self::assertTrue(true); |
192
|
|
|
} |
193
|
|
|
self::assertFalse($failed); |
194
|
|
|
|
195
|
|
|
$job = new self::$jobClass(self::$worker, false, null); |
196
|
|
|
try { |
197
|
|
|
$job->setPriority(256)->fibonacci(1); |
198
|
|
|
$failed = true; |
199
|
|
|
} catch (\Exception $exception) { |
200
|
|
|
self::assertTrue(true); |
201
|
|
|
} |
202
|
|
|
self::assertFalse($failed); |
203
|
|
|
|
204
|
|
|
$job = new self::$jobClass(self::$worker, false, null); |
205
|
|
|
$job->setPriority(100)->fibonacci(1); |
206
|
|
|
self::assertNotNull($job->getId(), 'Job id should be generated'); |
207
|
|
|
|
208
|
|
|
$jobInQueue = self::$jobManager->getJob(); |
209
|
|
|
self::assertNotNull($jobInQueue, 'There should be a job.'); |
210
|
|
|
self::$jobManager->saveHistory($jobInQueue); |
211
|
|
|
|
212
|
|
|
$job = new self::$jobClass(self::$worker, false, null); |
213
|
|
|
$job->fibonacci(1); |
214
|
|
|
self::assertNotNull($job->getId(), 'Job id should be generated'); |
215
|
|
|
|
216
|
|
|
$failed = false; |
217
|
|
|
try { |
218
|
|
|
self::$jobManager->getJob('fibonacci'); |
219
|
|
|
$failed = true; |
220
|
|
|
} catch (\Exception $exception) { |
221
|
|
|
self::assertTrue(true); |
222
|
|
|
} |
223
|
|
|
self::assertFalse($failed); |
224
|
|
|
|
225
|
|
|
$failed = false; |
226
|
|
|
try { |
227
|
|
|
self::$jobManager->getJob(null, 'fibonacci'); |
228
|
|
|
$failed = true; |
229
|
|
|
} catch (\Exception $exception) { |
230
|
|
|
self::assertTrue(true); |
231
|
|
|
} |
232
|
|
|
self::assertFalse($failed); |
233
|
|
|
|
234
|
|
|
$jobInQueue = self::$jobManager->getJob(); |
235
|
|
|
self::assertNotNull($jobInQueue, 'There should be a job.'); |
236
|
|
|
self::assertEquals( |
237
|
|
|
$job->getId(), |
238
|
|
|
$jobInQueue->getId(), |
239
|
|
|
'Job id returned by manager should be the same' |
240
|
|
|
); |
241
|
|
|
|
242
|
|
|
$job->setStatus(BaseJob::STATUS_SUCCESS); |
243
|
|
|
$badJob = new Job(); |
244
|
|
|
$failed = false; |
245
|
|
|
try { |
246
|
|
|
self::$jobManager->saveHistory($badJob); |
247
|
|
|
$failed = true; |
248
|
|
|
} catch (\Exception $exception) { |
249
|
|
|
self::assertTrue(true); |
250
|
|
|
} |
251
|
|
|
self::assertFalse($failed); |
252
|
|
|
self::$jobManager->saveHistory($jobInQueue); |
253
|
|
|
} |
254
|
|
|
|
255
|
|
|
protected static function drainQueue($channel) |
256
|
|
|
{ |
257
|
|
|
$drained = 0; |
258
|
|
|
do { |
259
|
|
|
$message = $channel->basic_get('dtc_queue'); |
260
|
|
|
if ($message) { |
261
|
|
|
$channel->basic_ack($message->delivery_info['delivery_tag']); |
262
|
|
|
++$drained; |
263
|
|
|
} |
264
|
|
|
} while ($message); |
265
|
|
|
|
266
|
|
|
return $drained; |
267
|
|
|
} |
268
|
|
|
|
269
|
|
|
public function testGetWaitingJobCount() |
270
|
|
|
{ |
271
|
|
|
/** @var JobManager $jobManager */ |
272
|
|
|
$jobManager = self::$jobManager; |
273
|
|
|
self::drainQueue($jobManager->getChannel()); |
274
|
|
|
|
275
|
|
|
$job = new self::$jobClass(self::$worker, false, null); |
276
|
|
|
$job->fibonacci(1); |
277
|
|
|
self::assertNotNull($job->getId(), 'Job id should be generated'); |
278
|
|
|
|
279
|
|
|
self::assertEquals(1, self::getWaitingJobCount(2)); |
280
|
|
|
|
281
|
|
|
$failure = false; |
282
|
|
|
try { |
283
|
|
|
$jobManager->getWaitingJobCount('fibonacci'); |
284
|
|
|
$failure = true; |
285
|
|
|
} catch (UnsupportedException $exception) { |
286
|
|
|
self::assertTrue(true); |
287
|
|
|
} |
288
|
|
|
self::assertFalse($failure); |
289
|
|
|
|
290
|
|
|
$failure = false; |
291
|
|
|
try { |
292
|
|
|
$jobManager->getWaitingJobCount(null, 'fibonacci'); |
293
|
|
|
$failure = true; |
294
|
|
|
} catch (UnsupportedException $exception) { |
295
|
|
|
self::assertTrue(true); |
296
|
|
|
} |
297
|
|
|
self::assertFalse($failure); |
298
|
|
|
|
299
|
|
|
$failure = false; |
300
|
|
|
try { |
301
|
|
|
$jobManager->getWaitingJobCount('fibonacci', 'fibonacci'); |
302
|
|
|
$failure = true; |
303
|
|
|
} catch (UnsupportedException $exception) { |
304
|
|
|
self::assertTrue(true); |
305
|
|
|
} |
306
|
|
|
self::assertFalse($failure); |
307
|
|
|
|
308
|
|
|
$job = new self::$jobClass(self::$worker, false, null); |
309
|
|
|
$job->fibonacci(1); |
310
|
|
|
self::assertNotNull($job->getId(), 'Job id should be generated'); |
311
|
|
|
|
312
|
|
|
self::assertEquals(2, self::getWaitingJobCount(2)); |
313
|
|
|
} |
314
|
|
|
} |
315
|
|
|
|
Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation.
You can also find more detailed suggestions in the “Code” section of your repository.