[Messenger][AMQP] Add routing key to data passed to serializer decode.

jenkoian · jenkoian · commit 6a5e2b5531ba · 2022-10-24T15:41:26.000+01:00
It's often useful to have the routing key available when decoding messages from AMQP. For example, sometimes the routing key contains meta information required for the processing of the message. This introduces an `extra` element which currently only contains the routing key, but there is scope to add other meta information from \AMQPEnvelope such as the exchange name for example. Fixes #43039
diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/Tests/Transport/AmqpReceiverTest.php b/src/Symfony/Component/Messenger/Bridge/Amqp/Tests/Transport/AmqpReceiverTest.php
@@ -74,13 +74,35 @@ public function testItThrowsATransportExceptionIfItCannotRejectMessage()
         $receiver->reject(new Envelope(new \stdClass(), [new AmqpReceivedStamp($amqpEnvelope, 'queueName')]));
     }
 
+    public function testItPassesExpectedDataToSerializerDecode()
+    {
+        $amqpEnvelope = $this->createAMQPEnvelope();
+
+        $connection = $this->createMock(Connection::class);
+        $connection->method('getQueueNames')->willReturn(['queueName']);
+        $connection->method('get')->with('queueName')->willReturn($amqpEnvelope);
+
+        $serializer = $this->createMock(SerializerInterface::class);
+        $serializer->expects($this->once())->method('decode')->with($this->equalTo([
+            'body' => $amqpEnvelope->getBody(),
+            'headers' => $amqpEnvelope->getHeaders(),
+            'extra' => [
+                'routingKey' => $amqpEnvelope->getRoutingKey(),
+            ],
+        ]))->willReturn(new Envelope(new DummyMessage('Hi')));
+
+        $receiver = new AmqpReceiver($connection, $serializer);
+        iterator_to_array($receiver->get());
+    }
+
     private function createAMQPEnvelope(): \AMQPEnvelope
     {
         $envelope = $this->createMock(\AMQPEnvelope::class);
         $envelope->method('getBody')->willReturn('{"message": "Hi"}');
         $envelope->method('getHeaders')->willReturn([
             'type' => DummyMessage::class,
         ]);
+        $envelope->method('getRoutingKey')->willReturn('Dummy_RoutingKey');
 
         return $envelope;
     }
diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/Tests/Transport/AmqpTransportTest.php b/src/Symfony/Component/Messenger/Bridge/Amqp/Tests/Transport/AmqpTransportTest.php
@@ -43,8 +43,9 @@ public function testReceivesMessages()
         $amqpEnvelope = $this->createMock(\AMQPEnvelope::class);
         $amqpEnvelope->method('getBody')->willReturn('body');
         $amqpEnvelope->method('getHeaders')->willReturn(['my' => 'header']);
+        $amqpEnvelope->method('getRoutingKey')->willReturn('routing-key');
 
-        $serializer->method('decode')->with(['body' => 'body', 'headers' => ['my' => 'header']])->willReturn(new Envelope($decodedMessage));
+        $serializer->method('decode')->with(['body' => 'body', 'headers' => ['my' => 'header'], 'extra' => ['routingKey' => 'routing-key']])->willReturn(new Envelope($decodedMessage));
         $connection->method('getQueueNames')->willReturn(['queueName']);
         $connection->method('get')->with('queueName')->willReturn($amqpEnvelope);
 
diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/Transport/AmqpReceiver.php b/src/Symfony/Component/Messenger/Bridge/Amqp/Transport/AmqpReceiver.php
@@ -66,6 +66,9 @@ private function getEnvelope(string $queueName): iterable
             $envelope = $this->serializer->decode([
                 'body' => false === $body ? '' : $body, // workaround https://github.com/pdezwart/php-amqp/issues/351
                 'headers' => $amqpEnvelope->getHeaders(),
+                'extra' => [
+                    'routingKey' => $amqpEnvelope->getRoutingKey(),
+                ],
             ]);
         } catch (MessageDecodingFailedException $exception) {
             // invalid message of some type
