refactor: move subqueue context handling from AbstractQueue to Subqueue

src/AbstractQueue.php

@@ -17,9 +17,6 @@
 
 abstract class AbstractQueue
 {
-    /** @var array<string, mixed> */
-    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<string, mixed> $options
-     *
-     * @return array<string, mixed>
-     */
-    protected function withContext(array $options) : array
-    {
-        return $this->context + $options;
-    }
 }

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]);
     }
 }

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'));
     }
 }