Update for queue message API changes

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"
     }
 }

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

tests/AdapterTest.php

@@ -0,0 +1,127 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\Db\Tests;
+
+use PHPUnit\Framework\TestCase;
+use Yiisoft\Db\Command\CommandInterface;
+use Yiisoft\Db\Connection\ConnectionInterface;
+use Yiisoft\Db\Schema\TableSchemaInterface;
+use Yiisoft\Mutex\MutexFactoryInterface;
+use Yiisoft\Mutex\MutexInterface;
+use Yiisoft\Queue\Cli\LoopInterface;
+use Yiisoft\Queue\Db\Adapter;
+use Yiisoft\Queue\Message\DelayEnvelope;
+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
+{
+    public function testPushUsesMessageMetaAndReturnsIdEnvelope(): void
+    {
+        $db = $this->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'] === QueueProviderInterface::DEFAULT_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 . QueueProviderInterface::DEFAULT_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 . QueueProviderInterface::DEFAULT_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,
+        );
+    }
+}