Update for queue message API changes

src/Adapter.php

@@ -9,7 +9,7 @@
 use Yiisoft\Queue\Cli\LoopInterface;
 use Yiisoft\Queue\Message\IdEnvelope;
 use Yiisoft\Queue\Message\MessageInterface;
-use Yiisoft\Queue\Message\MessageSerializerInterface;
+use Yiisoft\Queue\Message\Serializer\MessageSerializerInterface;
 use Yiisoft\Queue\MessageStatus;
 
 final class Adapter implements AdapterInterface
@@ -63,7 +63,7 @@ public function status(int|string $id): MessageStatus
     public function push(MessageInterface $message): MessageInterface
     {
         $payload = $this->serializer->serialize($message);
-        $id = $this->provider->pushMessage($payload, $message->getMetadata());
+        $id = $this->provider->pushMessage($payload, $message->getMeta());
         return new IdEnvelope($message, $id);
     }
 

src/Message/Message.php

@@ -4,28 +4,34 @@
 
 namespace Yiisoft\Queue\Redis\Message;
 
+use Yiisoft\Queue\Message\DelayEnvelope;
 use Yiisoft\Queue\Message\MessageInterface;
 
+/**
+ * @psalm-import-type MessagePayload from MessageInterface
+ * @psalm-import-type MessageMeta from MessageInterface
+ */
 final class Message implements MessageInterface
 {
     /**
-     * @param array<string, mixed> $metadata
+     * @psalm-param MessagePayload $payload
+     * @psalm-param MessageMeta $meta
      */
     public function __construct(
         private string $handlerName,
-        private mixed $data,
-        private array $metadata,
+        private bool|int|float|string|array|null $payload,
+        private array $meta,
         private int $delay = 0 //delay in seconds
     ) {
         if ($this->delay > 0) {
-            $this->metadata['delay'] = $delay;
+            $this->meta[DelayEnvelope::META_DELAY_SECONDS] = $delay;
         }
     }
 
     public function withDelay(int $delay): self
     {
         $message = clone $this;
-        $message->metadata['delay'] = $delay;
+        $message->meta[DelayEnvelope::META_DELAY_SECONDS] = $delay;
         return $message;
     }
 
@@ -39,31 +45,37 @@ public function getType(): string
         return $this->handlerName;
     }
 
-    public function getData(): mixed
+    /**
+     * @psalm-return MessagePayload
+     */
+    public function getPayload(): bool|int|float|string|array|null
     {
-        return $this->data;
+        return $this->payload;
     }
 
     /**
-     * @return array<string, mixed>
+     * @psalm-return MessageMeta
      */
-    public function getMetadata(): array
+    public function getMeta(): array
     {
-        return $this->metadata;
+        return $this->meta;
     }
 
     /**
-     * @param array<string, mixed> $metadata
+     * @psalm-param MessageMeta $meta
      */
-    public function withMetadata(array $metadata): static
+    public function withMeta(array $meta): static
     {
         $message = clone $this;
-        $message->metadata = $metadata;
+        $message->meta = $meta;
         return $message;
     }
 
-    public static function fromData(string $type, mixed $data): self
+    /**
+     * @psalm-param MessagePayload $payload
+     */
+    public static function fromPayload(string $type, bool|int|float|string|array|null $payload): static
     {
-        return new self($type, $data, []);
+        return new self($type, $payload, []);
     }
 }

src/QueueProvider.php

@@ -5,6 +5,7 @@
 namespace Yiisoft\Queue\Redis;
 
 use RedisException;
+use Yiisoft\Queue\Message\DelayEnvelope;
 use Yiisoft\Queue\Redis\Exception\NotConnectedRedisException;
 
 class QueueProvider implements QueueProviderInterface
@@ -26,15 +27,19 @@ public function __construct(
     }
 
     /**
+     * @param array<string, array|bool|float|int|string|null> $meta
+     *
      * @throws RedisException
      */
-    public function pushMessage(string $message, array $metadata = []): int
+    public function pushMessage(string $message, array $meta = []): int
     {
         $this->checkConnection();
         $id = $this->getId();
         $this->redis->hset("$this->channelName.messages", (string) $id, $message);
 
-        $delay = isset($metadata['delay']) && is_int($metadata['delay']) ? $metadata['delay'] : 0;
+        /** @var array|bool|float|int|string|null $delay */
+        $delay = $meta[DelayEnvelope::META_DELAY_SECONDS] ?? 0;
+        $delay = is_int($delay) || is_float($delay) ? $delay : 0;
         if ($delay > 0) {
             $this->redis->zadd("$this->channelName.delayed", time() + $delay, $id);
         } else {

src/QueueProviderInterface.php

@@ -6,7 +6,10 @@
 
 interface QueueProviderInterface
 {
-    public function pushMessage(string $message, array $metadata = []): int;
+    /**
+     * @param array<string, array|bool|float|int|string|null> $meta
+     */
+    public function pushMessage(string $message, array $meta = []): int;
 
     /**
      * @return Reserve|null Payload and id, or null if queue is empty.

tests/Integration/QueueProviderTest.php

@@ -36,7 +36,7 @@ public function test__construct(): QueueProvider
     public function testDelay(QueueProvider $provider): void
     {
         $message = new Message('test', ['key' => 'value'], [], 2);
-        $id = $provider->pushMessage(json_encode($message->getData(), JSON_THROW_ON_ERROR), $message->getMetadata());
+        $id = $provider->pushMessage(json_encode($message->getPayload(), JSON_THROW_ON_ERROR), $message->getMeta());
         $this->assertGreaterThan(0, $id);
         $reserv = $provider->reserve($id);
         $this->assertNull($reserv);

tests/Integration/QueueTest.php

@@ -6,10 +6,11 @@
 
 use Yiisoft\Queue\Adapter\AdapterInterface;
 use Yiisoft\Queue\Cli\LoopInterface;
-use Yiisoft\Queue\Message\JsonMessageSerializer;
+use Yiisoft\Queue\Message\Serializer\JsonMessageEncoder;
+use Yiisoft\Queue\Message\Serializer\MessageSerializer;
 use Yiisoft\Queue\Message\GenericMessage as Message;
 use Yiisoft\Queue\Message\MessageInterface;
-use Yiisoft\Queue\Message\MessageSerializerInterface;
+use Yiisoft\Queue\Message\Serializer\MessageSerializerInterface;
 use Yiisoft\Queue\MessageStatus;
 use Yiisoft\Queue\Queue;
 use Yiisoft\Queue\Redis\Adapter;
@@ -65,7 +66,7 @@ public function testStatus(): void
 
         $mockReserved = $this->createMock(QueueProviderInterface::class);
         $mockReserved->method('existInReserved')->willReturn(true);
-        $adapter = new Adapter($mockReserved, new JsonMessageSerializer(), $this->getLoop());
+        $adapter = new Adapter($mockReserved, new MessageSerializer(new JsonMessageEncoder()), $this->getLoop());
 
         $status = $adapter->status('1');
         $this->assertEquals(MessageStatus::RESERVED, $status);
@@ -83,7 +84,7 @@ public function testListen(): void
         $adapter = new Adapter(
             $queueProvider
                 ->withChannelName('yii-queue'),
-            new JsonMessageSerializer(),
+            new MessageSerializer(new JsonMessageEncoder()),
             $mockLoop,
         );
         $queue = $this->getDefaultQueue($adapter);
@@ -131,7 +132,7 @@ public function testAdapterNullMessage()
 
         $adapter = new Adapter(
             $provider,
-            new JsonMessageSerializer(),
+            new MessageSerializer(new JsonMessageEncoder()),
             $mockLoop,
         );
         $notUseHandler = true;

tests/Support/ExtendedSimpleMessageHandler.php

@@ -17,8 +17,13 @@ public function __construct(private FileHelper $fileHelper)
 
     public function handle(MessageInterface $message): void
     {
-        $data = $message->getData();
-        if (null !== $data) {
+        $data = $message->getPayload();
+        if (
+            is_array($data)
+            && is_string($data['file_name'] ?? null)
+            && is_array($data['payload'] ?? null)
+            && (is_int($data['payload']['time'] ?? null) || is_string($data['payload']['time'] ?? null))
+        ) {
             $this->fileHelper->put($data['file_name'], $data['payload']['time']);
         }
     }

tests/Support/IntegrationTestCase.php

@@ -11,7 +11,8 @@
 use Yiisoft\Queue\Adapter\AdapterInterface;
 use Yiisoft\Queue\Cli\LoopInterface;
 use Yiisoft\Queue\Cli\SignalLoop;
-use Yiisoft\Queue\Message\JsonMessageSerializer;
+use Yiisoft\Queue\Message\Serializer\JsonMessageEncoder;
+use Yiisoft\Queue\Message\Serializer\MessageSerializer;
 use Yiisoft\Queue\Message\MessageInterface;
 use Yiisoft\Queue\Middleware\CallableFactory;
 use Yiisoft\Queue\Middleware\Consume\ConsumeMiddlewareDispatcher;
@@ -91,8 +92,12 @@ protected function getMessageHandlers(): array
         return [
             'ext-simple' => [new ExtendedSimpleMessageHandler(new FileHelper()), 'handle'],
             'exception-listen' => static function (MessageInterface $message) {
-                $data = $message->getData();
-                if (null !== $data) {
+                $data = $message->getPayload();
+                if (
+                    is_array($data)
+                    && is_array($data['payload'] ?? null)
+                    && (is_int($data['payload']['time'] ?? null) || is_string($data['payload']['time'] ?? null))
+                ) {
                     throw new \RuntimeException((string) $data['payload']['time']);
                 }
             },
@@ -134,7 +139,7 @@ protected function getAdapter(): AdapterInterface
     {
         return $this->adapter ??= new Adapter(
             $this->getQueueProvider(),
-            new JsonMessageSerializer(),
+            new MessageSerializer(new JsonMessageEncoder()),
             $this->getLoop(),
         );
     }

tests/Unit/Message/MessageTest.php

@@ -5,6 +5,7 @@
 namespace Unit\Message;
 
 use PHPUnit\Framework\TestCase;
+use Yiisoft\Queue\Message\DelayEnvelope;
 use Yiisoft\Queue\Redis\Message\Message;
 
 class MessageTest extends TestCase
@@ -16,31 +17,31 @@ public function testGetHandlerName(): void
         $this->assertEquals('handler', $message->getType());
     }
 
-    public function testGetData(): void
+    public function testGetPayload(): void
     {
         $message = new Message('handler', 'data', []);
-        $this->assertEquals('data', $message->getData());
+        $this->assertEquals('data', $message->getPayload());
     }
 
-    public function testGetMetadata(): void
+    public function testGetMeta(): void
     {
-        $metadata = ['key' => 'value'];
-        $message = new Message('handler', 'data', $metadata);
-        $this->assertEquals($metadata, $message->getMetadata());
+        $meta = ['key' => 'value'];
+        $message = new Message('handler', 'data', $meta);
+        $this->assertEquals($meta, $message->getMeta());
 
-        $message = new Message('handler', 'data', $metadata, 2);
-        $metadata['delay'] = 2;
-        $this->assertEquals($metadata, $message->getMetadata());
+        $message = new Message('handler', 'data', $meta, 2);
+        $meta[DelayEnvelope::META_DELAY_SECONDS] = 2;
+        $this->assertEquals($meta, $message->getMeta());
     }
 
-    public function testWithMetadata(): void
+    public function testWithMeta(): void
     {
         $message = new Message('handler', 'data', []);
-        $messageWithMetadata = $message->withMetadata(['key' => 'value']);
+        $messageWithMeta = $message->withMeta(['key' => 'value']);
 
-        $this->assertNotSame($message, $messageWithMetadata);
-        $this->assertSame([], $message->getMetadata());
-        $this->assertSame(['key' => 'value'], $messageWithMetadata->getMetadata());
+        $this->assertNotSame($message, $messageWithMeta);
+        $this->assertSame([], $message->getMeta());
+        $this->assertSame(['key' => 'value'], $messageWithMeta->getMeta());
     }
 
     public function testWithDelay(): void
@@ -49,19 +50,19 @@ public function testWithDelay(): void
         $delayedMessage = $message->withDelay(5);
 
         $this->assertNotSame($message, $delayedMessage);
-        $this->assertEquals(5, $delayedMessage->getMetadata()['delay']);
+        $this->assertEquals(5, $delayedMessage->getMeta()[DelayEnvelope::META_DELAY_SECONDS]);
     }
 
-    public function testFromData(): void
+    public function testFromPayload(): void
     {
         $handlerName = 'test-handler';
         $data = ['key' => 'value'];
 
-        $message = Message::fromData($handlerName, $data);
+        $message = Message::fromPayload($handlerName, $data);
 
         $this->assertInstanceOf(Message::class, $message);
         $this->assertEquals($handlerName, $message->getHandlerName());
-        $this->assertEquals($data, $message->getData());
-        $this->assertEquals([], $message->getMetadata());
+        $this->assertEquals($data, $message->getPayload());
+        $this->assertEquals([], $message->getMeta());
     }
 }

tests/Unit/QueueTest.php

@@ -6,10 +6,11 @@
 
 use PHPUnit\Framework\TestCase;
 use Yiisoft\Queue\Cli\LoopInterface;
-use Yiisoft\Queue\Message\JsonMessageSerializer;
+use Yiisoft\Queue\Message\Serializer\JsonMessageEncoder;
+use Yiisoft\Queue\Message\Serializer\MessageSerializer;
 use Yiisoft\Queue\Message\Message;
 use Yiisoft\Queue\Message\MessageInterface;
-use Yiisoft\Queue\Message\MessageSerializerInterface;
+use Yiisoft\Queue\Message\Serializer\MessageSerializerInterface;
 use Yiisoft\Queue\Redis\Adapter;
 use Yiisoft\Queue\Redis\QueueProviderInterface;
 
@@ -37,7 +38,7 @@ public function testAdapterNullMessage()
 
         $adapter = new Adapter(
             $provider,
-            new JsonMessageSerializer(),
+            new MessageSerializer(new JsonMessageEncoder()),
             $mockLoop,
         );
         $notUseHandler = true;