feature symfony#46257 [Messenger] Add AMQP exchange to exchange bindings (Samik081)

This PR was squashed before being merged into the 7.4 branch.

Discussion
----------

[Messenger] Add AMQP exchange to exchange bindings

## Q/A
| Q             | A
| ------------- | ---
| Branch?       | 7.4
| Bug fix?      | no
| New feature?  | yes
| Deprecations? | no
| Tickets       |
| License       | MIT
| Doc PR        | symfony/symfony-docs#16783

## Changes description
This PR introduces very similar changes to this one: symfony#34737, which was closed due to the lack of the feedback.
I'd like to continue this topic, share the missing feedback and discuss it further if needed.

I have introduced the possibility to configure `exchange-to-exchange` bindings in `amqp` transport. This feature uses `\AMQPExchange::bind()` method already provided by the stub in `php-amqp/php-amqp`: https://github.com/php-amqp/php-amqp/blob/bb7611220e341039a7f5d72e606ca1e16eda4642/stubs/AMQPExchange.php#L22

Example `messenger.yaml`:
```yaml
framework:
    messenger:
        transports:
            some_transport:
                dsn: 'amqp://'
                options:
                    exchange:
                        type: topic
                        name: some_exchange
                        bindings: # added configuration
                            another_exchange:
                                binding_keys:
                                    - key1
                                    - key2
                                binding_arguments:
                                    x-match: all
```
With the above configuration, the `Connection` class creates `some_exchange` and binds it to `another_exchange` using `['key1', 'key2']` keys and `[x-match =>'all']` arguments.

## Reasoning

Binding an exchange to an exchange feature can be used to create more complex RabbitMQ topologies. It is also briefly described here: https://www.cloudamqp.com/blog/exchange-to-exchange-binding-in-rabbitmq.html

A real-world example could be kind of RabbitMQ publisher/subscriber pattern implementation between microservices, that could be visualized as follows:

![rabbitmq](https://user-images.githubusercontent.com/13415865/166811854-7d8b50e9-85c4-448c-a514-1449b3778cae.png)

In the above example (all the exchanges in this example are of the `topic` type):
```
App `Foo` publishes events to its own exchange AND subscribes to the events that app `Bar` publishes on its exchange.
App `Bar` publishes events to its own exchange AND subscribes to the events that app `Foo` publishes on its exchange.
App `...` subscribes to the events that apps `Foo` and `Bar` publish on their exchanges.
```

This approach might have few advantages, such as easier maintainability, dependency management and monitoring, or better separation between microservices.

### My thoughts
I feel that the fact, that https://github.com/php-amqp/php-amqp has `\AMQPExchange::bind()` implemented is already sufficient reason to have this supported in `symfony/amqp-messenger`. I am aware this feature might be rarely used, but it's already there in the extension, and having the ability to use it within `amqp-messenger` seems reasonable to me.

Commits
-------

9fd9049 [Messenger] Add AMQP exchange to exchange bindings

Author: fabpot

src/Symfony/Component/Messenger/Bridge/Amqp/Tests/Transport/ConnectionTest.php

@@ -393,7 +393,7 @@ public function testItSetupsTheTTLConnection()
         $connection->publish('body');
     }
 
-    public function testBindingArguments()
+    public function testQueueBindingArguments()
     {
         $amqpConnection = $this->createMock(\AMQPConnection::class);
         $amqpChannel = $this->createMock(\AMQPChannel::class);
@@ -418,6 +418,62 @@ 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')
+            ->willReturnCallback(function (...$args) {
+                static $series = [
+                    ['exchange0', 'binding_key0', ['x-match' => 'all']],
+                    ['exchange0', 'binding_key1', ['x-match' => 'all']],
+                    ['exchange1', 'binding_key2', ['x-match' => 'any']],
+                    ['exchange1', 'binding_key3', ['x-match' => 'any']],
+                ];
+
+                $expectedArgs = array_shift($series);
+                $this->assertSame($expectedArgs, $args);
+            })
+        ;
+        $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 testNoBindingKeysInExchangeBindings()
+    {
+        $this->expectException(\InvalidArgumentException::class);
+        $this->expectExceptionMessage('The "binding_keys" option must be set to a non-empty array for exchange "exchange0".');
+
+        $factory = new TestAmqpFactory(
+            $this->createMock(\AMQPConnection::class),
+            $this->createMock(\AMQPChannel::class),
+            $this->createMock(\AMQPQueue::class),
+            $this->createMock(\AMQPExchange::class)
+        );
+
+        $dsn = 'amqp://localhost?exchange[type]=headers'.
+            '&exchange[bindings][exchange0][binding_arguments][x-match]=all';
+
+        $connection = Connection::fromDsn($dsn, [], $factory);
+        $connection->publish('body');
+    }
+
     public function testItCanDisableTheSetup()
     {
         $factory = new TestAmqpFactory(

src/Symfony/Component/Messenger/Bridge/Amqp/Transport/Connection.php

@@ -79,6 +79,7 @@ class Connection
         'default_publish_routing_key',
         'flags',
         'arguments',
+        'bindings',
     ];
 
     private AmqpFactory $amqpFactory;
@@ -140,6 +141,9 @@ public function __construct(
      *     * 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 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")
@@ -460,6 +464,15 @@ private function setupExchangeAndQueues(): void
             $exchange->declareExchange();
         }
 
+        foreach ($this->exchangeOptions['bindings'] ?? [] as $exchangeName => $exchangeConfig) {
+            if (!\is_array($exchangeConfig['binding_keys'] ?? false) || !$exchangeConfig['binding_keys']) {
+                throw new InvalidArgumentException(\sprintf('The "binding_keys" option must be set to a non-empty array for exchange "%s".', $exchangeName));
+            }
+            foreach ($exchangeConfig['binding_keys'] as $bindingKey) {
+                $this->exchange()->bind($exchangeName, $bindingKey, $exchangeConfig['binding_arguments'] ?? []);
+            }
+        }
+
         foreach ($this->queuesOptions as $queueName => $queueConfig) {
             $this->queue($queueName)->declareQueue();
             if ('' !== $this->exchangeOptions['name']) {

src/Symfony/Component/Messenger/CHANGELOG.md

@@ -7,6 +7,7 @@ CHANGELOG
  * Add `--exclude-receivers` option to the `messenger:consume command`
  * Allow any `ServiceResetterInterface` implementation in `ResetServicesListener`
  * Add `Symfony\Component\Messenger\Middleware\AddDefaultStampsMiddleware` and `Symfony\Component\Messenger\Message\DefaultStampsProviderInterface`
+ * Add the possibility to configure exchange to exchange bindings in AMQP transport
 
 7.3
 ---