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

jenkoian · jenkoian · commit ac379188ed90 · 2022-10-24T15:24:54.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/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
