Add InteropMessageStamp

Author: KonstantinCodes

EnvelopeItem/InteropMessageStamp.php

@@ -0,0 +1,22 @@
+<?php
+
+namespace Enqueue\MessengerAdapter\EnvelopeItem;
+
+use Interop\Queue\Message;
+use Symfony\Component\Messenger\Stamp\StampInterface;
+
+final class InteropMessageStamp implements StampInterface
+{
+    /** @var Message */
+    private $message;
+
+    public function __construct(Message $message)
+    {
+        $this->message = $message;
+    }
+
+    public function getMessage(): Message
+    {
+        return $this->message;
+    }
+}

QueueInteropTransport.php

@@ -14,6 +14,7 @@
 use Enqueue\AmqpTools\DelayStrategyAware;
 use Enqueue\AmqpTools\RabbitMqDelayPluginDelayStrategy;
 use Enqueue\AmqpTools\RabbitMqDlxDelayStrategy;
+use Enqueue\MessengerAdapter\EnvelopeItem\InteropMessageStamp;
 use Interop\Queue\Consumer;
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
@@ -77,13 +78,18 @@ public function get(): iterable
             throw $e;
         }
 
-        return array(
-            $this->serializer->decode(array(
-                'body' => $interopMessage->getBody(),
-                'headers' => $interopMessage->getHeaders(),
-                'properties' => $interopMessage->getProperties(),
-            )),
-        );
+        /** @var Envelope $envelope */
+        $envelope = $this->serializer->decode(array(
+            'body' => $interopMessage->getBody(),
+            'headers' => $interopMessage->getHeaders(),
+            'properties' => $interopMessage->getProperties(),
+        ));
+
+        if ($envelope) {
+            $envelope = $envelope->with(new InteropMessageStamp($interopMessage));
+        }
+
+        return array($envelope);
     }
 
     /**
@@ -217,6 +223,13 @@ private function setMessageMetadata(Message $interopMessage, Envelope $envelope)
 
     private function encodeMessage(Envelope $envelope): Message
     {
+        /** @var InteropMessageStamp $interopStamp */
+        $interopStamp = $envelope->last(InteropMessageStamp::class);
+
+        if ($interopStamp) {
+            return $interopStamp->getMessage();
+        }
+
         $context = $this->contextManager->context();
         $encodedMessage = $this->serializer->encode($envelope);