@@ -70,7 +70,7 @@ discard block |
||
| 70 | 70 | ]; |
| 71 | 71 | |
| 72 | 72 | try { |
| 73 | - $this->predis->transaction($options, function ($tx) use ($key, &$element) { |
|
| 73 | + $this->predis->transaction($options, function($tx) use ($key, &$element) { |
|
| 74 | 74 | @list($element) = $tx->zrange($key, 0, 0); |
| 75 | 75 | |
| 76 | 76 | if (isset($element)) { |
@@ -95,7 +95,7 @@ discard block |
||
| 95 | 95 | ]; |
| 96 | 96 | |
| 97 | 97 | try { |
| 98 | - $this->predis->transaction($options, function ($tx) use ($key, $max, &$element) { |
|
| 98 | + $this->predis->transaction($options, function($tx) use ($key, $max, &$element) { |
|
| 99 | 99 | @list($element) = $tx->zrangebyscore($key, 0, $max, ['LIMIT' => [0, 1]]); |
| 100 | 100 | |
| 101 | 101 | if (isset($element)) { |
@@ -522,13 +522,13 @@ |
||
| 522 | 522 | |
| 523 | 523 | return $node; |
| 524 | 524 | })->end() |
| 525 | - ->validate()->ifTrue(function ($node) { |
|
| 526 | - if (isset($node['ssl_options']) && !$node['ssl']) { |
|
| 527 | - return true; |
|
| 528 | - } |
|
| 525 | + ->validate()->ifTrue(function ($node) { |
|
| 526 | + if (isset($node['ssl_options']) && !$node['ssl']) { |
|
| 527 | + return true; |
|
| 528 | + } |
|
| 529 | 529 | |
| 530 | - return false; |
|
| 531 | - })->thenInvalid('ssl must be true in order to set ssl_options')->end() |
|
| 530 | + return false; |
|
| 531 | + })->thenInvalid('ssl must be true in order to set ssl_options')->end() |
|
| 532 | 532 | ->end(); |
| 533 | 533 | |
| 534 | 534 | return $rootNode; |
@@ -272,7 +272,7 @@ discard block |
||
| 272 | 272 | ->scalarNode('alias') |
| 273 | 273 | ->defaultNull()->end() |
| 274 | 274 | ->end() |
| 275 | - ->validate()->ifTrue(function ($node) { |
|
| 275 | + ->validate()->ifTrue(function($node) { |
|
| 276 | 276 | if (isset($node['type']) && !isset($node['alias'])) { |
| 277 | 277 | return true; |
| 278 | 278 | } |
@@ -288,7 +288,7 @@ discard block |
||
| 288 | 288 | ->scalarNode('dsn')->defaultNull()->end() |
| 289 | 289 | ->append($this->addPredisArgs()) |
| 290 | 290 | ->end() |
| 291 | - ->validate()->ifTrue(function ($node) { |
|
| 291 | + ->validate()->ifTrue(function($node) { |
|
| 292 | 292 | if (isset($node['dsn']) && (isset($node['connection_parameters']['host']) || isset($node['connection_parameters']['port']))) { |
| 293 | 293 | return true; |
| 294 | 294 | } |
@@ -298,7 +298,7 @@ discard block |
||
| 298 | 298 | ->end() |
| 299 | 299 | ->append($this->addPhpRedisArgs()) |
| 300 | 300 | ->end() |
| 301 | - ->validate()->ifTrue(function ($node) { |
|
| 301 | + ->validate()->ifTrue(function($node) { |
|
| 302 | 302 | if ((isset($node['predis']['dsn']) || isset($node['predis']['connection_parameters']['host'])) && |
| 303 | 303 | (isset($node['snc_redis']['type']) || isset($node['phpredis']['host']))) { |
| 304 | 304 | return true; |
@@ -334,7 +334,7 @@ discard block |
||
| 334 | 334 | ->floatNode('read_timeout')->defaultValue(0)->end() |
| 335 | 335 | ->scalarNode('auth')->end() |
| 336 | 336 | ->end() |
| 337 | - ->validate()->ifTrue(function ($node) { |
|
| 337 | + ->validate()->ifTrue(function($node) { |
|
| 338 | 338 | if (!empty($node) && !isset($node['host'])) { |
| 339 | 339 | return true; |
| 340 | 340 | } |
@@ -367,7 +367,7 @@ discard block |
||
| 367 | 367 | ->booleanNode('iterable_multibulk')->defaultFalse()->end() |
| 368 | 368 | ->booleanNode('throw_errors')->defaultTrue()->end() |
| 369 | 369 | ->end() |
| 370 | - ->validate()->ifTrue(function ($node) { |
|
| 370 | + ->validate()->ifTrue(function($node) { |
|
| 371 | 371 | if (isset($node['host']) && !isset($node['port'])) { |
| 372 | 372 | return true; |
| 373 | 373 | } |
@@ -407,7 +407,7 @@ discard block |
||
| 407 | 407 | $rootNode |
| 408 | 408 | ->prototype('variable')->end() |
| 409 | 409 | ->validate() |
| 410 | - ->ifTrue(function ($node) { |
|
| 410 | + ->ifTrue(function($node) { |
|
| 411 | 411 | if (!is_array($node)) { |
| 412 | 412 | return true; |
| 413 | 413 | } |
@@ -484,7 +484,7 @@ discard block |
||
| 484 | 484 | ->append($this->addRabbitMqArgs()) |
| 485 | 485 | ->append($this->addRabbitMqExchange()) |
| 486 | 486 | ->end() |
| 487 | - ->validate()->always(function ($node) { |
|
| 487 | + ->validate()->always(function($node) { |
|
| 488 | 488 | if (empty($node['ssl_options'])) { |
| 489 | 489 | unset($node['ssl_options']); |
| 490 | 490 | } |
@@ -494,7 +494,7 @@ discard block |
||
| 494 | 494 | |
| 495 | 495 | return $node; |
| 496 | 496 | })->end() |
| 497 | - ->validate()->ifTrue(function ($node) { |
|
| 497 | + ->validate()->ifTrue(function($node) { |
|
| 498 | 498 | if (isset($node['ssl_options']) && !$node['ssl']) { |
| 499 | 499 | return true; |
| 500 | 500 | } |
@@ -286,7 +286,7 @@ |
||
| 286 | 286 | self::assertEquals(1, $count); |
| 287 | 287 | $allCount = $this->runCountQuery($jobManager->getJobClass()); |
| 288 | 288 | $counter = 0; |
| 289 | - $countJobs = function ($count) use (&$counter) { |
|
| 289 | + $countJobs = function($count) use (&$counter) { |
|
| 290 | 290 | $counter += $count; |
| 291 | 291 | }; |
| 292 | 292 | $jobManager->archiveAllJobs(null, null, $countJobs); |
@@ -60,13 +60,13 @@ |
||
| 60 | 60 | $methodName = $request->get('method'); |
| 61 | 61 | |
| 62 | 62 | $jobManager = $this->get('dtc_queue.manager.job'); |
| 63 | - $callback = function ($count) { |
|
| 63 | + $callback = function($count) { |
|
| 64 | 64 | echo json_encode(['count' => $count]); |
| 65 | 65 | echo "\n"; |
| 66 | 66 | flush(); |
| 67 | 67 | }; |
| 68 | 68 | |
| 69 | - $streamingResponse = new StreamedResponse(function () use ($jobManager, $callback, $workerName, $methodName) { |
|
| 69 | + $streamingResponse = new StreamedResponse(function() use ($jobManager, $callback, $workerName, $methodName) { |
|
| 70 | 70 | $total = $jobManager->countLiveJobs($workerName, $methodName); |
| 71 | 71 | echo json_encode(['total' => $total]); |
| 72 | 72 | echo "\n"; |
@@ -26,6 +26,9 @@ discard block |
||
| 26 | 26 | protected $hostname; |
| 27 | 27 | protected $pid; |
| 28 | 28 | |
| 29 | + /** |
|
| 30 | + * @param string $cacheKeyPrefix |
|
| 31 | + */ |
|
| 29 | 32 | public function __construct(RunManager $runManager, JobTimingManager $jobTimingManager, $jobClass, $cacheKeyPrefix) |
| 30 | 33 | { |
| 31 | 34 | $this->cacheKeyPrefix = $cacheKeyPrefix; |
@@ -45,6 +48,9 @@ discard block |
||
| 45 | 48 | return $this->cacheKeyPrefix.'_job_'.$jobId; |
| 46 | 49 | } |
| 47 | 50 | |
| 51 | + /** |
|
| 52 | + * @param string $jobCrc |
|
| 53 | + */ |
|
| 48 | 54 | protected function getJobCrcHashKey($jobCrc) |
| 49 | 55 | { |
| 50 | 56 | return $this->cacheKeyPrefix.'_job_crc_'.$jobCrc; |
@@ -102,6 +108,10 @@ discard block |
||
| 102 | 108 | return null; |
| 103 | 109 | } |
| 104 | 110 | |
| 111 | + /** |
|
| 112 | + * @param string $foundJobCacheKey |
|
| 113 | + * @param boolean $foundJobMessage |
|
| 114 | + */ |
|
| 105 | 115 | protected function batchFoundJob(\Dtc\QueueBundle\Redis\Job $job, $foundJobCacheKey, $foundJobMessage) |
| 106 | 116 | { |
| 107 | 117 | $when = $job->getWhenUs(); |
@@ -127,6 +137,10 @@ discard block |
||
| 127 | 137 | return $this->finishBatchFoundJob($foundJob, $foundJobCacheKey, $crcCacheKey, $newFoundWhen, $newFoundPriority); |
| 128 | 138 | } |
| 129 | 139 | |
| 140 | + /** |
|
| 141 | + * @param string $crcCacheKey |
|
| 142 | + * @param integer|null $newFoundPriority |
|
| 143 | + */ |
|
| 130 | 144 | protected function finishBatchFoundJob(Job $foundJob, $foundJobCacheKey, $crcCacheKey, $newFoundWhen, $newFoundPriority) |
| 131 | 145 | { |
| 132 | 146 | // Now how do we adjust this job's priority or time? |
@@ -146,6 +160,9 @@ discard block |
||
| 146 | 160 | return $this->addFoundJob($adjust, $foundJob, $foundJobCacheKey, $crcCacheKey); |
| 147 | 161 | } |
| 148 | 162 | |
| 163 | + /** |
|
| 164 | + * @param boolean $adjust |
|
| 165 | + */ |
|
| 149 | 166 | protected function addFoundJob($adjust, Job $foundJob, $foundJobCacheKey, $crcCacheKey) |
| 150 | 167 | { |
| 151 | 168 | $whenQueue = $this->getWhenQueueCacheKey(); |
@@ -163,6 +180,9 @@ discard block |
||
| 163 | 180 | return $result ?: false; |
| 164 | 181 | } |
| 165 | 182 | |
| 183 | + /** |
|
| 184 | + * @param string $queue |
|
| 185 | + */ |
|
| 166 | 186 | private function adjustJob($adjust, $queue, Job $foundJob, $foundJobCacheKey, $crcCacheKey, $zScore) |
| 167 | 187 | { |
| 168 | 188 | if ($adjust && $this->redis->zRem($queue, $foundJob->getId()) > 0) { |
@@ -183,7 +203,7 @@ discard block |
||
| 183 | 203 | /** |
| 184 | 204 | * @param \Dtc\QueueBundle\Model\Job $job |
| 185 | 205 | * |
| 186 | - * @return \Dtc\QueueBundle\Model\Job |
|
| 206 | + * @return null|Job |
|
| 187 | 207 | * |
| 188 | 208 | * @throws ClassNotSubclassException |
| 189 | 209 | */ |
@@ -302,6 +322,9 @@ discard block |
||
| 302 | 322 | } |
| 303 | 323 | } |
| 304 | 324 | |
| 325 | + /** |
|
| 326 | + * @param string $workerName |
|
| 327 | + */ |
|
| 305 | 328 | protected function verifyGetJobArgs($workerName = null, $methodName = null, $prioritize = true) |
| 306 | 329 | { |
| 307 | 330 | if (null !== $workerName || null !== $methodName || (null !== $this->maxPriority && true !== $prioritize)) { |