diff --git a/src/Illuminate/Queue/DatabaseQueue.php b/src/Illuminate/Queue/DatabaseQueue.php index 931651fa55d1..3f671abd7e9e 100644 --- a/src/Illuminate/Queue/DatabaseQueue.php +++ b/src/Illuminate/Queue/DatabaseQueue.php @@ -44,6 +44,8 @@ class DatabaseQueue extends Queue implements QueueContract, ClearableQueue /** * The expiration time of a job. * + * @deprecated No longer necessary, reservations are now based on the job timeout. + * * @var int|null */ protected $retryAfter = 60; @@ -597,8 +599,18 @@ protected function marshalJob($queue, $job) */ protected function markJobAsReserved($job) { + $timeout = json_decode($job->payload, true)['timeout'] ?? null; + + if (! is_numeric($timeout)) { + $timeout = $this->workerTimeout; + } elseif ($timeout <= 0) { + $timeout = 9999999999; + } + + $reservationOffset = (int) $timeout + 10 - $this->retryAfter; + $this->database->table($this->table)->where('id', $job->id)->update([ - 'reserved_at' => $job->touch(), + 'reserved_at' => $job->touch($reservationOffset), 'attempts' => $job->increment(), ]); diff --git a/src/Illuminate/Queue/FailoverQueue.php b/src/Illuminate/Queue/FailoverQueue.php index af3fb1a4afc6..2aceceae1758 100644 --- a/src/Illuminate/Queue/FailoverQueue.php +++ b/src/Illuminate/Queue/FailoverQueue.php @@ -250,6 +250,25 @@ public function pop($queue = null) return $this->manager->connection($this->connections[0])->pop($queue); } + /** + * Set the job timeout of the worker processing the queue. + * + * @param int|string|null $timeout + * @return $this + */ + public function setWorkerTimeout($timeout) + { + foreach ($this->connections as $connection) { + $connection = $this->manager->connection($connection); + + if (method_exists($connection, 'setWorkerTimeout')) { + $connection->setWorkerTimeout($timeout); + } + } + + return parent::setWorkerTimeout($timeout); + } + /** * Attempt the given method on all connections. * diff --git a/src/Illuminate/Queue/Jobs/DatabaseJobRecord.php b/src/Illuminate/Queue/Jobs/DatabaseJobRecord.php index 207f2b529c82..d8ac84787304 100644 --- a/src/Illuminate/Queue/Jobs/DatabaseJobRecord.php +++ b/src/Illuminate/Queue/Jobs/DatabaseJobRecord.php @@ -40,11 +40,12 @@ public function increment() /** * Update the "reserved at" timestamp of the job. * + * @param int $offset * @return int */ - public function touch() + public function touch($offset = 0) { - $this->record->reserved_at = $this->currentTime(); + $this->record->reserved_at = min($this->currentTime() + $offset, 2147483647); return $this->record->reserved_at; } diff --git a/src/Illuminate/Queue/Listener.php b/src/Illuminate/Queue/Listener.php index fc56a569ebaa..1ea034cee6a7 100755 --- a/src/Illuminate/Queue/Listener.php +++ b/src/Illuminate/Queue/Listener.php @@ -161,6 +161,7 @@ protected function createCommand($connection, $queue, ListenerOptions $options) "--backoff={$options->backoff}", "--memory={$options->memory}", "--sleep={$options->sleep}", + "--timeout={$options->timeout}", "--tries={$options->maxTries}", $options->force ? '--force' : null, ], function ($value) { diff --git a/src/Illuminate/Queue/LuaScripts.php b/src/Illuminate/Queue/LuaScripts.php index f95e9afd68cf..bfd088102cc4 100644 --- a/src/Illuminate/Queue/LuaScripts.php +++ b/src/Illuminate/Queue/LuaScripts.php @@ -62,7 +62,8 @@ public static function later() * KEYS[1] - The queue to pop jobs from, for example: queues:foo * KEYS[2] - The queue to place reserved jobs on, for example: queues:foo:reserved * KEYS[3] - The notify queue - * ARGV[1] - The time at which the reserved job will expire + * ARGV[1] - The current UNIX timestamp + * ARGV[2] - The timeout of the worker popping the job * * @return string */ @@ -77,8 +78,17 @@ public static function pop() -- Increment the attempt count and place job on the reserved queue... reserved = cjson.decode(job) reserved['attempts'] = reserved['attempts'] + 1 + + local timeout = tonumber(reserved['timeout']) or tonumber(ARGV[2]) + + if(timeout <= 0) then + timeout = 9999999999 + end + + local expiration = ARGV[1] + timeout + 10 + reserved = cjson.encode(reserved) - redis.call('zadd', KEYS[2], ARGV[1], reserved) + redis.call('zadd', KEYS[2], expiration, reserved) redis.call('lpop', KEYS[3]) end diff --git a/src/Illuminate/Queue/Queue.php b/src/Illuminate/Queue/Queue.php index 910588856d4a..5e55d4049089 100755 --- a/src/Illuminate/Queue/Queue.php +++ b/src/Illuminate/Queue/Queue.php @@ -62,6 +62,13 @@ abstract class Queue */ protected $dispatchAfterCommit; + /** + * The job timeout set on the worker processing the queue. + * + * @var int + */ + protected $workerTimeout = 60; + /** * The create payload callbacks. * @@ -530,6 +537,23 @@ public function setConnectionName($name) return $this; } + /** + * Set the job timeout of the worker processing the queue. + * + * @param int|string|null $timeout + * @return $this + */ + public function setWorkerTimeout($timeout) + { + $this->workerTimeout = match (true) { + ! is_numeric($timeout) => 60, + $timeout <= 0 => 9999999999, + default => (int) $timeout, + }; + + return $this; + } + /** * Get the queue configuration array. * diff --git a/src/Illuminate/Queue/RedisQueue.php b/src/Illuminate/Queue/RedisQueue.php index 2f5615eb876c..b90c85791415 100644 --- a/src/Illuminate/Queue/RedisQueue.php +++ b/src/Illuminate/Queue/RedisQueue.php @@ -42,6 +42,8 @@ class RedisQueue extends Queue implements QueueContract, ClearableQueue /** * The expiration time of a job. * + * @deprecated No longer necessary, reservations are now based on the job timeout. + * * @var int|null */ protected $retryAfter = 60; @@ -491,7 +493,8 @@ protected function retrieveNextJob($queue, $block = true) { $nextJob = $this->getConnection()->eval( LuaScripts::pop(), 3, $queue, $queue.':reserved', $queue.':notify', - $this->availableAt($this->retryAfter) + $this->currentTime(), + $this->workerTimeout ); if (empty($nextJob)) { diff --git a/src/Illuminate/Queue/Worker.php b/src/Illuminate/Queue/Worker.php index 8b3d4b72b0b2..e90e2ec08ce2 100644 --- a/src/Illuminate/Queue/Worker.php +++ b/src/Illuminate/Queue/Worker.php @@ -229,6 +229,8 @@ public function daemon($connectionName, $queue, WorkerOptions $options) $this->raiseWorkerStartingEvent($connectionName, $queue, $options); + $this->shareWorkerTimeout($connectionName, $options); + while (true) { // Before reserving any jobs, we will make sure this queue is not paused and // if it is we will just pause this worker for a given amount of time and @@ -417,6 +419,8 @@ protected function stopIfNecessary(WorkerOptions $options, $lastRestart, $startT */ public function runNextJob($connectionName, $queue, WorkerOptions $options) { + $this->shareWorkerTimeout($connectionName, $options); + $job = $this->getNextJob( $this->manager->connection($connectionName), $queue ); @@ -431,6 +435,22 @@ public function runNextJob($connectionName, $queue, WorkerOptions $options) $this->sleep($options->sleep); } + /** + * Share the worker's job timeout with the given queue connection. + * + * @param string $connectionName + * @param \Illuminate\Queue\WorkerOptions $options + * @return void + */ + protected function shareWorkerTimeout($connectionName, WorkerOptions $options) + { + $connection = $this->manager->connection($connectionName); + + if (method_exists($connection, 'setWorkerTimeout')) { + $connection->setWorkerTimeout($options->timeout); + } + } + /** * Get the next job from the queue connection. * diff --git a/tests/Integration/Queue/RedisQueueTest.php b/tests/Integration/Queue/RedisQueueTest.php index 131bfc04dc05..272dbe8f00cd 100644 --- a/tests/Integration/Queue/RedisQueueTest.php +++ b/tests/Integration/Queue/RedisQueueTest.php @@ -171,11 +171,81 @@ public function testPopProperlyPopsJobOffOfRedis($driver) $result = $this->redis[$driver]->connection()->zrangebyscore("$redisKey:reserved", -INF, INF, ['withscores' => true]); $reservedJob = array_keys($result)[0]; $score = (int) $result[$reservedJob]; - $this->assertLessThanOrEqual($score, $before + 60); - $this->assertGreaterThanOrEqual($score, $after + 60); + $this->assertLessThanOrEqual($score, $before + 70); + $this->assertGreaterThanOrEqual($score, $after + 70); $this->assertEquals($job, unserialize(json_decode($reservedJob)->data->command)); } + /** + * @param string $driver + */ + #[DataProvider('redisDriverProvider')] + public function testPopReservesJobsBasedOnTheJobTimeout($driver) + { + $default = config('queue.connections.redis.queue', 'default'); + $this->queue = new RedisQueue($this->redis[$driver], $default, null, retryAfter: 12345); + $this->queue->setContainer($this->container = Mockery::spy(Container::class)); + $redisKey = $this->getQueueRedisKey($default); + + $getJobExpirationTimestamp = function () use ($driver, $redisKey) { + $this->queue->pop(); + + $result = $this->redis[$driver]->connection()->zrangebyscore("$redisKey:reserved", -INF, INF, ['withscores' => true]); + + $this->queue->clear(); + + return (int) $result[array_keys($result)[0]]; + }; + + Carbon::setTestNow($time = Carbon::now()); + + // Timeout defined on the job is used + $job = new RedisQueueIntegrationTestJob(123); + $job->timeout = 25; + $this->queue->push($job); + $this->assertSame($time->getTimestamp() + 25 + 10, $getJobExpirationTimestamp()); + + // No timeout on the job, worker timeout unknown: the default worker timeout is used + $this->queue->push(new RedisQueueIntegrationTestJob(123)); + $this->assertSame($time->getTimestamp() + 60 + 10, $getJobExpirationTimestamp()); + + // Timeout of 0 on the job means it is never killed: reserved effectively forever + $job = new RedisQueueIntegrationTestJob(123); + $job->timeout = 0; + $this->queue->push($job); + $this->assertSame($time->getTimestamp() + 9999999999 + 10, $getJobExpirationTimestamp()); + + // A negative timeout on the job is treated the same as zero + $job = new RedisQueueIntegrationTestJob(123); + $job->timeout = -5; + $this->queue->push($job); + $this->assertSame($time->getTimestamp() + 9999999999 + 10, $getJobExpirationTimestamp()); + + // No timeout on the job, the queue worker's timeout is used + $this->queue->setWorkerTimeout(15); + $this->queue->push(new RedisQueueIntegrationTestJob(123)); + $this->assertSame($time->getTimestamp() + 15 + 10, $getJobExpirationTimestamp()); + + // A worker timeout of 0 means jobs are never killed: reserved effectively forever + $this->queue->setWorkerTimeout(0); + $this->queue->push(new RedisQueueIntegrationTestJob(123)); + $this->assertSame($time->getTimestamp() + 9999999999 + 10, $getJobExpirationTimestamp()); + + // A null worker timeout resets the queue to the default worker timeout + $this->queue->setWorkerTimeout(null); + $this->queue->push(new RedisQueueIntegrationTestJob(123)); + $this->assertSame($time->getTimestamp() + 60 + 10, $getJobExpirationTimestamp()); + + // Timeout on the job takes precedence over the queue worker timeout + $this->queue->setWorkerTimeout(15); + $job = new RedisQueueIntegrationTestJob(123); + $job->timeout = 25; + $this->queue->push($job); + $this->assertSame($time->getTimestamp() + 25 + 10, $getJobExpirationTimestamp()); + + Carbon::setTestNow(); + } + /** * @param string $driver */ @@ -199,8 +269,8 @@ public function testPopProperlyPopsDelayedJobOffOfRedis($driver) $result = $this->redis[$driver]->connection()->zrangebyscore("$redisKey:reserved", -INF, INF, ['withscores' => true]); $reservedJob = array_keys($result)[0]; $score = (int) $result[$reservedJob]; - $this->assertLessThanOrEqual($score, $before + 60); - $this->assertGreaterThanOrEqual($score, $after + 60); + $this->assertLessThanOrEqual($score, $before + 70); + $this->assertGreaterThanOrEqual($score, $after + 70); $this->assertEquals($job, unserialize(json_decode($reservedJob)->data->command)); } @@ -230,8 +300,8 @@ public function testPopPopsDelayedJobOffOfRedisWhenExpireNull($driver) $result = $this->redis[$driver]->connection()->zrangebyscore("$redisKey:reserved", -INF, INF, ['withscores' => true]); $reservedJob = array_keys($result)[0]; $score = (int) $result[$reservedJob]; - $this->assertLessThanOrEqual($score, $before); - $this->assertGreaterThanOrEqual($score, $after); + $this->assertLessThanOrEqual($score, $before + 70); + $this->assertGreaterThanOrEqual($score, $after + 70); $this->assertEquals($job, unserialize(json_decode($reservedJob)->data->command)); } @@ -329,11 +399,11 @@ public function testNotExpireJobsWhenExpireNull($driver) $score = (int) $score; if ($command->i == 10) { - $this->assertLessThanOrEqual($score, $before); - $this->assertGreaterThanOrEqual($score, $after); + $this->assertLessThanOrEqual($score, $before + 70); + $this->assertGreaterThanOrEqual($score, $after + 70); } else { - $this->assertLessThanOrEqual($score, $beforeFailPop); - $this->assertGreaterThanOrEqual($score, $afterFailPop); + $this->assertLessThanOrEqual($score, $beforeFailPop + 70); + $this->assertGreaterThanOrEqual($score, $afterFailPop + 70); } } } @@ -363,8 +433,8 @@ public function testExpireJobsWhenExpireSet($driver) $result = $this->redis[$driver]->connection()->zrangebyscore("$redisKey:reserved", -INF, INF, ['withscores' => true]); $reservedJob = array_keys($result)[0]; $score = (int) $result[$reservedJob]; - $this->assertLessThanOrEqual($score, $before + 30); - $this->assertGreaterThanOrEqual($score, $after + 30); + $this->assertLessThanOrEqual($score, $before + 70); + $this->assertGreaterThanOrEqual($score, $after + 70); $this->assertEquals($job, unserialize(json_decode($reservedJob)->data->command)); } @@ -768,6 +838,8 @@ class RedisQueueIntegrationTestJob { public $i; + public $timeout; + public function __construct($i) { $this->i = $i; diff --git a/tests/Queue/FailoverQueueTest.php b/tests/Queue/FailoverQueueTest.php index f2aca6388a30..cf79fd8f1d11 100644 --- a/tests/Queue/FailoverQueueTest.php +++ b/tests/Queue/FailoverQueueTest.php @@ -57,6 +57,24 @@ public function test_bulk_respects_job_delays() $failover->bulk([new FailoverJobWithDelayAttribute, new FailoverJobWithDelayProperty, 'regular-job']); } + + public function test_set_worker_timeout_is_forwarded_to_underlying_connections() + { + $queue = Mockery::mock(QueueManager::class); + $failover = new FailoverQueue($queue, Mockery::mock(Dispatcher::class), [ + 'redis', + 'sync', + ]); + + $redis = new FailoverQueueTestFakeConnection; + $queue->expects('connection')->with('redis')->andReturn($redis); + + // Connections without the method are skipped instead of failing + $queue->expects('connection')->with('sync')->andReturn(new \stdClass); + + $this->assertSame($failover, $failover->setWorkerTimeout(90)); + $this->assertSame(90, $redis->workerTimeout); + } } #[Delay(15)] @@ -68,3 +86,15 @@ class FailoverJobWithDelayProperty { public $delay = 30; } + +class FailoverQueueTestFakeConnection +{ + public $workerTimeout = false; + + public function setWorkerTimeout($timeout) + { + $this->workerTimeout = $timeout; + + return $this; + } +} diff --git a/tests/Queue/QueueDatabaseQueueIntegrationTest.php b/tests/Queue/QueueDatabaseQueueIntegrationTest.php index fd5b8e7d8788..94e265bf6809 100644 --- a/tests/Queue/QueueDatabaseQueueIntegrationTest.php +++ b/tests/Queue/QueueDatabaseQueueIntegrationTest.php @@ -154,6 +154,64 @@ public function testPoppedJobsIncrementAttempts() $this->assertEquals(1, $popped_job->attempts(), 'The "attempts" attribute of the Job object was not updated by pop!'); } + public function testPoppedJobsAreReservedBasedOnTheJobTimeout() + { + Carbon::setTestNow($time = Carbon::now()); + + $retryAfter = 3600; + + $queue = new DatabaseQueue($this->connection(), $this->table, 'default', $retryAfter); + $queue->setContainer($this->container); + + $getJobReservedAt = function ($timeout) use ($queue) { + $this->connection()->table('jobs')->insert([ + 'queue' => 'default', + 'payload' => json_encode(['timeout' => $timeout]), + 'attempts' => 0, + 'reserved_at' => null, + 'available_at' => Carbon::now()->subSecond()->getTimestamp(), + 'created_at' => Carbon::now()->getTimestamp(), + ]); + + $queue->pop('default'); + + $reservedAt = (int) $this->connection()->table('jobs')->latest()->value('reserved_at'); + + $queue->clear(); + + return $reservedAt; + }; + + // "reserved_at" holds the expiry (job timeout + 10-second buffer) minus the retry_after window + $this->assertSame($time->getTimestamp() + 600 + 10 - $retryAfter, $getJobReservedAt(timeout: 600)); + + // No timeout on the job, worker timeout unknown: the default worker timeout is used + $this->assertSame($time->getTimestamp() + 60 + 10 - $retryAfter, $getJobReservedAt(timeout: null)); + + // Timeout of 0 on the job means it is never killed: reserved effectively forever + $this->assertSame(2147483647, $getJobReservedAt(timeout: 0)); + + // Worker timeout is used if the job doesn't define a timeout + $queue->setWorkerTimeout(120); + $this->assertSame($time->getTimestamp() + 120 + 10 - $retryAfter, $getJobReservedAt(timeout: null)); + + // Timeout of 0 on the job wins over the worker timeout + $this->assertSame(2147483647, $getJobReservedAt(timeout: 0)); + + // Timeout on the job takes precedence over the queue worker timeout + $this->assertSame($time->getTimestamp() + 45 + 10 - $retryAfter, $getJobReservedAt(timeout: 45)); + + // A worker timeout of 0 means jobs are never killed: reserved effectively forever + $queue->setWorkerTimeout(0); + $this->assertSame(2147483647, $getJobReservedAt(timeout: null)); + + // A null worker timeout resets the queue to the default worker timeout + $queue->setWorkerTimeout(null); + $this->assertSame($time->getTimestamp() + 60 + 10 - $retryAfter, $getJobReservedAt(timeout: null)); + + Carbon::setTestNow(); + } + /** * Test that the queue can be cleared. */ diff --git a/tests/Queue/QueueListenerTest.php b/tests/Queue/QueueListenerTest.php index 74aa6d272c75..d93c93ba0560 100755 --- a/tests/Queue/QueueListenerTest.php +++ b/tests/Queue/QueueListenerTest.php @@ -53,7 +53,7 @@ public function testMakeProcessCorrectlyFormatsCommandLine() $this->assertInstanceOf(Process::class, $process); $this->assertEquals(__DIR__, $process->getWorkingDirectory()); $this->assertEquals(3, $process->getTimeout()); - $this->assertEquals($escape.php_binary().$escape." {$escape}{$artisanBinary}{$escape} {$escape}queue:work{$escape} {$escape}connection{$escape} {$escape}--once{$escape} {$escapeMsys}--name=default{$escapeMsys} {$escapeMsys}--queue=queue{$escapeMsys} {$escapeMsys}--backoff=1{$escapeMsys} {$escapeMsys}--memory=2{$escapeMsys} {$escapeMsys}--sleep=3{$escapeMsys} {$escapeMsys}--tries=1{$escapeMsys}", $process->getCommandLine()); + $this->assertEquals($escape.php_binary().$escape." {$escape}{$artisanBinary}{$escape} {$escape}queue:work{$escape} {$escape}connection{$escape} {$escape}--once{$escape} {$escapeMsys}--name=default{$escapeMsys} {$escapeMsys}--queue=queue{$escapeMsys} {$escapeMsys}--backoff=1{$escapeMsys} {$escapeMsys}--memory=2{$escapeMsys} {$escapeMsys}--sleep=3{$escapeMsys} {$escapeMsys}--timeout=3{$escapeMsys} {$escapeMsys}--tries=1{$escapeMsys}", $process->getCommandLine()); } public function testMakeProcessCorrectlyFormatsCommandLineWithAnEnvironmentSpecified() @@ -75,7 +75,7 @@ public function testMakeProcessCorrectlyFormatsCommandLineWithAnEnvironmentSpeci $this->assertInstanceOf(Process::class, $process); $this->assertEquals(__DIR__, $process->getWorkingDirectory()); $this->assertEquals(3, $process->getTimeout()); - $this->assertEquals($escape.php_binary().$escape." {$escape}{$artisanBinary}{$escape} {$escape}queue:work{$escape} {$escape}connection{$escape} {$escape}--once{$escape} {$escapeMsys}--name=default{$escapeMsys} {$escapeMsys}--queue=queue{$escapeMsys} {$escapeMsys}--backoff=1{$escapeMsys} {$escapeMsys}--memory=2{$escapeMsys} {$escapeMsys}--sleep=3{$escapeMsys} {$escapeMsys}--tries=1{$escapeMsys} {$escapeMsys}--env=test{$escapeMsys}", $process->getCommandLine()); + $this->assertEquals($escape.php_binary().$escape." {$escape}{$artisanBinary}{$escape} {$escape}queue:work{$escape} {$escape}connection{$escape} {$escape}--once{$escape} {$escapeMsys}--name=default{$escapeMsys} {$escapeMsys}--queue=queue{$escapeMsys} {$escapeMsys}--backoff=1{$escapeMsys} {$escapeMsys}--memory=2{$escapeMsys} {$escapeMsys}--sleep=3{$escapeMsys} {$escapeMsys}--timeout=3{$escapeMsys} {$escapeMsys}--tries=1{$escapeMsys} {$escapeMsys}--env=test{$escapeMsys}", $process->getCommandLine()); } public function testMakeProcessCorrectlyFormatsCommandLineWhenTheConnectionIsNotSpecified() @@ -97,6 +97,6 @@ public function testMakeProcessCorrectlyFormatsCommandLineWhenTheConnectionIsNot $this->assertInstanceOf(Process::class, $process); $this->assertEquals(__DIR__, $process->getWorkingDirectory()); $this->assertEquals(3, $process->getTimeout()); - $this->assertEquals($escape.php_binary().$escape." {$escape}{$artisanBinary}{$escape} {$escape}queue:work{$escape} {$escape}--once{$escape} {$escapeMsys}--name=default{$escapeMsys} {$escapeMsys}--queue=queue{$escapeMsys} {$escapeMsys}--backoff=1{$escapeMsys} {$escapeMsys}--memory=2{$escapeMsys} {$escapeMsys}--sleep=3{$escapeMsys} {$escapeMsys}--tries=1{$escapeMsys} {$escapeMsys}--env=test{$escapeMsys}", $process->getCommandLine()); + $this->assertEquals($escape.php_binary().$escape." {$escape}{$artisanBinary}{$escape} {$escape}queue:work{$escape} {$escape}--once{$escape} {$escapeMsys}--name=default{$escapeMsys} {$escapeMsys}--queue=queue{$escapeMsys} {$escapeMsys}--backoff=1{$escapeMsys} {$escapeMsys}--memory=2{$escapeMsys} {$escapeMsys}--sleep=3{$escapeMsys} {$escapeMsys}--timeout=3{$escapeMsys} {$escapeMsys}--tries=1{$escapeMsys} {$escapeMsys}--env=test{$escapeMsys}", $process->getCommandLine()); } } diff --git a/tests/Queue/QueueWorkerTest.php b/tests/Queue/QueueWorkerTest.php index 6daa67f4c48f..4233fc8f420b 100755 --- a/tests/Queue/QueueWorkerTest.php +++ b/tests/Queue/QueueWorkerTest.php @@ -673,6 +673,25 @@ public function interrupted(int $signal): void $this->assertSame(15, $interruptible->receivedSignal); } + public function testWorkerSharesItsTimeoutWithTheConnectionWhenRunningTheNextJob() + { + $connection = new WorkerFakeConnection('default', ['queue' => [$job = new WorkerFakeJob]]); + + $worker = new InsomniacWorker( + new WorkerFakeManager('default', $connection), + $this->events, + $this->exceptionHandler, + function () { + return false; + } + ); + + $worker->runNextJob('default', 'queue', $this->workerOptions(['timeout' => 123])); + + $this->assertSame(123, $connection->workerTimeout); + $this->assertTrue($job->fired); + } + /** * Helpers... */ @@ -770,6 +789,7 @@ class WorkerFakeConnection { public $connectionName; public $jobs = []; + public $workerTimeout = false; public function __construct($connectionName, $jobs) { @@ -782,6 +802,13 @@ public function pop($queue) return array_shift($this->jobs[$queue]); } + public function setWorkerTimeout($timeout) + { + $this->workerTimeout = $timeout; + + return $this; + } + public function getConnectionName() { return $this->connectionName;