From e2d94922d99efce931422c532e7938e015b0b0fc Mon Sep 17 00:00:00 2001 From: landa Date: Wed, 21 Oct 2026 21:33:35 +0300 Subject: [PATCH 1/2] tests: fix queue stats integration tests In latest queue update new subqueuettl driver was added. This driver needs per-subqueue statistics and adds a new `stats.extra` field for it. This addition also allows each driver to report meaningful operational metrics without extending the generic statistics implementation for every driver-specific use case. --- tests/Integration/QueueTest.php | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/tests/Integration/QueueTest.php b/tests/Integration/QueueTest.php index af45e04..5034f05 100644 --- a/tests/Integration/QueueTest.php +++ b/tests/Integration/QueueTest.php @@ -220,6 +220,8 @@ public function testStats() : void 'ttr' => 0, 'delay' => 0, 'ttl' => 0, + ], + 'extra' => [ ], ], $stats); } @@ -252,6 +254,8 @@ public function testEmptyStats() : void 'ttr' => 0, 'delay' => 0, 'ttl' => 0, + ], + 'extra' => [ ], ], $stats); } From 6d75f690c6f9cf8336c7eec4a59ae0963f7e2f33 Mon Sep 17 00:00:00 2001 From: landa Date: Thu, 22 Oct 2026 00:53:03 +0300 Subject: [PATCH 2/2] Add subqueuettl driver to the test Adds subqueuettl driver to the tests. This driver contains named subqueues stored in one space. It requires a subqueue name for put() and take(), so consumers can take tasks from a specific subqueue. Supports TTL, TTR, priority, and delay. In order to support this driver php functions have to support a subqueue name as an optional argument, also tests have to be updated providing this names. --- src/Queue.php | 32 ++--- tests/Integration/QueueTest.php | 132 +++++++++--------- tests/Integration/SubqueuettlQueueTest.php | 50 +++++++ tests/Integration/TestCase.php | 14 ++ tests/Integration/Ttl.php | 66 ++++----- tests/Integration/TubePlaceholderResolver.php | 3 + 6 files changed, 182 insertions(+), 115 deletions(-) create mode 100644 tests/Integration/SubqueuettlQueueTest.php diff --git a/src/Queue.php b/src/Queue.php index 152f9ae..a96db89 100644 --- a/src/Queue.php +++ b/src/Queue.php @@ -46,24 +46,24 @@ public function put($data, array $options = []) : Task ); } - public function take(float $timeout = null) : ?Task + public function take(float $timeout = null, array $opts = []) : ?Task { - $result = $this->client->call("queue.tube.$this->name:take", $timeout); + $result = $this->client->call("queue.tube.$this->name:take", $timeout, $opts); return empty($result[0]) ? null : Task::fromTuple($result[0]); } - public function touch(int $taskId, float $increment) : ?Task + public function touch(int $taskId, float $increment, array $opts = []) : ?Task { - $result = $this->client->call("queue.tube.$this->name:touch", $taskId, $increment); + $result = $this->client->call("queue.tube.$this->name:touch", $taskId, $increment, $opts); return empty($result[0]) ? null : Task::fromTuple($result[0]); } - public function ack(int $taskId) : Task + public function ack(int $taskId, array $opts = []) : Task { return Task::fromTuple( - $this->client->call("queue.tube.$this->name:ack", $taskId)[0] + $this->client->call("queue.tube.$this->name:ack", $taskId, $opts)[0] ); } @@ -74,35 +74,35 @@ public function release(int $taskId, array $options = []) : Task ); } - public function peek(int $taskId) : Task + public function peek(int $taskId, array $opts = []) : Task { return Task::fromTuple( - $this->client->call("queue.tube.$this->name:peek", $taskId)[0] + $this->client->call("queue.tube.$this->name:peek", $taskId, $opts)[0] ); } - public function bury(int $taskId) : Task + public function bury(int $taskId, array $opts = []) : Task { return Task::fromTuple( - $this->client->call("queue.tube.$this->name:bury", $taskId)[0] + $this->client->call("queue.tube.$this->name:bury", $taskId, $opts)[0] ); } - public function kick(int $count) : int + public function kick(int $count, array $opts = []) : int { - return $this->client->call("queue.tube.$this->name:kick", $count)[0]; + return $this->client->call("queue.tube.$this->name:kick", $count, $opts)[0]; } - public function delete(int $taskId) : Task + public function delete(int $taskId, array $opts = []) : Task { return Task::fromTuple( - $this->client->call("queue.tube.$this->name:delete", $taskId)[0] + $this->client->call("queue.tube.$this->name:delete", $taskId, $opts)[0] ); } - public function truncate() : void + public function truncate(array $opts = []) : void { - $this->client->call("queue.tube.$this->name:truncate"); + $this->client->call("queue.tube.$this->name:truncate", $opts); } /** diff --git a/tests/Integration/QueueTest.php b/tests/Integration/QueueTest.php index 5034f05..e8bdb33 100644 --- a/tests/Integration/QueueTest.php +++ b/tests/Integration/QueueTest.php @@ -25,7 +25,7 @@ abstract class QueueTest extends TestCase */ public function testPut($data) : void { - $task = $this->queue->put($data); + $task = $this->queue->put($data, $this->withSubqueue()); self::assertTask($task, 0, States::READY, $data); } @@ -46,22 +46,22 @@ public function provideTaskData() : iterable /** * @lua tube = create_tube('%tube_name%', '%tube_type%') - * @lua tube:put('peek_0') + * @lua tube:put('peek_0', {subqueue = %subqueue%}) */ public function testPeek() : void { - $task = $this->queue->peek(0); + $task = $this->queue->peek(0, $this->withSubqueue()); self::assertTask($task, 0, States::READY, 'peek_0'); } /** * @lua tube = create_tube('%tube_name%', '%tube_type%') - * @lua tube:put('take') + * @lua tube:put('take', {subqueue = %subqueue%}) */ public function testTake() : void { - $task = $this->queue->take(); + $task = $this->queue->take(0, $this->withSubqueue()); self::assertTask($task, 0, States::TAKEN, 'take'); } @@ -72,7 +72,7 @@ public function testTake() : void public function testTakeNone() : void { $time = time(); - $task = $this->queue->take(2); + $task = $this->queue->take(2, $this->withSubqueue()); self::assertGreaterThanOrEqual(2, time() - $time); self::assertNull($task); @@ -80,88 +80,88 @@ public function testTakeNone() : void /** * @lua tube = create_tube('%tube_name%', '%tube_type%') - * @lua tube:put('release_0') - * @lua tube:take() + * @lua tube:put('release_0', {subqueue = %subqueue%}) + * @lua tube:take(0, {subqueue = %subqueue%}) */ public function testRelease() : void { - $task = $this->queue->release(0); + $task = $this->queue->release(0, $this->withSubqueue()); self::assertTask($task, 0, States::READY, 'release_0'); } /** * @lua tube = create_tube('%tube_name%', '%tube_type%') - * @lua tube:put('ack_0') - * @lua tube:take() + * @lua tube:put('ack_0', {subqueue = %subqueue%}) + * @lua tube:take(0, {subqueue = %subqueue%}) */ public function testAck() : void { - $task = $this->queue->ack(0); + $task = $this->queue->ack(0, $this->withSubqueue()); self::assertTask($task, 0, States::DONE, 'ack_0'); } /** * @lua tube = create_tube('%tube_name%', '%tube_type%') - * @lua tube:put('bury_0') + * @lua tube:put('bury_0', {subqueue = %subqueue%}) */ public function testBury() : void { - $task = $this->queue->bury(0); + $task = $this->queue->bury(0, $this->withSubqueue()); self::assertTask($task, 0, States::BURIED, 'bury_0'); } /** * @lua tube = create_tube('%tube_name%', '%tube_type%') - * @lua tube:put('kick_1') - * @lua tube:bury(0) + * @lua tube:put('kick_1', {subqueue = %subqueue%}) + * @lua tube:bury(0, {subqueue = %subqueue%}) */ public function testKickOne() : void { - $count = $this->queue->kick(1); + $count = $this->queue->kick(1, $this->withSubqueue()); self::assertSame(1, $count); } /** * @lua tube = create_tube('%tube_name%', '%tube_type%') - * @lua tube:put('kick_1') - * @lua tube:put('kick_2') - * @lua tube:put('kick_3') - * @lua tube:bury(0) - * @lua tube:bury(1) - * @lua tube:bury(2) + * @lua tube:put('kick_1', {subqueue = %subqueue%}) + * @lua tube:put('kick_2', {subqueue = %subqueue%}) + * @lua tube:put('kick_3', {subqueue = %subqueue%}) + * @lua tube:bury(0, {subqueue = %subqueue%}) + * @lua tube:bury(1, {subqueue = %subqueue%}) + * @lua tube:bury(2, {subqueue = %subqueue%}) */ public function testKickMany() : void { - $count = $this->queue->kick(3); + $count = $this->queue->kick(3, $this->withSubqueue()); self::assertSame(3, $count); } /** * @lua tube = create_tube('%tube_name%', '%tube_type%') - * @lua tube:put('delete_0') + * @lua tube:put('delete_0', {subqueue = %subqueue%}) */ public function testDelete() : void { - $task = $this->queue->delete(0); + $task = $this->queue->delete(0, $this->withSubqueue()); self::assertTask($task, 0, States::DONE, 'delete_0'); } /** * @lua tube = create_tube('%tube_name%', '%tube_type%') - * @lua tube:put('truncate_0') - * @lua tube:put('truncate_1') + * @lua tube:put('truncate_0', {subqueue = %subqueue%}) + * @lua tube:put('truncate_1', {subqueue = %subqueue%}) */ public function testTruncate() : void { self::assertSame(2, $this->queue->stats('tasks.total')); - $this->queue->truncate(); + $this->queue->truncate($this->withSubqueue()); self::assertSame(0, $this->queue->stats('tasks.total')); } @@ -180,20 +180,20 @@ public function testTruncateEmpty() : void /** * @lua tube = create_tube('%tube_name%', '%tube_type%') - * @lua tube:put('stat_0') - * @lua tube:put('stat_1') - * @lua tube:put('stat_2') - * @lua tube:put('stat_3') - * @lua tube:put('stat_4') - * @lua tube:delete(4) - * @lua tube:take(.001) - * @lua tube:release(0) - * @lua tube:take(.001) - * @lua tube:ack(0) - * @lua tube:bury(1) - * @lua tube:bury(2) - * @lua tube:kick(1) - * @lua tube:take(.001) + * @lua tube:put('stat_0', {subqueue = %subqueue%}) + * @lua tube:put('stat_1', {subqueue = %subqueue%}) + * @lua tube:put('stat_2', {subqueue = %subqueue%}) + * @lua tube:put('stat_3', {subqueue = %subqueue%}) + * @lua tube:put('stat_4', {subqueue = %subqueue%}) + * @lua tube:delete(4, {subqueue = %subqueue%}) + * @lua tube:take(.001, {subqueue = %subqueue%}) + * @lua tube:release(0, {subqueue = %subqueue%}) + * @lua tube:take(.001, {subqueue = %subqueue%}) + * @lua tube:ack(0, {subqueue = %subqueue%}) + * @lua tube:bury(1, {subqueue = %subqueue%}) + * @lua tube:bury(2, {subqueue = %subqueue%}) + * @lua tube:kick(1, {subqueue = %subqueue%}) + * @lua tube:take(.001, {subqueue = %subqueue%}) */ public function testStats() : void { @@ -262,28 +262,28 @@ public function testEmptyStats() : void /** * @lua tube = create_tube('%tube_name%', '%tube_type%') - * @lua tube:put('stat_0') - * @lua tube:put('stat_1') - * @lua tube:put('stat_2') - * @lua tube:put('stat_3') - * @lua tube:put('stat_4') - * @lua tube:put('stat_5') - * @lua tube:put('stat_6') - * @lua tube:put('stat_7') - * @lua tube:put('stat_8') - * @lua tube:put('stat_9') - * @lua tube:take(.001) - * @lua tube:release(0) - * @lua tube:take(.001) - * @lua tube:ack(0) - * @lua tube:bury(1) - * @lua tube:bury(2) - * @lua tube:bury(3) - * @lua tube:kick(1) - * @lua tube:delete(6) - * @lua tube:delete(7) - * @lua tube:delete(8) - * @lua tube:take(.001) + * @lua tube:put('stat_0', {subqueue = %subqueue%}) + * @lua tube:put('stat_1', {subqueue = %subqueue%}) + * @lua tube:put('stat_2', {subqueue = %subqueue%}) + * @lua tube:put('stat_3', {subqueue = %subqueue%}) + * @lua tube:put('stat_4', {subqueue = %subqueue%}) + * @lua tube:put('stat_5', {subqueue = %subqueue%}) + * @lua tube:put('stat_6', {subqueue = %subqueue%}) + * @lua tube:put('stat_7', {subqueue = %subqueue%}) + * @lua tube:put('stat_8', {subqueue = %subqueue%}) + * @lua tube:put('stat_9', {subqueue = %subqueue%}) + * @lua tube:take(.001, {subqueue = %subqueue%}) + * @lua tube:release(0, {subqueue = %subqueue%}) + * @lua tube:take(.001, {subqueue = %subqueue%}) + * @lua tube:ack(0, {subqueue = %subqueue%}) + * @lua tube:bury(1, {subqueue = %subqueue%}) + * @lua tube:bury(2, {subqueue = %subqueue%}) + * @lua tube:bury(3, {subqueue = %subqueue%}) + * @lua tube:kick(1, {subqueue = %subqueue%}) + * @lua tube:delete(6, {subqueue = %subqueue%}) + * @lua tube:delete(7, {subqueue = %subqueue%}) + * @lua tube:delete(8, {subqueue = %subqueue%}) + * @lua tube:take(.001, {subqueue = %subqueue%}) */ public function testStatsPath() : void { @@ -372,7 +372,7 @@ public function testThrowException(string $methodName, array $args) : void { $this->expectException(\Throwable::class); - $this->queue->$methodName(...$args); + $this->queue->$methodName(...array_merge($args, [$this->withSubqueue()])); } public function provideFailureCallbackData() : iterable diff --git a/tests/Integration/SubqueuettlQueueTest.php b/tests/Integration/SubqueuettlQueueTest.php new file mode 100644 index 0000000..08fc79a --- /dev/null +++ b/tests/Integration/SubqueuettlQueueTest.php @@ -0,0 +1,50 @@ +queue->stats(); + self::assertSame(0, $stats['tasks']['total']); + self::assertArrayHasKey('subqueues', $stats['extra']); + } + + public function getSubqueue() : ?string + { + return self::SUBQUEUE; + } + + public function testStats() : void + { + $stats = $this->queue->stats(); + + self::assertEquals([ + 'tasks' => [ + 'taken' => 0, 'buried' => 1, 'ready' => 2, + 'done' => 2, 'delayed' => 0, 'total' => 3, + ], + 'calls' => [ + 'ack' => 1, 'delete' => 1, 'take' => 3, 'kick' => 1, + 'release' => 2, 'touch' => 0, 'put' => 5, 'bury' => 2, + 'ttr' => 0, 'delay' => 0, 'ttl' => 0, + ], + ], array_diff_key($stats, ['extra' => ''])); + + self::assertArrayHasKey('subqueues', $stats['extra']); + self::assertArrayHasKey('test_subqueue', $stats['extra']['subqueues']); + + $sub = $stats['extra']['subqueues']['test_subqueue']; + self::assertSame(2, $sub['ready']); + self::assertSame(0, $sub['taken']); + self::assertSame(1, $sub['buried']); + self::assertSame(0, $sub['delayed']); + self::assertSame(3, $sub['total']); + } +} diff --git a/tests/Integration/TestCase.php b/tests/Integration/TestCase.php index 09c7e7c..ed4b183 100644 --- a/tests/Integration/TestCase.php +++ b/tests/Integration/TestCase.php @@ -48,6 +48,20 @@ public function getQueueType() : string return strtolower($type); } + public function getSubqueue() : ?string + { + return null; + } + + final protected function withSubqueue(array $opts = []) : array + { + if (null !== $subqueue = $this->getSubqueue()) { + $opts['subqueue'] = $subqueue; + } + + return $opts; + } + final protected function getClient() : Client { if ($this->client) { diff --git a/tests/Integration/Ttl.php b/tests/Integration/Ttl.php index ef2850a..9fc4c24 100644 --- a/tests/Integration/Ttl.php +++ b/tests/Integration/Ttl.php @@ -20,15 +20,15 @@ trait Ttl { /** * @lua tube = create_tube('%tube_name%', '%tube_type%') - * @lua tube:put('ttr_1', {ttr = 1}) + * @lua tube:put('ttr_1', {ttr = 1, subqueue = %subqueue%}) */ public function testTimeToRun() : void { $queue = $this->getQueue(); - $task1 = $queue->take(.1); + $task1 = $queue->take(.1, $this->withSubqueue()); sleep(1); - $task2 = $queue->take(.1); + $task2 = $queue->take(.1, $this->withSubqueue()); self::assertTaskInstance($task1); self::assertSame('ttr_1', $task1->getData()); @@ -37,28 +37,28 @@ public function testTimeToRun() : void /** * @lua tube = create_tube('%tube_name%', '%tube_type%') - * @lua tube:put('ttl_1', {ttl = 1}) + * @lua tube:put('ttl_1', {ttl = 1, subqueue = %subqueue%}) */ public function testTimeToLive() : void { sleep(1); - $task = $this->getQueue()->take(.1); + $task = $this->getQueue()->take(.1, $this->withSubqueue()); self::assertNull($task); } /** * @lua tube = create_tube('%tube_name%', '%tube_type%') - * @lua tube:put('touch_ttr_1', {ttr = 1}) + * @lua tube:put('touch_ttr_1', {ttr = 1, subqueue = %subqueue%}) */ public function testTouchTimeToRun() : void { $queue = $this->getQueue(); - $task1 = $queue->take(.1); - $task2 = $queue->touch($task1->getId(), 1); + $task1 = $queue->take(.1, $this->withSubqueue()); + $task2 = $queue->touch($task1->getId(), 1, $this->withSubqueue()); sleep(1); - $task3 = $queue->take(.1); + $task3 = $queue->take(.1, $this->withSubqueue()); self::assertTaskInstance($task2); self::assertSame('touch_ttr_1', $task2->getData()); @@ -68,17 +68,17 @@ public function testTouchTimeToRun() : void /** * @lua tube = create_tube('%tube_name%', '%tube_type%') - * @lua tube:put('touch_ttl_1', {ttl = 1}) + * @lua tube:put('touch_ttl_1', {ttl = 1, subqueue = %subqueue%}) */ public function testTouchTimeToLive() : void { $queue = $this->getQueue(); - $task1 = $queue->take(.1); - $task2 = $queue->touch($task1->getId(), 1); - $queue->release($task1->getId()); + $task1 = $queue->take(.1, $this->withSubqueue()); + $task2 = $queue->touch($task1->getId(), 1, $this->withSubqueue()); + $queue->release($task1->getId(), $this->withSubqueue()); sleep(1); - $task3 = $queue->take(.1); + $task3 = $queue->take(.1, $this->withSubqueue()); self::assertTaskInstance($task2); self::assertSame('touch_ttl_1', $task2->getData()); @@ -88,32 +88,32 @@ public function testTouchTimeToLive() : void /** * @lua tube = create_tube('%tube_name%', '%tube_type%') - * @lua tube:put('touch_invalid_interval') + * @lua tube:put('touch_invalid_interval', {subqueue = %subqueue%}) */ public function testTouchInvalidInterval() : void { $queue = $this->getQueue(); - $task = $queue->take(.1); + $task = $queue->take(.1, $this->withSubqueue()); foreach ([0, -1] as $interval) { - self::assertNull($queue->touch($task->getId(), $interval)); + self::assertNull($queue->touch($task->getId(), $interval, $this->withSubqueue())); } } /** * @lua tube = create_tube('%tube_name%', '%tube_type%') - * @lua tube:put('pri_low', {pri = 2}) - * @lua tube:put('pri_high', {pri = 1}) + * @lua tube:put('pri_low', {pri = 2, subqueue = %subqueue%}) + * @lua tube:put('pri_high', {pri = 1, subqueue = %subqueue%}) */ public function testPriority() : void { $queue = $this->getQueue(); - $task1 = $queue->take(.1); - $queue->delete($task1->getId()); + $task1 = $queue->take(.1, $this->withSubqueue()); + $queue->delete($task1->getId(), $this->withSubqueue()); - $task2 = $queue->take(.1); - $queue->delete($task2->getId()); + $task2 = $queue->take(.1, $this->withSubqueue()); + $queue->delete($task2->getId(), $this->withSubqueue()); self::assertSame('pri_high', $task1->getData()); self::assertSame('pri_low', $task2->getData()); @@ -121,45 +121,45 @@ public function testPriority() : void /** * @lua tube = create_tube('%tube_name%', '%tube_type%') - * @lua tube:put('delay_1', {delay = 1}) + * @lua tube:put('delay_1', {delay = 1, subqueue = %subqueue%}) */ public function testDelay() : void { $queue = $this->getQueue(); - $task = $queue->peek(0); + $task = $queue->peek(0, $this->withSubqueue()); self::assertTask($task, 0, States::DELAYED, 'delay_1'); sleep(1); - $task = $queue->take(.1); + $task = $queue->take(.1, $this->withSubqueue()); self::assertTask($task, 0, States::TAKEN, 'delay_1'); } /** * @lua tube = create_tube('%tube_name%', '%tube_type%') - * @lua tube:put('release_0') - * @lua tube:take() + * @lua tube:put('release_0', {subqueue = %subqueue%}) + * @lua tube:take(0, {subqueue = %subqueue%}) */ public function testDelayedRelease() : void { $queue = $this->getQueue(); - $queue->release(0, ['delay' => 1]); + $queue->release(0, $this->withSubqueue(['delay' => 1])); - $task = $queue->peek(0); + $task = $queue->peek(0, $this->withSubqueue()); self::assertTask($task, 0, States::DELAYED, 'release_0'); sleep(1); - $task = $queue->take(.1); + $task = $queue->take(.1, $this->withSubqueue()); self::assertTask($task, 0, States::TAKEN, 'release_0'); } /** * @lua tube = create_tube('%tube_name%', '%tube_type%') - * @lua tube:put('stat_delayed_0', {delay = 9999}) - * @lua tube:put('stat_delayed_1', {delay = 9999}) + * @lua tube:put('stat_delayed_0', {delay = 9999, subqueue = %subqueue%}) + * @lua tube:put('stat_delayed_1', {delay = 9999, subqueue = %subqueue%}) */ public function testStatsDelayed() : void { diff --git a/tests/Integration/TubePlaceholderResolver.php b/tests/Integration/TubePlaceholderResolver.php index 2cadeda..bfff43b 100644 --- a/tests/Integration/TubePlaceholderResolver.php +++ b/tests/Integration/TubePlaceholderResolver.php @@ -32,9 +32,12 @@ public function getName() : string public function resolve(string $value, Target $target) : string { + $subqueue = $this->testCase->getSubqueue(); + return strtr($value, [ '%tube_name%' => $this->testCase->getQueueName(), '%tube_type%' => $this->testCase->getQueueType(), + '%subqueue%' => $subqueue === null ? 'nil' : "'".$subqueue."'", ]); } }