From 578f6a68d2ab837292a17eae77d1d23f81110f7f Mon Sep 17 00:00:00 2001 From: Eugene Leonovich Date: Fri, 23 Oct 2026 14:36:35 +0200 Subject: [PATCH 1/2] feat: add typed queue APIs --- README.md | 26 ++++ src/AbstractQueue.php | 143 +++++++++++++++++++++ src/FifoQueue.php | 27 ++++ src/FifottlQueue.php | 41 ++++++ src/LimfifottlQueue.php | 44 +++++++ src/Queue.php | 106 ++++----------- src/Subqueue.php | 49 +++++++ src/SubqueuettlQueue.php | 30 +++++ src/UtubeQueue.php | 27 ++++ src/UtubettlQueue.php | 43 +++++++ tests/Integration/FifottlQueueTest.php | 13 ++ tests/Integration/LimfifottlQueueTest.php | 30 +++++ tests/Integration/SubqueuettlQueueTest.php | 41 ++++++ tests/Unit/TypedQueueTest.php | 110 ++++++++++++++++ 14 files changed, 646 insertions(+), 84 deletions(-) create mode 100644 src/AbstractQueue.php create mode 100644 src/FifoQueue.php create mode 100644 src/FifottlQueue.php create mode 100644 src/LimfifottlQueue.php create mode 100644 src/Subqueue.php create mode 100644 src/SubqueuettlQueue.php create mode 100644 src/UtubeQueue.php create mode 100644 src/UtubettlQueue.php create mode 100644 tests/Integration/LimfifottlQueueTest.php create mode 100644 tests/Integration/SubqueuettlQueueTest.php create mode 100644 tests/Unit/TypedQueueTest.php diff --git a/README.md b/README.md index 142c939..8597d55 100644 --- a/README.md +++ b/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 diff --git a/src/AbstractQueue.php b/src/AbstractQueue.php new file mode 100644 index 0000000..5fbfe9c --- /dev/null +++ b/src/AbstractQueue.php @@ -0,0 +1,143 @@ + + * + * 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 */ + 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 $options + * + * @return array + */ + protected function withContext(array $options) : array + { + return $this->context + $options; + } +} diff --git a/src/FifoQueue.php b/src/FifoQueue.php new file mode 100644 index 0000000..62b082a --- /dev/null +++ b/src/FifoQueue.php @@ -0,0 +1,27 @@ + + * + * 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]); + } +} diff --git a/src/FifottlQueue.php b/src/FifottlQueue.php new file mode 100644 index 0000000..16b6735 --- /dev/null +++ b/src/FifottlQueue.php @@ -0,0 +1,41 @@ + + * + * 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]); + } +} diff --git a/src/LimfifottlQueue.php b/src/LimfifottlQueue.php new file mode 100644 index 0000000..92a21c7 --- /dev/null +++ b/src/LimfifottlQueue.php @@ -0,0 +1,44 @@ + + * + * 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]); + } +} diff --git a/src/Queue.php b/src/Queue.php index 7b3ec7e..883197e 100644 --- a/src/Queue.php +++ b/src/Queue.php @@ -15,115 +15,53 @@ use Tarantool\Client\Client; -class Queue +class Queue extends AbstractQueue { - 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 put(mixed $data, array $options = []) : Task + public static function fifo(Client $client, string $name) : FifoQueue { - return Task::fromTuple( - $this->client->call("queue.tube.$this->name:put", $data, $options)[0] - ); + return new FifoQueue($client, $name); } - public function take(?float $timeout = null) : ?Task + public static function fifottl(Client $client, string $name) : FifottlQueue { - $result = $this->client->call("queue.tube.$this->name:take", $timeout); - - return empty($result[0]) ? null : Task::fromTuple($result[0]); - } - - public function touch(int $taskId, float $increment) : ?Task - { - $result = $this->client->call("queue.tube.$this->name:touch", $taskId, $increment); - - return empty($result[0]) ? null : Task::fromTuple($result[0]); + return new FifottlQueue($client, $name); } - public function ack(int $taskId) : Task + public static function utube(Client $client, string $name) : UtubeQueue { - return Task::fromTuple( - $this->client->call("queue.tube.$this->name:ack", $taskId)[0] - ); + return new UtubeQueue($client, $name); } - public function release(int $taskId, array $options = []) : Task + public static function utubettl(Client $client, string $name) : UtubettlQueue { - return Task::fromTuple( - $this->client->call("queue.tube.$this->name:release", $taskId, $options)[0] - ); + return new UtubettlQueue($client, $name); } - public function peek(int $taskId) : Task + public static function limfifottl(Client $client, string $name) : LimfifottlQueue { - return Task::fromTuple( - $this->client->call("queue.tube.$this->name:peek", $taskId)[0] - ); + return new LimfifottlQueue($client, $name); } - public function bury(int $taskId) : Task + public static function subqueuettl(Client $client, string $name) : SubqueuettlQueue { - return Task::fromTuple( - $this->client->call("queue.tube.$this->name:bury", $taskId)[0] - ); - } - - public function kick(int $count) : int - { - return $this->client->call("queue.tube.$this->name:kick", $count)[0]; - } - - public function delete(int $taskId) : Task - { - return Task::fromTuple( - $this->client->call("queue.tube.$this->name:delete", $taskId)[0] - ); - } - - public function truncate() : void - { - $this->client->call("queue.tube.$this->name:truncate"); + return new SubqueuettlQueue($client, $name); } /** - * @throws \InvalidArgumentException - * - * @return array|int + * @param array $options */ - public function stats(?string $path = null) + public function put(mixed $data, array $options = []) : ?Task { - [$stats] = $this->client->call('queue.stats', $this->name); - - if (null === $path) { - return $stats; - } + $result = $this->call('put', $data, $options); - foreach (explode('.', $path) as $key) { - if (!isset($stats[$key])) { - throw new \InvalidArgumentException(\sprintf('Invalid path "%s"', $path)); - } - $stats = $stats[$key]; - } - - return $stats; + return empty($result[0]) ? null : Task::fromTuple($result[0]); } - public function call(string $methodName, ...$args) : array + /** + * @param array $options + */ + public function release(int $taskId, array $options = []) : Task { - return $this->client->call("queue.tube.$this->name:$methodName", ...$args); + return Task::fromTuple($this->call('release', $taskId, $options)[0]); } } diff --git a/src/Subqueue.php b/src/Subqueue.php new file mode 100644 index 0000000..451ff68 --- /dev/null +++ b/src/Subqueue.php @@ -0,0 +1,49 @@ + + * + * 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; + +final class Subqueue extends AbstractQueue +{ + public function __construct(Client $client, string $queueName, string $subqueueName) + { + parent::__construct($client, $queueName); + $this->context = ['subqueue' => $subqueueName]; + } + + 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, $this->withContext($options))[0]); + } + + public function release(int $taskId, ?float $delay = null) : Task + { + $options = null === $delay ? [] : ['delay' => $delay]; + + return Task::fromTuple($this->call('release', $taskId, $this->withContext($options))[0]); + } +} diff --git a/src/SubqueuettlQueue.php b/src/SubqueuettlQueue.php new file mode 100644 index 0000000..9f90b43 --- /dev/null +++ b/src/SubqueuettlQueue.php @@ -0,0 +1,30 @@ + + * + * 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; + +final class SubqueuettlQueue +{ + public function __construct( + private readonly Client $client, + private readonly string $name, + ) { + } + + public function subqueue(string $name) : Subqueue + { + return new Subqueue($this->client, $this->name, $name); + } +} diff --git a/src/UtubeQueue.php b/src/UtubeQueue.php new file mode 100644 index 0000000..fa46873 --- /dev/null +++ b/src/UtubeQueue.php @@ -0,0 +1,27 @@ + + * + * 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 UtubeQueue extends AbstractQueue +{ + public function put(mixed $data, string $utube) : Task + { + return Task::fromTuple($this->call('put', $data, ['utube' => $utube])[0]); + } + + public function release(int $taskId) : Task + { + return Task::fromTuple($this->call('release', $taskId, [])[0]); + } +} diff --git a/src/UtubettlQueue.php b/src/UtubettlQueue.php new file mode 100644 index 0000000..32d23d8 --- /dev/null +++ b/src/UtubettlQueue.php @@ -0,0 +1,43 @@ + + * + * 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 UtubettlQueue extends AbstractQueue +{ + public function put( + mixed $data, + string $utube, + ?int $pri = null, + ?float $ttl = null, + ?float $ttr = null, + ?float $delay = null, + ) : Task { + $options = array_filter([ + 'utube' => $utube, + '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]); + } +} diff --git a/tests/Integration/FifottlQueueTest.php b/tests/Integration/FifottlQueueTest.php index 721770a..07f13ff 100644 --- a/tests/Integration/FifottlQueueTest.php +++ b/tests/Integration/FifottlQueueTest.php @@ -13,7 +13,20 @@ namespace Tarantool\Queue\Tests\Integration; +use Tarantool\PhpUnit\Attribute\Lua; +use Tarantool\Queue\Queue; +use Tarantool\Queue\States; + final class FifottlQueueTest extends QueueTest { use Ttl; + + #[Lua("create_tube('%tube_name%', '%tube_type%')")] + public function testTypedQueuePut() : void + { + $queue = Queue::fifottl($this->getClient(), $this->getQueueName()); + $task = $queue->put('typed', ttl: 60.0, pri: 0); + + self::assertTask($task, 0, States::READY, 'typed'); + } } diff --git a/tests/Integration/LimfifottlQueueTest.php b/tests/Integration/LimfifottlQueueTest.php new file mode 100644 index 0000000..2154e5a --- /dev/null +++ b/tests/Integration/LimfifottlQueueTest.php @@ -0,0 +1,30 @@ + + * + * 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\Tests\Integration; + +use Tarantool\PhpUnit\Attribute\Lua; +use Tarantool\Queue\Queue; +use Tarantool\Queue\States; + +final class LimfifottlQueueTest extends TestCase +{ + #[Lua("create_tube('%tube_name%', '%tube_type%', {capacity = 1})")] + public function testPutReturnsNullWhenCapacityIsReached() : void + { + $queue = Queue::limfifottl($this->getClient(), $this->getQueueName()); + + self::assertTask($queue->put('first'), 0, States::READY, 'first'); + self::assertNull($queue->put('second', timeout: 0.0)); + } +} diff --git a/tests/Integration/SubqueuettlQueueTest.php b/tests/Integration/SubqueuettlQueueTest.php new file mode 100644 index 0000000..e82966b --- /dev/null +++ b/tests/Integration/SubqueuettlQueueTest.php @@ -0,0 +1,41 @@ + + * + * 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\Tests\Integration; + +use Tarantool\PhpUnit\Attribute\Lua; +use Tarantool\Queue\Queue; +use Tarantool\Queue\States; + +final class SubqueuettlQueueTest extends TestCase +{ + #[Lua("tube = create_tube('%tube_name%', '%tube_type%')")] + public function testTakeFromNewSubqueueReturnsNull() : void + { + $subqueue = Queue::subqueuettl($this->getClient(), $this->getQueueName()) + ->subqueue('created-on-first-put'); + + self::assertNull($subqueue->take(0.0)); + } + + #[Lua("tube = create_tube('%tube_name%', '%tube_type%')")] + public function testSubqueueRoutesOperations() : void + { + $subqueue = Queue::subqueuettl($this->getClient(), $this->getQueueName()) + ->subqueue('emails'); + + 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'); + } +} diff --git a/tests/Unit/TypedQueueTest.php b/tests/Unit/TypedQueueTest.php new file mode 100644 index 0000000..7ec7c31 --- /dev/null +++ b/tests/Unit/TypedQueueTest.php @@ -0,0 +1,110 @@ + + * + * 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\Tests\Unit; + +use PHPUnit\Framework\TestCase; +use Tarantool\Client\Request\CallRequest; +use Tarantool\PhpUnit\Client\TestDoubleClient; +use Tarantool\PhpUnit\Client\TestDoubleFactory; +use Tarantool\Queue\FifoQueue; +use Tarantool\Queue\FifottlQueue; +use Tarantool\Queue\LimfifottlQueue; +use Tarantool\Queue\Queue; +use Tarantool\Queue\Subqueue; +use Tarantool\Queue\SubqueuettlQueue; +use Tarantool\Queue\Task; +use Tarantool\Queue\UtubeQueue; +use Tarantool\Queue\UtubettlQueue; + +final class TypedQueueTest extends TestCase +{ + use TestDoubleClient; + + public function testFactoriesReturnDriverSpecificQueueTypes() : void + { + $client = $this->createDummyClient(); + + self::assertInstanceOf(FifoQueue::class, Queue::fifo($client, 'fifo')); + self::assertInstanceOf(FifottlQueue::class, Queue::fifottl($client, 'fifottl')); + self::assertInstanceOf(UtubeQueue::class, Queue::utube($client, 'utube')); + self::assertInstanceOf(UtubettlQueue::class, Queue::utubettl($client, 'utubettl')); + self::assertInstanceOf(LimfifottlQueue::class, Queue::limfifottl($client, 'limfifottl')); + self::assertInstanceOf(SubqueuettlQueue::class, Queue::subqueuettl($client, 'subqueuettl')); + } + + public function testFifottlPutSendsNamedOptions() : void + { + $client = $this->getTestDoubleClientBuilder() + ->shouldSend(new CallRequest('queue.tube.emails:put', ['payload', [ + 'pri' => 2, + 'ttl' => 60.0, + 'ttr' => 10.0, + 'delay' => 5.0, + ]])) + ->willReceive(TestDoubleFactory::createResponseFromData([[7, 'r', 'payload']])) + ->build(); + + $task = Queue::fifottl($client, 'emails')->put( + 'payload', + pri: 2, + ttl: 60.0, + ttr: 10.0, + delay: 5.0, + ); + + self::assertSame(7, $task->getId()); + self::assertSame('payload', $task->getData()); + } + + public function testLimfifottlPutReturnsNullWhenQueueIsFull() : void + { + $client = $this->getTestDoubleClientBuilder() + ->shouldSend(new CallRequest('queue.tube.bounded:put', ['payload', ['timeout' => 0.0]])) + ->willReceive(TestDoubleFactory::createEmptyResponse()) + ->build(); + + self::assertNull(Queue::limfifottl($client, 'bounded')->put('payload', timeout: 0.0)); + } + + public function testGenericQueuePutCanReturnNull() : void + { + $client = $this->getTestDoubleClientBuilder() + ->shouldSend(new CallRequest('queue.tube.bounded:put', ['payload', ['timeout' => 0.0]])) + ->willReceive(TestDoubleFactory::createEmptyResponse()) + ->build(); + + self::assertNull((new Queue($client, 'bounded'))->put('payload', ['timeout' => 0.0])); + } + + public function testSubqueueBindsContextToOperations() : void + { + $taskTuple = [[7, 'r', 'payload']]; + $response = TestDoubleFactory::createResponseFromData($taskTuple); + $client = $this->getTestDoubleClientBuilder() + ->shouldSend( + new CallRequest('queue.tube.jobs:put', ['payload', ['subqueue' => 'emails', 'ttl' => 60.0]]), + new CallRequest('queue.tube.jobs:take', [2.0, ['subqueue' => 'emails']]), + new CallRequest('queue.tube.jobs:ack', [7, ['subqueue' => 'emails']]), + ) + ->willReceive($response, $response, $response) + ->build(); + + $subqueue = Queue::subqueuettl($client, 'jobs')->subqueue('emails'); + + self::assertInstanceOf(Subqueue::class, $subqueue); + self::assertInstanceOf(Task::class, $subqueue->put('payload', ttl: 60.0)); + self::assertInstanceOf(Task::class, $subqueue->take(timeout: 2.0)); + self::assertInstanceOf(Task::class, $subqueue->ack(7)); + } +} From 1d960ae41484bbd6237998cac95bda078f0eddd9 Mon Sep 17 00:00:00 2001 From: Eugene Leonovich Date: Fri, 23 Oct 2026 14:46:20 +0200 Subject: [PATCH 2/2] test: cover typed queue methods --- tests/Unit/TypedQueueTest.php | 115 +++++++++++++++++++++++++++++++++- 1 file changed, 114 insertions(+), 1 deletion(-) diff --git a/tests/Unit/TypedQueueTest.php b/tests/Unit/TypedQueueTest.php index 7ec7c31..460796f 100644 --- a/tests/Unit/TypedQueueTest.php +++ b/tests/Unit/TypedQueueTest.php @@ -67,6 +67,83 @@ public function testFifottlPutSendsNamedOptions() : void self::assertSame('payload', $task->getData()); } + public function testFifottlReleaseSendsOptionalDelay() : void + { + $taskResponse = TestDoubleFactory::createResponseFromData([[1, 'r', 'payload']]); + $client = $this->getTestDoubleClientBuilder() + ->shouldSend( + new CallRequest('queue.tube.emails:release', [1, ['delay' => 5.0]]), + new CallRequest('queue.tube.emails:release', [2, []]), + ) + ->willReceive($taskResponse, $taskResponse) + ->build(); + $queue = Queue::fifottl($client, 'emails'); + + self::assertInstanceOf(Task::class, $queue->release(1, delay: 5.0)); + self::assertInstanceOf(Task::class, $queue->release(2)); + } + + public function testFifoPutAndReleaseUseOptionlessCalls() : void + { + $taskResponse = TestDoubleFactory::createResponseFromData([[1, 'r', 'payload']]); + $client = $this->getTestDoubleClientBuilder() + ->shouldSend( + new CallRequest('queue.tube.events:put', ['payload']), + new CallRequest('queue.tube.events:release', [1]), + ) + ->willReceive($taskResponse, $taskResponse) + ->build(); + $queue = Queue::fifo($client, 'events'); + + self::assertInstanceOf(Task::class, $queue->put('payload')); + self::assertInstanceOf(Task::class, $queue->release(1)); + } + + public function testUtubePutAndReleasePassSubqueueOptions() : void + { + $taskResponse = TestDoubleFactory::createResponseFromData([[1, 'r', 'payload']]); + $client = $this->getTestDoubleClientBuilder() + ->shouldSend( + new CallRequest('queue.tube.events:put', ['payload', ['utube' => 'domain-a']]), + new CallRequest('queue.tube.events:release', [1, []]), + ) + ->willReceive($taskResponse, $taskResponse) + ->build(); + $queue = Queue::utube($client, 'events'); + + self::assertInstanceOf(Task::class, $queue->put('payload', 'domain-a')); + self::assertInstanceOf(Task::class, $queue->release(1)); + } + + public function testUtubettlPutAndReleasePassTypedOptions() : void + { + $taskResponse = TestDoubleFactory::createResponseFromData([[1, 'r', 'payload']]); + $client = $this->getTestDoubleClientBuilder() + ->shouldSend( + new CallRequest('queue.tube.events:put', ['payload', [ + 'utube' => 'domain-a', + 'pri' => 2, + 'ttl' => 60.0, + 'ttr' => 10.0, + 'delay' => 5.0, + ]]), + new CallRequest('queue.tube.events:release', [1, ['delay' => 5.0]]), + ) + ->willReceive($taskResponse, $taskResponse) + ->build(); + $queue = Queue::utubettl($client, 'events'); + + self::assertInstanceOf(Task::class, $queue->put( + 'payload', + 'domain-a', + pri: 2, + ttl: 60.0, + ttr: 10.0, + delay: 5.0, + )); + self::assertInstanceOf(Task::class, $queue->release(1, delay: 5.0)); + } + public function testLimfifottlPutReturnsNullWhenQueueIsFull() : void { $client = $this->getTestDoubleClientBuilder() @@ -87,6 +164,22 @@ public function testGenericQueuePutCanReturnNull() : void self::assertNull((new Queue($client, 'bounded'))->put('payload', ['timeout' => 0.0])); } + public function testLimfifottlReleasePassesOptionalDelay() : void + { + $taskResponse = TestDoubleFactory::createResponseFromData([[1, 'r', 'payload']]); + $client = $this->getTestDoubleClientBuilder() + ->shouldSend( + new CallRequest('queue.tube.bounded:release', [1, ['delay' => 5.0]]), + new CallRequest('queue.tube.bounded:release', [2, []]), + ) + ->willReceive($taskResponse, $taskResponse) + ->build(); + $queue = Queue::limfifottl($client, 'bounded'); + + self::assertInstanceOf(Task::class, $queue->release(1, delay: 5.0)); + self::assertInstanceOf(Task::class, $queue->release(2)); + } + public function testSubqueueBindsContextToOperations() : void { $taskTuple = [[7, 'r', 'payload']]; @@ -96,8 +189,9 @@ public function testSubqueueBindsContextToOperations() : void new CallRequest('queue.tube.jobs:put', ['payload', ['subqueue' => 'emails', 'ttl' => 60.0]]), new CallRequest('queue.tube.jobs:take', [2.0, ['subqueue' => 'emails']]), new CallRequest('queue.tube.jobs:ack', [7, ['subqueue' => 'emails']]), + new CallRequest('queue.tube.jobs:touch', [7, 1.0, ['subqueue' => 'emails']]), ) - ->willReceive($response, $response, $response) + ->willReceive($response, $response, $response, $response) ->build(); $subqueue = Queue::subqueuettl($client, 'jobs')->subqueue('emails'); @@ -106,5 +200,24 @@ public function testSubqueueBindsContextToOperations() : void self::assertInstanceOf(Task::class, $subqueue->put('payload', ttl: 60.0)); self::assertInstanceOf(Task::class, $subqueue->take(timeout: 2.0)); self::assertInstanceOf(Task::class, $subqueue->ack(7)); + self::assertInstanceOf(Task::class, $subqueue->touch(7, 1.0)); + } + + public function testSubqueueReleaseKeepsContextAndAddsOptionalDelay() : void + { + $taskResponse = TestDoubleFactory::createResponseFromData([[7, 'r', 'payload']]); + $client = $this->getTestDoubleClientBuilder() + ->shouldSend( + new CallRequest('queue.tube.jobs:release', [7, ['subqueue' => 'emails', 'delay' => 5.0]]), + new CallRequest('queue.tube.jobs:release', [8, ['subqueue' => 'emails']]), + new CallRequest('queue.tube.jobs:truncate', [['subqueue' => 'emails']]), + ) + ->willReceive($taskResponse, $taskResponse, TestDoubleFactory::createEmptyResponse()) + ->build(); + $subqueue = Queue::subqueuettl($client, 'jobs')->subqueue('emails'); + + self::assertInstanceOf(Task::class, $subqueue->release(7, delay: 5.0)); + self::assertInstanceOf(Task::class, $subqueue->release(8)); + $subqueue->truncate(); } }