Add envelope stack + MessageSerializer (#188)

config/di.php

@@ -8,6 +8,8 @@
 use Yiisoft\Queue\Cli\SimpleLoop;
 use Yiisoft\Queue\Command\ListenAllCommand;
 use Yiisoft\Queue\Command\RunCommand;
+use Yiisoft\Queue\Message\JsonMessageSerializer;
+use Yiisoft\Queue\Message\MessageSerializerInterface;
 use Yiisoft\Queue\Middleware\Consume\ConsumeMiddlewareDispatcher;
 use Yiisoft\Queue\Middleware\Consume\MiddlewareFactoryConsume;
 use Yiisoft\Queue\Middleware\Consume\MiddlewareFactoryConsumeInterface;
@@ -54,6 +56,7 @@
     FailureMiddlewareDispatcher::class => [
         '__construct()' => ['middlewareDefinitions' => $params['yiisoft/queue']['middlewares-fail']],
     ],
+    MessageSerializerInterface::class => JsonMessageSerializer::class,
     RunCommand::class => [
         '__construct()' => [
             'channels' => array_keys($params['yiisoft/yii-queue']['channel-definitions']),

src/Message/EnvelopeInterface.php

@@ -9,6 +9,10 @@
  */
 interface EnvelopeInterface extends MessageInterface
 {
+    public const ENVELOPE_STACK_KEY = 'envelopes';
+
+    public static function fromMessage(MessageInterface $message): self;
+
     public function getMessage(): MessageInterface;
 
     public function withMessage(MessageInterface $message): self;

src/Message/EnvelopeTrait.php

@@ -31,8 +31,27 @@ public function getData(): mixed
         return $this->message->getData();
     }
 
+    public static function fromMessage(MessageInterface $message): self
+    {
+        return new static($message);
+    }
+
     public function getMetadata(): array
     {
-        return $this->message->getMetadata();
+        return array_merge(
+            $this->message->getMetadata(),
+            [
+                self::ENVELOPE_STACK_KEY => array_merge(
+                    $this->message->getMetadata()[self::ENVELOPE_STACK_KEY] ?? [],
+                    [self::class],
+                ),
+            ],
+            $this->getEnvelopeMetadata(),
+        );
+    }
+
+    public function getEnvelopeMetadata(): array
+    {
+        return [];
     }
 }

src/Message/IdEnvelope.php

@@ -29,10 +29,8 @@ public function getId(): string|int|null
         return $this->id ?? $this->message->getMetadata()[self::MESSAGE_ID_KEY] ?? null;
     }
 
-    public function getMetadata(): array
+    private function getEnvelopeMetadata(): array
     {
-        return array_merge($this->message->getMetadata(), [
-            self::MESSAGE_ID_KEY => $this->getId(),
-        ]);
+        return [self::MESSAGE_ID_KEY => $this->getId()];
     }
 }

src/Message/JsonMessageSerializer.php

@@ -0,0 +1,59 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\Message;
+
+use InvalidArgumentException;
+use JsonException;
+
+final class JsonMessageSerializer implements MessageSerializerInterface
+{
+    /**
+     * @throws JsonException
+     */
+    public function serialize(MessageInterface $message): string
+    {
+        $payload = [
+            'name' => $message->getHandlerName(),
+            'data' => $message->getData(),
+            'meta' => $message->getMetadata(),
+        ];
+
+        return json_encode($payload, JSON_THROW_ON_ERROR);
+    }
+
+    /**
+     * @throws JsonException
+     * @throws InvalidArgumentException
+     */
+    public function unserialize(string $value): MessageInterface
+    {
+        $payload = json_decode($value, true, 512, JSON_THROW_ON_ERROR);
+        if (!is_array($payload)) {
+            throw new InvalidArgumentException('Payload must be array. Got ' . get_debug_type($payload) . '.');
+        }
+
+        $meta = $payload['meta'] ?? [];
+        if (!is_array($meta)) {
+            throw new InvalidArgumentException('Metadata must be array. Got ' . get_debug_type($meta) . '.');
+        }
+
+        // TODO: will be removed later
+        $message = new Message($payload['name'] ?? '$name', $payload['data'] ?? null, $meta);
+
+        if (isset($meta[EnvelopeInterface::ENVELOPE_STACK_KEY]) && is_array($meta[EnvelopeInterface::ENVELOPE_STACK_KEY])) {
+            $message = $message->withMetadata(
+                array_merge($message->getMetadata(), [EnvelopeInterface::ENVELOPE_STACK_KEY => []]),
+            );
+            foreach ($meta[EnvelopeInterface::ENVELOPE_STACK_KEY] as $envelope) {
+                if (is_string($envelope) && class_exists($envelope) && is_subclass_of($envelope, EnvelopeInterface::class)) {
+                    $message = $envelope::fromMessage($message);
+                }
+            }
+        }
+
+
+        return $message;
+    }
+}

src/Message/Message.php

@@ -32,4 +32,12 @@ public function getMetadata(): array
     {
         return $this->metadata;
     }
+
+    public function withMetadata(array $metadata): self
+    {
+        $instance = clone $this;
+        $instance->metadata = $metadata;
+
+        return $instance;
+    }
 }

src/Message/MessageSerializerInterface.php

@@ -0,0 +1,12 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\Message;
+
+interface MessageSerializerInterface
+{
+    public function serialize(MessageInterface $message): string;
+
+    public function unserialize(string $value): MessageInterface;
+}

src/Middleware/FailureHandling/FailureEnvelope.php

@@ -14,7 +14,7 @@ final class FailureEnvelope implements EnvelopeInterface
 
     public function __construct(
         private MessageInterface $message,
-        private array $meta,
+        private array $meta = [],
     ) {
     }
 

src/QueueInterface.php

@@ -19,18 +19,14 @@ interface QueueInterface
     /**
      * Pushes a message into the queue.
      *
-     * @param MessageInterface $message
      * @param array|callable|MiddlewarePushInterface|string ...$middlewareDefinitions
-     *
      * @return MessageInterface
      */
     public function push(MessageInterface $message, MiddlewarePushInterface|callable|array|string ...$middlewareDefinitions): MessageInterface;
 
     /**
      * Execute all existing jobs and exit
      *
-     * @param int $max
-     *
      * @return int How many messages were processed
      */
     public function run(int $max = 0): int;

src/Worker/Worker.php

@@ -51,7 +51,7 @@ public function process(MessageInterface $message, QueueInterface $queue): Messa
         $name = $message->getHandlerName();
         $handler = $this->getHandler($name);
         if ($handler === null) {
-            throw new RuntimeException("Queue handler with name $name doesn't exist");
+            throw new RuntimeException(sprintf('Queue handler with name "%s" does not exist', $name));
         }
 
         $request = new ConsumeRequest($message, $queue);

tests/Unit/EnvelopeTest.php

@@ -0,0 +1,47 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Unit;
+
+use PHPUnit\Framework\TestCase;
+use Yiisoft\Queue\Message\EnvelopeInterface;
+use Yiisoft\Queue\Message\IdEnvelope;
+use Yiisoft\Queue\Message\Message;
+
+final class EnvelopeTest extends TestCase
+{
+    public function testEnvelopeStack(): void
+    {
+        $message = new Message('handler', 'test');
+        $message = new IdEnvelope($message, 'test-id');
+
+        $this->assertEquals('test', $message->getMessage()->getData());
+
+        $stack = $message->getMetadata()[EnvelopeInterface::ENVELOPE_STACK_KEY];
+        $this->assertIsArray($stack);
+
+        $this->assertEquals([
+            IdEnvelope::class,
+        ], $stack);
+    }
+
+    public function testEnvelopeDuplicates(): void
+    {
+        $message = new Message('handler', 'test');
+        $message = new IdEnvelope($message, 'test-id');
+        $message = new IdEnvelope($message, 'test-id');
+        $message = new IdEnvelope($message, 'test-id');
+
+        $this->assertEquals('test', $message->getMessage()->getData());
+
+        $stack = $message->getMetadata()[EnvelopeInterface::ENVELOPE_STACK_KEY];
+        $this->assertIsArray($stack);
+
+        $this->assertEquals([
+            IdEnvelope::class,
+            IdEnvelope::class,
+            IdEnvelope::class,
+        ], $stack);
+    }
+}