Add typed queue APIs and scoped subqueues

README.md

@@ -72,6 +72,32 @@ $queue = new Queue($client, 'foobar');
 
 where `$client` is an instance of `Tarantool\Client\Client` from the [tarantool/client](https://github.com/tarantool-php/client) package.
 
+If you want PHP to expose only the options supported by a specific built-in queue type, use its typed queue class:
+
+```php
+$queue = Queue::fifottl($client, 'foobar');
+$task = $queue->put('foo', ttl: 60.0, delay: 5.0);
+
+$limitedQueue = Queue::limfifottl($client, 'bounded');
+$taskOrNull = $limitedQueue->put('bar', timeout: 0.0); // null when the queue is full
+```
+
+Typed queue classes describe the queue type expected by the caller; they do not create the tube or verify its server-side type.
+The original `new Queue($client, $name)` API remains available for generic and custom queues.
+With this generic API, PHP does not validate option arrays against the queue type; the server-side driver decides whether to accept them.
+Its `put()` method returns `Task|null` because some drivers can decline a task, for example when a size limit is reached.
+
+For queues with named subqueues, create a `Subqueue` object once and use it for subsequent operations:
+
+```php
+$subqueue = Queue::subqueuettl($client, 'jobs')->subqueue('emails');
+$task = $subqueue->put('send message', ttl: 60.0);
+$task = $subqueue->take(timeout: 2.0);
+$subqueue->ack($task->getId());
+```
+
+Subqueue names do not need to be created in advance. Calling `subqueue()` only creates a PHP object; the driver registers a name when a task is put there. Taking from an empty subqueue returns `null`. The driver takes from one named subqueue at a time and does not provide a take-from-any-subqueue operation.
+
 
 ### Data types
 

src/AbstractQueue.php

@@ -0,0 +1,143 @@
+<?php
+
+/**
+ * This file is part of the tarantool/queue package.
+ *
+ * (c) Eugene Leonovich <gen.work@gmail.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+declare(strict_types=1);
+
+namespace Tarantool\Queue;
+
+use Tarantool\Client\Client;
+
+abstract class AbstractQueue
+{
+    /** @var array<string, mixed> */
+    protected array $context = [];
+
+    public function __construct(
+        private readonly Client $client,
+        private readonly string $name,
+    ) {
+    }
+
+    public function getClient() : Client
+    {
+        return $this->client;
+    }
+
+    public function getName() : string
+    {
+        return $this->name;
+    }
+
+    public function take(?float $timeout = null) : ?Task
+    {
+        $args = [$timeout];
+        if ([] !== $this->context) {
+            $args[] = $this->context;
+        }
+
+        $result = $this->call('take', ...$args);
+
+        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);
+
+        return empty($result[0]) ? null : Task::fromTuple($result[0]);
+    }
+
+    public function ack(int $taskId) : Task
+    {
+        return Task::fromTuple($this->callWithContext('ack', $taskId)[0]);
+    }
+
+    public function peek(int $taskId) : Task
+    {
+        return Task::fromTuple($this->callWithContext('peek', $taskId)[0]);
+    }
+
+    public function bury(int $taskId) : Task
+    {
+        return Task::fromTuple($this->callWithContext('bury', $taskId)[0]);
+    }
+
+    public function kick(int $count) : int
+    {
+        return $this->callWithContext('kick', $count)[0];
+    }
+
+    public function delete(int $taskId) : Task
+    {
+        return Task::fromTuple($this->callWithContext('delete', $taskId)[0]);
+    }
+
+    public function truncate() : void
+    {
+        $args = [];
+        if ([] !== $this->context) {
+            $args[] = $this->context;
+        }
+        $this->call('truncate', ...$args);
+    }
+
+    /**
+     * @throws \InvalidArgumentException
+     *
+     * @return array|int
+     */
+    public function stats(?string $path = null)
+    {
+        [$stats] = $this->client->call('queue.stats', $this->name);
+
+        if (null === $path) {
+            return $stats;
+        }
+
+        foreach (explode('.', $path) as $key) {
+            if (!isset($stats[$key])) {
+                throw new \InvalidArgumentException(\sprintf('Invalid path "%s"', $path));
+            }
+            $stats = $stats[$key];
+        }
+
+        return $stats;
+    }
+
+    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/FifoQueue.php

@@ -0,0 +1,27 @@
+<?php
+
+/**
+ * This file is part of the tarantool/queue package.
+ *
+ * (c) Eugene Leonovich <gen.work@gmail.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+declare(strict_types=1);
+
+namespace Tarantool\Queue;
+
+final class FifoQueue extends AbstractQueue
+{
+    public function put(mixed $data) : Task
+    {
+        return Task::fromTuple($this->call('put', $data)[0]);
+    }
+
+    public function release(int $taskId) : Task
+    {
+        return Task::fromTuple($this->call('release', $taskId)[0]);
+    }
+}

src/FifottlQueue.php

@@ -0,0 +1,41 @@
+<?php
+
+/**
+ * This file is part of the tarantool/queue package.
+ *
+ * (c) Eugene Leonovich <gen.work@gmail.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+declare(strict_types=1);
+
+namespace Tarantool\Queue;
+
+final class FifottlQueue extends AbstractQueue
+{
+    public function put(
+        mixed $data,
+        ?int $pri = null,
+        ?float $ttl = null,
+        ?float $ttr = null,
+        ?float $delay = null,
+    ) : Task {
+        $options = array_filter([
+            'pri' => $pri,
+            'ttl' => $ttl,
+            'ttr' => $ttr,
+            'delay' => $delay,
+        ], static fn ($value) => null !== $value);
+
+        return Task::fromTuple($this->call('put', $data, $options)[0]);
+    }
+
+    public function release(int $taskId, ?float $delay = null) : Task
+    {
+        $options = null === $delay ? [] : ['delay' => $delay];
+
+        return Task::fromTuple($this->call('release', $taskId, $options)[0]);
+    }
+}

src/LimfifottlQueue.php

@@ -0,0 +1,44 @@
+<?php
+
+/**
+ * This file is part of the tarantool/queue package.
+ *
+ * (c) Eugene Leonovich <gen.work@gmail.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+declare(strict_types=1);
+
+namespace Tarantool\Queue;
+
+final class LimfifottlQueue extends AbstractQueue
+{
+    public function put(
+        mixed $data,
+        ?int $pri = null,
+        ?float $ttl = null,
+        ?float $ttr = null,
+        ?float $delay = null,
+        ?float $timeout = null,
+    ) : ?Task {
+        $options = array_filter([
+            'pri' => $pri,
+            'ttl' => $ttl,
+            'ttr' => $ttr,
+            'delay' => $delay,
+            'timeout' => $timeout,
+        ], static fn ($value) => null !== $value);
+        $result = $this->call('put', $data, $options);
+
+        return empty($result[0]) ? null : Task::fromTuple($result[0]);
+    }
+
+    public function release(int $taskId, ?float $delay = null) : Task
+    {
+        $options = null === $delay ? [] : ['delay' => $delay];
+
+        return Task::fromTuple($this->call('release', $taskId, $options)[0]);
+    }
+}