[Messenger] [Amqp] Add default exchange support

Author: ilyachase

src/Symfony/Component/Messenger/Bridge/Amqp/CHANGELOG.md

@@ -1,10 +1,15 @@
 CHANGELOG
 =========
 
+7.3
+---
+
+ * Add default exchange support
+
 7.1
 ---
 
-* Implement the `CloseableTransportInterface` to allow closing the AMQP connection
+ * Implement the `CloseableTransportInterface` to allow closing the AMQP connection
  * Add option `delay[arguments]` in the transport definition
 
 6.0

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

@@ -75,6 +75,36 @@ public function testItSendsAndReceivesMessages()
         $this->assertSame([], iterator_to_array($receiver->get()));
     }
 
+    public function testItSendsAndReceivesMessagesThroughDefaultExchange()
+    {
+        $serializer = $this->createSerializer();
+
+        $connection = Connection::fromDsn(getenv('MESSENGER_AMQP_DSN'), ['exchange' => ['name' => '']]);
+        $connection->setup();
+        $connection->purgeQueues();
+
+        $sender = new AmqpSender($connection, $serializer);
+        $receiver = new AmqpReceiver($connection, $serializer);
+
+        $sender->send($first = new Envelope(new DummyMessage('First'), [new AmqpStamp('messages')]));
+        $sender->send($second = new Envelope(new DummyMessage('Second'), [new AmqpStamp('messages')]));
+
+        $envelopes = iterator_to_array($receiver->get());
+        $this->assertCount(1, $envelopes);
+        /** @var Envelope $envelope */
+        $envelope = $envelopes[0];
+        $this->assertEquals($first->getMessage(), $envelope->getMessage());
+        $this->assertInstanceOf(AmqpReceivedStamp::class, $envelope->last(AmqpReceivedStamp::class));
+
+        $envelopes = iterator_to_array($receiver->get());
+        $this->assertCount(1, $envelopes);
+        /** @var Envelope $envelope */
+        $envelope = $envelopes[0];
+        $this->assertEquals($second->getMessage(), $envelope->getMessage());
+
+        $this->assertEmpty(iterator_to_array($receiver->get()));
+    }
+
     public function testRetryAndDelay()
     {
         $connection = Connection::fromDsn(getenv('MESSENGER_AMQP_DSN'));

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

@@ -875,6 +875,55 @@ private function createDelayOrRetryConnection(\AMQPExchange $delayExchange, stri
 
         return Connection::fromDsn('amqp://localhost', [], $factory);
     }
+
+    public function testGettingDefaultExchange()
+    {
+        $factory = $this->createMock(AmqpFactory::class);
+
+        $amqpExchange = $this->createMock(\AMQPExchange::class);
+        $amqpExchange->expects($this->once())->method('setName')->with('');
+        $amqpExchange->expects($this->once())->method('setType')->with(\AMQP_EX_TYPE_DIRECT);
+        $amqpExchange->expects($this->never())->method('setFlags');
+        $amqpExchange->expects($this->never())->method('setArguments');
+
+        $factory->expects($this->once())->method('createExchange')->willReturn($amqpExchange);
+
+        $connection = new Connection([
+            'host' => 'localhost',
+            'port' => 5672,
+            'vhost' => '/',
+        ], [
+            'name' => '',
+        ], [
+            '' => [],
+        ], $factory);
+
+        $connection->exchange();
+    }
+
+    public function testBindIsNotCalledWhenPublishingInDefaultExchange()
+    {
+        $factory = $this->createMock(AmqpFactory::class);
+
+        $amqpExchange = $this->createMock(\AMQPExchange::class);
+        $amqpExchange->expects($this->never())->method('declareExchange');
+
+        $factory->expects($this->once())->method('createExchange')->willReturn($amqpExchange);
+        $factory->expects($this->once())->method('createQueue')->willReturn($queue = $this->createMock(\AMQPQueue::class));
+        $queue->expects($this->never())->method('bind');
+
+        $connection = new Connection([
+            'host' => 'localhost',
+            'port' => 5672,
+            'vhost' => '/',
+        ], [
+            'name' => '',
+        ], [
+            '' => [],
+        ], $factory);
+
+        $connection->publish('body');
+    }
 }
 
 class TestAmqpFactory extends AmqpFactory

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

@@ -134,7 +134,7 @@ public function __construct(
      *     * flags: Queue flags (Default: AMQP_DURABLE)
      *     * arguments: Extra arguments
      *   * exchange:
-     *     * name: Name of the exchange
+     *     * name: Name of the exchange. An empty string (name: '') can be used to use the default exchange
      *     * type: Type of exchange (Default: fanout)
      *     * default_publish_routing_key: Routing key to use when publishing, if none is specified on the message
      *     * flags: Exchange flags (Default: AMQP_DURABLE)
@@ -454,12 +454,17 @@ public function setup(): void
 
     private function setupExchangeAndQueues(): void
     {
-        $this->exchange()->declareExchange();
+        $exchange = $this->exchange();
+        if ('' !== $this->exchangeOptions['name']) {
+            $exchange->declareExchange();
+        }
 
         foreach ($this->queuesOptions as $queueName => $queueConfig) {
             $this->queue($queueName)->declareQueue();
-            foreach ($queueConfig['binding_keys'] ?? [null] as $bindingKey) {
-                $this->queue($queueName)->bind($this->exchangeOptions['name'], $bindingKey, $queueConfig['binding_arguments'] ?? []);
+            if ('' !== $this->exchangeOptions['name']) {
+                foreach ($queueConfig['binding_keys'] ?? [null] as $bindingKey) {
+                    $this->queue($queueName)->bind($this->exchangeOptions['name'], $bindingKey, $queueConfig['binding_arguments'] ?? []);
+                }
             }
         }
         $this->autoSetupExchange = false;
@@ -533,11 +538,14 @@ public function exchange(): \AMQPExchange
         if (!isset($this->amqpExchange)) {
             $this->amqpExchange = $this->amqpFactory->createExchange($this->channel());
             $this->amqpExchange->setName($this->exchangeOptions['name']);
-            $this->amqpExchange->setType($this->exchangeOptions['type'] ?? \AMQP_EX_TYPE_FANOUT);
-            $this->amqpExchange->setFlags($this->exchangeOptions['flags'] ?? \AMQP_DURABLE);
+            $defaultExchangeType = '' !== $this->exchangeOptions['name'] ? \AMQP_EX_TYPE_FANOUT : \AMQP_EX_TYPE_DIRECT;
+            $this->amqpExchange->setType($this->exchangeOptions['type'] ?? $defaultExchangeType);
+            if ('' !== $this->exchangeOptions['name']) {
+                $this->amqpExchange->setFlags($this->exchangeOptions['flags'] ?? \AMQP_DURABLE);
 
-            if (isset($this->exchangeOptions['arguments'])) {
-                $this->amqpExchange->setArguments($this->exchangeOptions['arguments']);
+                if (isset($this->exchangeOptions['arguments'])) {
+                    $this->amqpExchange->setArguments($this->exchangeOptions['arguments']);
+                }
             }
         }