[Messenger] Add AMQP exchange to exchange bindings

Samik081 · Samik081 · commit c7214102e6ad · 2022-05-31T10:46:54.000+02:00
Add possibility to configure exchange to exchange bindings in AMQP transport
diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/Tests/Transport/ConnectionTest.php b/src/Symfony/Component/Messenger/Bridge/Amqp/Tests/Transport/ConnectionTest.php
@@ -356,7 +356,7 @@ public function testItSetupsTheTTLConnection()
         $connection->publish('body');
     }
 
-    public function testBindingArguments()
+    public function testQueueBindingArguments()
     {
         $amqpConnection = $this->createMock(\AMQPConnection::class);
         $amqpChannel = $this->createMock(\AMQPChannel::class);
@@ -383,6 +383,36 @@ public function testBindingArguments()
         $connection->publish('body');
     }
 
+    public function testExchangeBindingArguments()
+    {
+        $factory = new TestAmqpFactory(
+            $this->createMock(\AMQPConnection::class),
+            $this->createMock(\AMQPChannel::class),
+            $this->createMock(\AMQPQueue::class),
+            $amqpExchange = $this->createMock(\AMQPExchange::class)
+        );
+
+        $amqpExchange->expects($this->once())->method('declareExchange');
+        $amqpExchange->expects($this->exactly(4))->method('bind')->withConsecutive(
+            ['exchange0', 'binding_key0', ['x-match' => 'all']],
+            ['exchange0', 'binding_key1', ['x-match' => 'all']],
+            ['exchange1', 'binding_key2', ['x-match' => 'any']],
+            ['exchange1', 'binding_key3', ['x-match' => 'any']],
+        );
+        $amqpExchange->expects($this->once())->method('publish')->with('body', null, \AMQP_NOPARAM, ['headers' => [], 'delivery_mode' => 2, 'timestamp' => time()]);
+
+        $dsn = 'amqp://localhost?exchange[type]=headers'.
+            '&exchange[bindings][exchange0][binding_arguments][x-match]=all'.
+            '&exchange[bindings][exchange0][binding_keys][0]=binding_key0'.
+            '&exchange[bindings][exchange0][binding_keys][1]=binding_key1'.
+            '&exchange[bindings][exchange1][binding_arguments][x-match]=any'.
+            '&exchange[bindings][exchange1][binding_keys][0]=binding_key2'.
+            '&exchange[bindings][exchange1][binding_keys][1]=binding_key3';
+
+        $connection = Connection::fromDsn($dsn, [], $factory);
+        $connection->publish('body');
+    }
+
     public function testItCanDisableTheSetup()
     {
         $factory = new TestAmqpFactory(
diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/Transport/Connection.php b/src/Symfony/Component/Messenger/Bridge/Amqp/Transport/Connection.php
@@ -73,6 +73,7 @@ class Connection
         'default_publish_routing_key',
         'flags',
         'arguments',
+        'bindings',
     ];
 
     private array $connectionOptions;
@@ -134,6 +135,9 @@ public function __construct(array $connectionOptions, array $exchangeOptions, ar
      *     * default_publish_routing_key: Routing key to use when publishing, if none is specified on the message
      *     * flags: Exchange flags (Default: AMQP_DURABLE)
      *     * arguments: Extra arguments
+     *     * bindings[name]: An array of the source exchanges to bind this exchange to, keyed by the name. Binding direction: source exchange -> this exchange
+     *       * binding_keys: The binding/routing keys (if any) to be used for the binding
+     *       * binding_arguments: Additional binding arguments
      *   * delay:
      *     * queue_name_pattern: Pattern to use to create the queues (Default: "delay_%exchange_name%_%routing_key%_%delay%")
      *     * exchange_name: Name of the exchange to be used for the delayed/retried messages (Default: "delays")
@@ -448,6 +452,12 @@ private function setupExchangeAndQueues(): void
     {
         $this->exchange()->declareExchange();
 
+        foreach ($this->exchangeOptions['bindings'] ?? [] as $exchangeName => $exchangeConfig) {
+            foreach ($exchangeConfig['binding_keys'] ?? [] as $bindingKey) {
+                $this->exchange()->bind($exchangeName, $bindingKey, $exchangeConfig['binding_arguments'] ?? []);
+            }
+        }
+
         foreach ($this->queuesOptions as $queueName => $queueConfig) {
             $this->queue($queueName)->declareQueue();
             foreach ($queueConfig['binding_keys'] ?? [null] as $bindingKey) {
diff --git a/src/Symfony/Component/Messenger/CHANGELOG.md b/src/Symfony/Component/Messenger/CHANGELOG.md
@@ -7,6 +7,7 @@ CHANGELOG
  * Add `SerializedMessageStamp` to avoid serializing a message when a retry occurs
  * Automatically resolve handled message type when method different from `__invoke` is used as handler
  * Allow `#[AsMessageHandler]` attribute on methods
+ * Add possibility to configure exchange to exchange bindings in AMQP transport
 
 6.0
 ---
