[Messenger] Use TransportMessageIdStamp in InMemoryTransport allows retrying

Author: alexndlm

src/Symfony/Component/Messenger/Tests/Transport/InMemoryTransportTest.php

@@ -41,9 +41,9 @@ public function testSend()
     public function testQueue()
     {
         $envelope1 = new Envelope(new \stdClass());
-        $this->transport->send($envelope1);
+        $envelope1 = $this->transport->send($envelope1);
         $envelope2 = new Envelope(new \stdClass());
-        $this->transport->send($envelope2);
+        $envelope2 = $this->transport->send($envelope2);
         $this->assertSame([$envelope1, $envelope2], $this->transport->get());
         $this->transport->ack($envelope1);
         $this->assertSame([$envelope2], $this->transport->get());
@@ -54,9 +54,9 @@ public function testQueue()
     public function testAcknowledgeSameMessageWithDifferentStamps()
     {
         $envelope1 = new Envelope(new \stdClass(), [new AnEnvelopeStamp()]);
-        $this->transport->send($envelope1);
+        $envelope1 = $this->transport->send($envelope1);
         $envelope2 = new Envelope(new \stdClass(), [new AnEnvelopeStamp()]);
-        $this->transport->send($envelope2);
+        $envelope2 = $this->transport->send($envelope2);
         $this->assertSame([$envelope1, $envelope2], $this->transport->get());
         $this->transport->ack($envelope1->with(new AnEnvelopeStamp()));
         $this->assertSame([$envelope2], $this->transport->get());
@@ -67,21 +67,23 @@ public function testAcknowledgeSameMessageWithDifferentStamps()
     public function testAck()
     {
         $envelope = new Envelope(new \stdClass());
+        $envelope = $this->transport->send($envelope);
         $this->transport->ack($envelope);
         $this->assertSame([$envelope], $this->transport->getAcknowledged());
     }
 
     public function testReject()
     {
         $envelope = new Envelope(new \stdClass());
+        $envelope = $this->transport->send($envelope);
         $this->transport->reject($envelope);
         $this->assertSame([$envelope], $this->transport->getRejected());
     }
 
     public function testReset()
     {
         $envelope = new Envelope(new \stdClass());
-        $this->transport->send($envelope);
+        $envelope = $this->transport->send($envelope);
         $this->transport->ack($envelope);
         $this->transport->reject($envelope);
 

src/Symfony/Component/Messenger/Transport/InMemoryTransport.php

@@ -12,6 +12,8 @@
 namespace Symfony\Component\Messenger\Transport;
 
 use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Exception\LogicException;
+use Symfony\Component\Messenger\Stamp\TransportMessageIdStamp;
 use Symfony\Contracts\Service\ResetInterface;
 
 /**
@@ -41,6 +43,8 @@ class InMemoryTransport implements TransportInterface, ResetInterface
      */
     private $queue = [];
 
+    private $nextId = 1;
+
     /**
      * {@inheritdoc}
      */
@@ -55,8 +59,12 @@ public function get(): iterable
     public function ack(Envelope $envelope): void
     {
         $this->acknowledged[] = $envelope;
-        $id = spl_object_hash($envelope->getMessage());
-        unset($this->queue[$id]);
+
+        if (!$transportMessageIdStamp = $envelope->last(TransportMessageIdStamp::class)) {
+            throw new LogicException('No TransportMessageIdStamp found on the Envelope.');
+        }
+
+        unset($this->queue[$transportMessageIdStamp->getId()]);
     }
 
     /**
@@ -65,8 +73,12 @@ public function ack(Envelope $envelope): void
     public function reject(Envelope $envelope): void
     {
         $this->rejected[] = $envelope;
-        $id = spl_object_hash($envelope->getMessage());
-        unset($this->queue[$id]);
+
+        if (!$transportMessageIdStamp = $envelope->last(TransportMessageIdStamp::class)) {
+            throw new LogicException('No TransportMessageIdStamp found on the Envelope.');
+        }
+
+        unset($this->queue[$transportMessageIdStamp->getId()]);
     }
 
     /**
@@ -75,7 +87,8 @@ public function reject(Envelope $envelope): void
     public function send(Envelope $envelope): Envelope
     {
         $this->sent[] = $envelope;
-        $id = spl_object_hash($envelope->getMessage());
+        $id = $this->nextId++;
+        $envelope = $envelope->with(new TransportMessageIdStamp($id));
         $this->queue[$id] = $envelope;
 
         return $envelope;