From f9a41342d27d016aa263268e26ba5ae31276246d Mon Sep 17 00:00:00 2001 From: Alexander Makarov Date: Thu, 9 Jul 2026 10:50:23 +0300 Subject: [PATCH 1/2] Update for queue message API changes --- composer.json | 8 ++- src/Adapter.php | 48 ++++++++++------ tests/AdapterTest.php | 126 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 162 insertions(+), 20 deletions(-) create mode 100644 tests/AdapterTest.php diff --git a/composer.json b/composer.json index d5ad2ef..c221a8a 100644 --- a/composer.json +++ b/composer.json @@ -33,7 +33,7 @@ "prefer-stable": true, "require": { "php": "^8.1", - "yiisoft/db": "^1.0", + "yiisoft/db": "^2.0", "yiisoft/mutex": "^1.0", "yiisoft/queue": "dev-master" }, @@ -43,7 +43,9 @@ "rector/rector": "^2.0.3", "roave/infection-static-analysis-plugin": "^1.34", "spatie/phpunit-watcher": "^1.23", - "vimeo/psalm": "^5.20" + "vimeo/psalm": "^5.20", + "yiisoft/cache": "^3.2", + "yiisoft/db-sqlite": "^2.0" }, "suggest": { "yiisoft/db": "For using with Yii Database", @@ -75,7 +77,7 @@ } }, "scripts": { - "test": "phpunit --testdox --no-interaction", + "test": "phpunit --testdox", "test-watch": "phpunit-watcher watch" } } diff --git a/src/Adapter.php b/src/Adapter.php index 6189ce5..bfc167f 100644 --- a/src/Adapter.php +++ b/src/Adapter.php @@ -4,18 +4,21 @@ namespace Yiisoft\Queue\Db; +use BackedEnum; use InvalidArgumentException; +use RuntimeException; use Yiisoft\Queue\Adapter\AdapterInterface; use Yiisoft\Queue\Cli\LoopInterface; -use Yiisoft\Queue\Enum\JobStatus; +use Yiisoft\Queue\Message\DelayEnvelope; use Yiisoft\Queue\Message\MessageInterface; -use Yiisoft\Queue\Message\MessageSerializerInterface; -use Yiisoft\Queue\QueueFactory; +use Yiisoft\Queue\Message\Serializer\MessageSerializerInterface; +use Yiisoft\Queue\MessageStatus; use Yiisoft\Queue\Message\IdEnvelope; use Yiisoft\Db\Connection\ConnectionInterface; use Yiisoft\Db\Query\Query; use Yiisoft\Mutex\MutexFactoryInterface; use Yiisoft\Mutex\MutexInterface; +use Yiisoft\Queue\Provider\QueueProviderInterface; final class Adapter implements AdapterInterface { @@ -41,7 +44,7 @@ public function __construct( private MessageSerializerInterface $serializer, private LoopInterface $loop, private MutexFactoryInterface $mutexFactory, - private string $channel = QueueFactory::DEFAULT_CHANNEL_NAME, + private string $channel = QueueProviderInterface::DEFAULT_QUEUE, ) { $this->mutex = $this->mutexFactory->create(self::class . $this->channel); } @@ -51,7 +54,7 @@ public function runExisting(callable $handlerCallback): void $this->run($handlerCallback, false); } - public function status(string|int $id): JobStatus + public function status(string|int $id): MessageStatus { $id = (int) $id; @@ -60,35 +63,39 @@ public function status(string|int $id): JobStatus ->where(['id' => $id]) ->one(); - if (!$payload) { + if ($payload === null) { if ($this->deleteReleased) { - return JobStatus::done(); + return MessageStatus::DONE; } throw new InvalidArgumentException("Unknown message ID: $id."); } + if (!is_array($payload)) { + throw new RuntimeException('Queue payload must be an array.'); + } + if (!$payload['reserved_at']) { - return JobStatus::waiting(); + return MessageStatus::WAITING; } if (!$payload['done_at']) { - return JobStatus::reserved(); + return MessageStatus::RESERVED; } - return JobStatus::done(); + return MessageStatus::DONE; } public function push(MessageInterface $message): MessageInterface { - $metadata = $message->getMetadata(); + $meta = $message->getMeta(); $this->db->createCommand()->insert($this->tableName, [ 'channel' => $this->channel, 'job' => $this->serializer->serialize($message), 'pushed_at' => time(), - 'ttr' => $metadata['ttr'] ?? 300, - 'delay' => $metadata['delay'] ?? 0, - 'priority' => $metadata['priority'] ?? 1024, + 'ttr' => $meta['ttr'] ?? 300, + 'delay' => $meta[DelayEnvelope::META_DELAY_SECONDS] ?? 0, + 'priority' => $meta['priority'] ?? 1024, ])->execute(); $tableSchema = $this->db->getTableSchema($this->tableName); $key = $tableSchema ? $this->db->getLastInsertID($tableSchema->getSequenceName()) : $tableSchema; @@ -101,15 +108,17 @@ public function subscribe(callable $handlerCallback): void $this->run($handlerCallback, true, 5); // TWK TODO timeout should not be hard coded } - public function withChannel(string $channel): self + public function withChannel(BackedEnum|string $channel): self { + $channel = is_string($channel) ? $channel : (string) $channel->value; + if ($channel === $this->channel) { return $this; } $new = clone $this; $new->channel = $channel; - $new->mutex = $this->mutexFactory->create(self::class . $this->channel); + $new->mutex = $this->mutexFactory->create(self::class . $new->channel); return $new; } @@ -138,7 +147,11 @@ protected function reserve(): array|null ->orderBy(['priority' => SORT_ASC, 'id' => SORT_ASC]) ->limit(1) ->one(); - if (is_array($payload)) { + if ($payload !== null && !is_array($payload)) { + throw new RuntimeException('Queue payload must be an array.'); + } + + if ($payload !== null) { $payload['reserved_at'] = time(); $payload['attempt'] = (int) $payload['attempt'] + 1; $this->db->createCommand()->update($this->tableName, [ @@ -191,6 +204,7 @@ private function moveExpired(): void $this->tableName, ['reserved_at' => null], '[[reserved_at]] < :time - [[ttr]] and [[reserved_at]] is not null and [[done_at]] is null', + null, [':time' => $this->reserveTime] )->execute(); } diff --git a/tests/AdapterTest.php b/tests/AdapterTest.php new file mode 100644 index 0000000..34b0f7f --- /dev/null +++ b/tests/AdapterTest.php @@ -0,0 +1,126 @@ +createMock(ConnectionInterface::class); + $command = $this->createMock(CommandInterface::class); + $serializer = $this->createMock(MessageSerializerInterface::class); + $tableSchema = $this->createMock(TableSchemaInterface::class); + + $message = new DelayEnvelope( + (new GenericMessage('test', 'payload'))->withMeta(['ttr' => 10, 'priority' => 20]), + 5, + ); + + $serializer + ->expects(self::once()) + ->method('serialize') + ->with($message) + ->willReturn('serialized-message'); + + $db + ->expects(self::once()) + ->method('createCommand') + ->willReturn($command); + + $command + ->expects(self::once()) + ->method('insert') + ->with('{{%queue}}', self::callback(static function (array $row): bool { + return $row['channel'] === 'yii-queue' + && $row['job'] === 'serialized-message' + && $row['ttr'] === 10 + && $row['delay'] === 5.0 + && $row['priority'] === 20; + })) + ->willReturnSelf(); + + $command + ->expects(self::once()) + ->method('execute') + ->willReturn(1); + + $db + ->expects(self::once()) + ->method('getTableSchema') + ->with('{{%queue}}') + ->willReturn($tableSchema); + + $tableSchema + ->expects(self::once()) + ->method('getSequenceName') + ->willReturn('queue_id_seq'); + + $db + ->expects(self::once()) + ->method('getLastInsertID') + ->with('queue_id_seq') + ->willReturn('42'); + + $result = $this->createAdapter($db, $serializer)->push($message); + + self::assertSame('42', IdEnvelope::fromMessage($result)->getId()); + } + + public function testWithChannelClonesAdapterAndUsesNewChannelMutex(): void + { + $mutexFactory = $this->createMock(MutexFactoryInterface::class); + $mutexNames = []; + $mutexFactory + ->expects(self::exactly(2)) + ->method('create') + ->with(self::callback(static function (string $name) use (&$mutexNames): bool { + $mutexNames[] = $name; + return in_array($name, [Adapter::class . 'yii-queue', Adapter::class . 'mail'], true); + })) + ->willReturn($this->createMock(MutexInterface::class)); + + $adapter = $this->createAdapter( + mutexFactory: $mutexFactory, + ); + + $new = $adapter->withChannel('mail'); + + self::assertNotSame($adapter, $new); + self::assertSame([Adapter::class . 'yii-queue', Adapter::class . 'mail'], $mutexNames); + } + + private function createAdapter( + ?ConnectionInterface $db = null, + ?MessageSerializerInterface $serializer = null, + ?MutexFactoryInterface $mutexFactory = null, + ): Adapter { + if ($mutexFactory === null) { + $mutexFactory = $this->createMock(MutexFactoryInterface::class); + $mutexFactory + ->method('create') + ->willReturn($this->createMock(MutexInterface::class)); + } + + return new Adapter( + $db ?? $this->createMock(ConnectionInterface::class), + $serializer ?? $this->createMock(MessageSerializerInterface::class), + $this->createMock(LoopInterface::class), + $mutexFactory, + ); + } +} From b17558e96b0b30079b1be0e52a67ba1f1b922777 Mon Sep 17 00:00:00 2001 From: Alexander Makarov Date: Thu, 9 Jul 2026 11:17:39 +0300 Subject: [PATCH 2/2] Address review feedback --- tests/AdapterTest.php | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/tests/AdapterTest.php b/tests/AdapterTest.php index 34b0f7f..211c7cc 100644 --- a/tests/AdapterTest.php +++ b/tests/AdapterTest.php @@ -16,6 +16,7 @@ use Yiisoft\Queue\Message\GenericMessage; use Yiisoft\Queue\Message\IdEnvelope; use Yiisoft\Queue\Message\Serializer\MessageSerializerInterface; +use Yiisoft\Queue\Provider\QueueProviderInterface; final class AdapterTest extends TestCase { @@ -46,7 +47,7 @@ public function testPushUsesMessageMetaAndReturnsIdEnvelope(): void ->expects(self::once()) ->method('insert') ->with('{{%queue}}', self::callback(static function (array $row): bool { - return $row['channel'] === 'yii-queue' + return $row['channel'] === QueueProviderInterface::DEFAULT_QUEUE && $row['job'] === 'serialized-message' && $row['ttr'] === 10 && $row['delay'] === 5.0 @@ -90,7 +91,7 @@ public function testWithChannelClonesAdapterAndUsesNewChannelMutex(): void ->method('create') ->with(self::callback(static function (string $name) use (&$mutexNames): bool { $mutexNames[] = $name; - return in_array($name, [Adapter::class . 'yii-queue', Adapter::class . 'mail'], true); + return in_array($name, [Adapter::class . QueueProviderInterface::DEFAULT_QUEUE, Adapter::class . 'mail'], true); })) ->willReturn($this->createMock(MutexInterface::class)); @@ -101,7 +102,7 @@ public function testWithChannelClonesAdapterAndUsesNewChannelMutex(): void $new = $adapter->withChannel('mail'); self::assertNotSame($adapter, $new); - self::assertSame([Adapter::class . 'yii-queue', Adapter::class . 'mail'], $mutexNames); + self::assertSame([Adapter::class . QueueProviderInterface::DEFAULT_QUEUE, Adapter::class . 'mail'], $mutexNames); } private function createAdapter(