From 8175cf0acf1d2f9cf1a464cb3a60a743471e305d Mon Sep 17 00:00:00 2001 From: Eugene Leonovich Date: Fri, 30 Oct 2026 00:08:59 +0100 Subject: [PATCH] refactor: move subqueue context handling from AbstractQueue to Subqueue Remove the generic context machinery ($context property, callWithContext(), withContext()) from AbstractQueue and make Subqueue explicitly manage its own subqueue parameter routing per method. This makes AbstractQueue simpler and context-free, while Subqueue becomes self-contained with explicit per-method parameter passing that matches the Tarantool subqueuettl Lua driver contract: - put/release: merge subqueue opts into the options array - take/touch/ack/peek/bury/kick/delete/truncate: append subqueue opts as a trailing argument --- src/AbstractQueue.php | 52 ++-------- src/Subqueue.php | 66 ++++++++++++- tests/Integration/SubqueueTtlQueueTest.php | 105 +++++++++++++++++++-- 3 files changed, 164 insertions(+), 59 deletions(-) diff --git a/src/AbstractQueue.php b/src/AbstractQueue.php index 5fbfe9c..b194ea9 100644 --- a/src/AbstractQueue.php +++ b/src/AbstractQueue.php @@ -17,9 +17,6 @@ abstract class AbstractQueue { - /** @var array */ - protected array $context = []; - public function __construct( private readonly Client $client, private readonly string $name, @@ -38,60 +35,46 @@ public function getName() : string public function take(?float $timeout = null) : ?Task { - $args = [$timeout]; - if ([] !== $this->context) { - $args[] = $this->context; - } - - $result = $this->call('take', ...$args); + $result = $this->call('take', $timeout); return empty($result[0]) ? null : Task::fromTuple($result[0]); } public function touch(int $taskId, float $increment) : ?Task { - $args = [$taskId, $increment]; - if ([] !== $this->context) { - $args[] = $this->context; - } - - $result = $this->call('touch', ...$args); + $result = $this->call('touch', $taskId, $increment); return empty($result[0]) ? null : Task::fromTuple($result[0]); } public function ack(int $taskId) : Task { - return Task::fromTuple($this->callWithContext('ack', $taskId)[0]); + return Task::fromTuple($this->call('ack', $taskId)[0]); } public function peek(int $taskId) : Task { - return Task::fromTuple($this->callWithContext('peek', $taskId)[0]); + return Task::fromTuple($this->call('peek', $taskId)[0]); } public function bury(int $taskId) : Task { - return Task::fromTuple($this->callWithContext('bury', $taskId)[0]); + return Task::fromTuple($this->call('bury', $taskId)[0]); } public function kick(int $count) : int { - return $this->callWithContext('kick', $count)[0]; + return $this->call('kick', $count)[0]; } public function delete(int $taskId) : Task { - return Task::fromTuple($this->callWithContext('delete', $taskId)[0]); + return Task::fromTuple($this->call('delete', $taskId)[0]); } public function truncate() : void { - $args = []; - if ([] !== $this->context) { - $args[] = $this->context; - } - $this->call('truncate', ...$args); + $this->call('truncate'); } /** @@ -121,23 +104,4 @@ public function call(string $methodName, ...$args) : array { return $this->client->call("queue.tube.$this->name:$methodName", ...$args); } - - private function callWithContext(string $methodName, mixed ...$args) : array - { - if ([] !== $this->context) { - $args[] = $this->context; - } - - return $this->call($methodName, ...$args); - } - - /** - * @param array $options - * - * @return array - */ - protected function withContext(array $options) : array - { - return $this->context + $options; - } } diff --git a/src/Subqueue.php b/src/Subqueue.php index 451ff68..b0d2d11 100644 --- a/src/Subqueue.php +++ b/src/Subqueue.php @@ -17,10 +17,12 @@ final class Subqueue extends AbstractQueue { + private readonly string $subqueueName; + public function __construct(Client $client, string $queueName, string $subqueueName) { parent::__construct($client, $queueName); - $this->context = ['subqueue' => $subqueueName]; + $this->subqueueName = $subqueueName; } public function put( @@ -30,20 +32,74 @@ public function put( ?float $ttr = null, ?float $delay = null, ) : Task { - $options = array_filter([ + $options = ['subqueue' => $this->subqueueName] + array_filter([ 'pri' => $pri, 'ttl' => $ttl, 'ttr' => $ttr, 'delay' => $delay, ], static fn ($value) => null !== $value); - return Task::fromTuple($this->call('put', $data, $this->withContext($options))[0]); + return Task::fromTuple($this->call('put', $data, $options)[0]); } public function release(int $taskId, ?float $delay = null) : Task { - $options = null === $delay ? [] : ['delay' => $delay]; + $options = null === $delay + ? ['subqueue' => $this->subqueueName] + : ['subqueue' => $this->subqueueName, 'delay' => $delay]; + + return Task::fromTuple($this->call('release', $taskId, $options)[0]); + } + + #[\Override] + public function take(?float $timeout = null) : ?Task + { + $result = $this->call('take', $timeout, ['subqueue' => $this->subqueueName]); + + return empty($result[0]) ? null : Task::fromTuple($result[0]); + } + + #[\Override] + public function touch(int $taskId, float $increment) : ?Task + { + $result = $this->call('touch', $taskId, $increment, ['subqueue' => $this->subqueueName]); + + return empty($result[0]) ? null : Task::fromTuple($result[0]); + } + + #[\Override] + public function ack(int $taskId) : Task + { + return Task::fromTuple($this->call('ack', $taskId, ['subqueue' => $this->subqueueName])[0]); + } - return Task::fromTuple($this->call('release', $taskId, $this->withContext($options))[0]); + #[\Override] + public function peek(int $taskId) : Task + { + return Task::fromTuple($this->call('peek', $taskId, ['subqueue' => $this->subqueueName])[0]); + } + + #[\Override] + public function bury(int $taskId) : Task + { + return Task::fromTuple($this->call('bury', $taskId, ['subqueue' => $this->subqueueName])[0]); + } + + #[\Override] + public function kick(int $count) : int + { + return $this->call('kick', $count, ['subqueue' => $this->subqueueName])[0]; + } + + #[\Override] + public function delete(int $taskId) : Task + { + return Task::fromTuple($this->call('delete', $taskId, ['subqueue' => $this->subqueueName])[0]); + } + + #[\Override] + public function truncate() : void + { + $this->call('truncate', ['subqueue' => $this->subqueueName]); } } diff --git a/tests/Integration/SubqueueTtlQueueTest.php b/tests/Integration/SubqueueTtlQueueTest.php index 2231ae7..10ea5fc 100644 --- a/tests/Integration/SubqueueTtlQueueTest.php +++ b/tests/Integration/SubqueueTtlQueueTest.php @@ -16,26 +16,111 @@ use Tarantool\PhpUnit\Attribute\Lua; use Tarantool\Queue\Queue; use Tarantool\Queue\States; +use Tarantool\Queue\Subqueue; final class SubqueueTtlQueueTest extends TestCase { - #[Lua("tube = create_tube('%tube_name%', '%tube_type%')")] + private function createSubqueue(string $name = 'emails') : Subqueue + { + return Queue::subqueueTtl($this->getClient(), $this->getQueueName()) + ->subqueue($name); + } + + #[Lua("create_tube('%tube_name%', '%tube_type%')")] public function testTakeFromNewSubqueueReturnsNull() : void { - $subqueue = Queue::subqueueTtl($this->getClient(), $this->getQueueName()) - ->subqueue('created-on-first-put'); + self::assertNull($this->createSubqueue('created-on-first-put')->take(0.0)); + } + + #[Lua("tube = create_tube('%tube_name%', '%tube_type%')")] + public function testPut() : void + { + $subqueue = $this->createSubqueue(); + + self::assertTask($subqueue->put('payload', ttl: 60.0), 0, States::READY, 'payload'); + } - self::assertNull($subqueue->take(0.0)); + #[Lua("tube = create_tube('%tube_name%', '%tube_type%')")] + #[Lua("tube:put('peek_0', {subqueue = 'emails'})")] + public function testPeek() : void + { + self::assertTask($this->createSubqueue()->peek(0), 0, States::READY, 'peek_0'); + } + + #[Lua("tube = create_tube('%tube_name%', '%tube_type%')")] + #[Lua("tube:put('take_0', {subqueue = 'emails'})")] + public function testTake() : void + { + self::assertTask($this->createSubqueue()->take(1.0), 0, States::TAKEN, 'take_0'); + } + + #[Lua("tube = create_tube('%tube_name%', '%tube_type%')")] + #[Lua("tube:put('ack_0', {subqueue = 'emails'})")] + #[Lua("tube:take(0.001, {subqueue = 'emails'})")] + public function testAck() : void + { + self::assertTask($this->createSubqueue()->ack(0), 0, States::DONE, 'ack_0'); + } + + #[Lua("tube = create_tube('%tube_name%', '%tube_type%')")] + #[Lua("tube:put('release_0', {subqueue = 'emails'})")] + #[Lua("tube:take(0.001, {subqueue = 'emails'})")] + public function testRelease() : void + { + self::assertTask($this->createSubqueue()->release(0), 0, States::READY, 'release_0'); + } + + #[Lua("tube = create_tube('%tube_name%', '%tube_type%')")] + #[Lua("tube:put('release_delay_0', {subqueue = 'emails'})")] + #[Lua("tube:take(0.001, {subqueue = 'emails'})")] + public function testReleaseWithDelay() : void + { + self::assertTask($this->createSubqueue()->release(0, delay: 0.1), 0, States::DELAYED, 'release_delay_0'); + } + + #[Lua("tube = create_tube('%tube_name%', '%tube_type%')")] + #[Lua("tube:put('touch_0', {subqueue = 'emails', ttr = 2.0})")] + #[Lua("tube:take(0.001, {subqueue = 'emails'})")] + public function testTouch() : void + { + $task = $this->createSubqueue()->touch(0, 5.0); + + self::assertNotNull($task); + self::assertSame(0, $task->getId()); + self::assertSame(States::TAKEN, $task->getState()); + } + + #[Lua("tube = create_tube('%tube_name%', '%tube_type%')")] + #[Lua("tube:put('bury_0', {subqueue = 'emails'})")] + public function testBury() : void + { + self::assertTask($this->createSubqueue()->bury(0), 0, States::BURIED, 'bury_0'); + } + + #[Lua("tube = create_tube('%tube_name%', '%tube_type%')")] + #[Lua("tube:put('kick_0', {subqueue = 'emails'})")] + #[Lua('tube:bury(0)')] + public function testKick() : void + { + self::assertSame(1, $this->createSubqueue()->kick(1)); + } + + #[Lua("tube = create_tube('%tube_name%', '%tube_type%')")] + #[Lua("tube:put('delete_0', {subqueue = 'emails'})")] + public function testDelete() : void + { + self::assertTask($this->createSubqueue()->delete(0), 0, States::DONE, 'delete_0'); } #[Lua("tube = create_tube('%tube_name%', '%tube_type%')")] - public function testSubqueueRoutesOperations() : void + #[Lua("tube:put('truncate_0', {subqueue = 'emails'})")] + #[Lua("tube:put('truncate_1', {subqueue = 'emails'})")] + public function testTruncate() : void { - $subqueue = Queue::subqueueTtl($this->getClient(), $this->getQueueName()) - ->subqueue('emails'); + $subqueue = $this->createSubqueue(); - self::assertTask($subqueue->put('typed', ttl: 60.0), 0, States::READY, 'typed'); - self::assertTask($subqueue->take(0.0), 0, States::TAKEN, 'typed'); - self::assertTask($subqueue->ack(0), 0, States::DONE, 'typed'); + self::assertSame(2, $this->queue->stats('tasks.total')); + $subqueue->truncate(); + self::assertSame(0, $this->queue->stats('tasks.total')); } }