Introduce Envelope (#185)

yiisoft · Jan 12, 2024 · 8498f59 · 8498f59
1 parent 6e7fec2
commit 8498f59
Show file tree

Hide file tree

Showing 23 changed files with 188 additions and 81 deletions.
diff --git a/composer.json b/composer.json
@@ -40,9 +40,9 @@
         "rector/rector": "^0.19.0",
         "roave/infection-static-analysis-plugin": "^1.16",
         "spatie/phpunit-watcher": "^1.23",
-        "yiisoft/yii-debug": "dev-master",
         "vimeo/psalm": "^4.30|^5.8",
-        "yiisoft/test-support": "^3.0"
+        "yiisoft/test-support": "^3.0",
+        "yiisoft/yii-debug": "dev-master|dev-php80"
     },
     "suggest": {
         "ext-pcntl": "Need for process signals"

diff --git a/src/Adapter/AdapterInterface.php b/src/Adapter/AdapterInterface.php
@@ -26,12 +26,12 @@ public function runExisting(callable $handlerCallback): void;
      *
      * @return JobStatus
      */
-    public function status(string $id): JobStatus;
+    public function status(string|int $id): JobStatus;
 
     /**
      * Pushing a message to the queue. Adapter sets message ID if available.
      */
-    public function push(MessageInterface $message): void;
+    public function push(MessageInterface $message): MessageInterface;
 
     /**
      * Listen to the queue and pass messages to the given handler as they come.

diff --git a/src/Adapter/SynchronousAdapter.php b/src/Adapter/SynchronousAdapter.php
@@ -10,6 +10,7 @@
 use Yiisoft\Queue\QueueFactory;
 use Yiisoft\Queue\QueueInterface;
 use Yiisoft\Queue\Worker\WorkerInterface;
+use Yiisoft\Queue\Message\IdEnvelope;
 
 final class SynchronousAdapter implements AdapterInterface
 {
@@ -42,7 +43,7 @@ public function runExisting(callable $handlerCallback): void
         }
     }
 
-    public function status(string $id): JobStatus
+    public function status(string|int $id): JobStatus
     {
         $id = (int) $id;
 
@@ -61,12 +62,12 @@ public function status(string $id): JobStatus
         throw new InvalidArgumentException('There is no message with the given ID.');
     }
 
-    public function push(MessageInterface $message): void
+    public function push(MessageInterface $message): MessageInterface
     {
         $key = count($this->messages) + $this->current;
         $this->messages[] = $message;
 
-        $message->setId((string) $key);
+        return new IdEnvelope($message, $key);
     }
 
     public function subscribe(callable $handlerCallback): void

diff --git a/src/Debug/QueueDecorator.php b/src/Debug/QueueDecorator.php
@@ -18,7 +18,7 @@ public function __construct(
     ) {
     }
 
-    public function status(string $id): JobStatus
+    public function status(string|int $id): JobStatus
     {
         $result = $this->queue->status($id);
         $this->collector->collectStatus($id, $result);

diff --git a/src/Exception/JobFailureException.php b/src/Exception/JobFailureException.php
@@ -7,13 +7,14 @@
 use RuntimeException;
 use Throwable;
 use Yiisoft\Queue\Message\MessageInterface;
+use Yiisoft\Queue\Message\IdEnvelope;
 
 class JobFailureException extends RuntimeException
 {
     public function __construct(private MessageInterface $queueMessage, Throwable $previous)
     {
         $error = $previous->getMessage();
-        $messageId = $queueMessage->getId() ?? 'null';
+        $messageId = $queueMessage->getMetadata()[IdEnvelope::MESSAGE_ID_KEY] ?? 'null';
         $messageText = "Processing of message #$messageId is stopped because of an exception:\n$error.";
 
         parent::__construct($messageText, 0, $previous);

diff --git a/src/Message/EnvelopeInterface.php b/src/Message/EnvelopeInterface.php
@@ -0,0 +1,15 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\Message;
+
+/**
+ * Envelope is a message container that adds additional metadata.
+ */
+interface EnvelopeInterface extends MessageInterface
+{
+    public function getMessage(): MessageInterface;
+
+    public function withMessage(MessageInterface $message): self;
+}
diff --git a/src/Message/EnvelopeTrait.php b/src/Message/EnvelopeTrait.php
@@ -0,0 +1,38 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\Message;
+
+trait EnvelopeTrait
+{
+    private MessageInterface $message;
+
+    public function getMessage(): MessageInterface
+    {
+        return $this->message;
+    }
+
+    public function withMessage(MessageInterface $message): self
+    {
+        $instance = clone $this;
+        $instance->message = $message;
+
+        return $instance;
+    }
+
+    public function getHandlerName(): string
+    {
+        return $this->message->getHandlerName();
+    }
+
+    public function getData(): mixed
+    {
+        return $this->message->getData();
+    }
+
+    public function getMetadata(): array
+    {
+        return $this->message->getMetadata();
+    }
+}
diff --git a/src/Message/IdEnvelope.php b/src/Message/IdEnvelope.php
@@ -0,0 +1,38 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\Message;
+
+/**
+ * ID envelope allows to identify a message.
+ */
+final class IdEnvelope implements EnvelopeInterface
+{
+    use EnvelopeTrait;
+
+    public const MESSAGE_ID_KEY = 'yii-message-id';
+
+    public function __construct(
+        private MessageInterface $message,
+        private string|int|null $id = null,
+    ) {
+    }
+
+    public function setId(string|int|null $id): void
+    {
+        $this->id = $id;
+    }
+
+    public function getId(): string|int|null
+    {
+        return $this->id ?? $this->message->getMetadata()[self::MESSAGE_ID_KEY] ?? null;
+    }
+
+    public function getMetadata(): array
+    {
+        return array_merge($this->message->getMetadata(), [
+            self::MESSAGE_ID_KEY => $this->getId(),
+        ]);
+    }
+}
diff --git a/src/Message/Message.php b/src/Message/Message.php
@@ -15,7 +15,6 @@ public function __construct(
         private string $handlerName,
         private mixed $data,
         private array $metadata = [],
-        private ?string $id = null
     ) {
     }
 
@@ -29,16 +28,6 @@ public function getData(): mixed
         return $this->data;
     }
 
-    public function setId(?string $id): void
-    {
-        $this->id = $id;
-    }
-
-    public function getId(): ?string
-    {
-        return $this->id;
-    }
-
     public function getMetadata(): array
     {
         return $this->metadata;

diff --git a/src/Message/MessageInterface.php b/src/Message/MessageInterface.php
@@ -6,15 +6,6 @@
 
 interface MessageInterface
 {
-    public function setId(?string $id): void;
-
-    /**
-     * Returns unique message ID.
-     *
-     * @return string|null
-     */
-    public function getId(): ?string;
-
     /**
      * Returns handler name.
      *

diff --git a/src/Middleware/FailureHandling/FailureEnvelope.php b/src/Middleware/FailureHandling/FailureEnvelope.php
@@ -0,0 +1,25 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\Middleware\FailureHandling;
+
+use Yiisoft\Queue\Message\EnvelopeInterface;
+use Yiisoft\Queue\Message\EnvelopeTrait;
+use Yiisoft\Queue\Message\MessageInterface;
+
+final class FailureEnvelope implements EnvelopeInterface
+{
+    use EnvelopeTrait;
+
+    public function __construct(
+        private MessageInterface $message,
+        private array $meta,
+    ) {
+    }
+
+    public function getMetadata(): array
+    {
+        return array_merge($this->message->getMetadata(), $this->meta);
+    }
+}
diff --git a/src/Middleware/FailureHandling/Implementation/ExponentialDelayMiddleware.php b/src/Middleware/FailureHandling/Implementation/ExponentialDelayMiddleware.php
@@ -12,6 +12,7 @@
 use Yiisoft\Queue\Middleware\FailureHandling\MiddlewareFailureInterface;
 use Yiisoft\Queue\Middleware\Push\Implementation\DelayMiddlewareInterface;
 use Yiisoft\Queue\QueueInterface;
+use Yiisoft\Queue\Middleware\FailureHandling\FailureEnvelope;
 
 /**
  * Failure strategy which resends the given message to a queue with an exponentially increasing delay.
@@ -62,15 +63,12 @@ public function processFailure(
     ): FailureHandlingRequest {
         $message = $request->getMessage();
         if ($this->suites($message)) {
-            $messageNew = new Message(
-                handlerName: $message->getHandlerName(),
-                data:        $message->getData(),
-                metadata:    $this->formNewMeta($message),
-                id:          $message->getId(),
-            );
-            ($this->queue ?? $request->getQueue())->push(
-                $messageNew,
-                $this->delayMiddleware->withDelay($this->getDelay($message))
+            $envelope = new FailureEnvelope($message, $this->createNewMeta($message));
+            $queue = $this->queue ?? $request->getQueue();
+            $middlewareDefinitions = $this->delayMiddleware->withDelay($this->getDelay($envelope));
+            $messageNew = $queue->push(
+                $envelope,
+                $middlewareDefinitions
             );
 
             return $request->withMessage($messageNew);
@@ -84,7 +82,7 @@ private function suites(MessageInterface $message): bool
         return $this->maxAttempts > $this->getAttempts($message);
     }
 
-    private function formNewMeta(MessageInterface $message): array
+    private function createNewMeta(MessageInterface $message): array
     {
         $meta = $message->getMetadata();
         $meta[self::META_KEY_DELAY . "-$this->id"] = $this->getDelay($message);

diff --git a/src/Middleware/FailureHandling/Implementation/SendAgainMiddleware.php b/src/Middleware/FailureHandling/Implementation/SendAgainMiddleware.php
@@ -5,8 +5,8 @@
 namespace Yiisoft\Queue\Middleware\FailureHandling\Implementation;
 
 use InvalidArgumentException;
-use Yiisoft\Queue\Message\Message;
 use Yiisoft\Queue\Message\MessageInterface;
+use Yiisoft\Queue\Middleware\FailureHandling\FailureEnvelope;
 use Yiisoft\Queue\Middleware\FailureHandling\FailureHandlingRequest;
 use Yiisoft\Queue\Middleware\FailureHandling\MessageFailureHandlerInterface;
 use Yiisoft\Queue\Middleware\FailureHandling\MiddlewareFailureInterface;
@@ -40,15 +40,11 @@ public function processFailure(
     ): FailureHandlingRequest {
         $message = $request->getMessage();
         if ($this->suites($message)) {
-            $message = new Message(
-                handlerName: $message->getHandlerName(),
-                data: $message->getData(),
-                metadata: $this->createMeta($message),
-                id: $message->getId(),
-            );
-            $message = $this->queue?->push($message) ?? $request->getQueue()->push($message);
+            $envelope = new FailureEnvelope($message, $this->createMeta($message));
+            $envelope = ($this->queue ?? $request->getQueue())->push($envelope);
 
-            return $request->withMessage($message)->withQueue($this->queue ?? $request->getQueue());
+            return $request->withMessage($envelope)
+                ->withQueue($this->queue ?? $request->getQueue());
         }
 
         return $handler->handleFailure($request);
@@ -59,11 +55,6 @@ private function suites(MessageInterface $message): bool
         return $this->getAttempts($message) < $this->maxAttempts;
     }
 
-    private function getMetaKey(): string
-    {
-        return self::META_KEY_RESEND . "-$this->id";
-    }
-
     private function createMeta(MessageInterface $message): array
     {
         $metadata = $message->getMetadata();
@@ -81,4 +72,9 @@ private function getAttempts(MessageInterface $message): int
 
         return (int) $result;
     }
+
+    private function getMetaKey(): string
+    {
+        return self::META_KEY_RESEND . "-$this->id";
+    }
 }
diff --git a/src/Middleware/Push/AdapterPushHandler.php b/src/Middleware/Push/AdapterPushHandler.php
@@ -13,11 +13,9 @@ final class AdapterPushHandler implements MessageHandlerPushInterface
 {
     public function handlePush(PushRequest $request): PushRequest
     {
-        if ($request->getAdapter() === null) {
+        if (($adapter = $request->getAdapter()) === null) {
             throw new AdapterNotConfiguredException();
         }
-        $request->getAdapter()->push($request->getMessage());
-
-        return $request;
+        return $request->withMessage($adapter->push($request->getMessage()));
     }
 }
diff --git a/src/Queue.php b/src/Queue.php
@@ -16,6 +16,7 @@
 use Yiisoft\Queue\Middleware\Push\PushMiddlewareDispatcher;
 use Yiisoft\Queue\Middleware\Push\PushRequest;
 use Yiisoft\Queue\Worker\WorkerInterface;
+use Yiisoft\Queue\Message\IdEnvelope;
 
 final class Queue implements QueueInterface
 {
@@ -57,9 +58,10 @@ public function push(
             ->dispatch($request, $this->createPushHandler($middlewareDefinitions))
             ->getMessage();
 
+        $messageId = $message->getMetadata()[IdEnvelope::MESSAGE_ID_KEY] ?? 'null';
         $this->logger->info(
             'Pushed message with handler name "{handlerName}" to the queue. Assigned ID #{id}.',
-            ['handlerName' => $message->getHandlerName(), 'id' => $message->getId() ?? 'null']
+            ['handlerName' => $message->getHandlerName(), 'id' => $messageId]
         );
 
         return $message;
@@ -100,7 +102,7 @@ public function listen(): void
         $this->logger->info('Finish listening to the queue.');
     }
 
-    public function status(string $id): JobStatus
+    public function status(string|int $id): JobStatus
     {
         $this->checkAdapter();
 
@@ -159,7 +161,7 @@ private function createPushHandler(array $middlewares): MessageHandlerPushInterf
         return new class (
             $this->adapterPushHandler,
             $this->pushMiddlewareDispatcher,
-            [...array_values($this->middlewareDefinitions), ...array_values($middlewares)]
+            array_merge($this->middlewareDefinitions, $middlewares)
         ) implements MessageHandlerPushInterface {
             public function __construct(
                 private AdapterPushHandler $adapterPushHandler,

diff --git a/src/QueueInterface.php b/src/QueueInterface.php
@@ -45,7 +45,7 @@ public function listen(): void;
      *
      * @return JobStatus
      */
-    public function status(string $id): JobStatus;
+    public function status(string|int $id): JobStatus;
 
     public function withAdapter(AdapterInterface $adapter): self;