bug #39608 [Messenger] Setup queues once in AMQP (jderusse)

This PR was merged into the 5.3-dev branch.

Discussion
----------

[Messenger] Setup queues once in AMQP

| Q             | A
| ------------- | ---
| Branch?       | 5.x
| Bug fix?      | no
| New feature?  | yes
| Deprecations? | no
| Tickets       | Fix #39605, Fix #38092, Fix #32172
| License       | MIT
| Doc PR        | -

To ease the setup, this PR also merge setup of exchange AND delayExchange.

/cc @Nyholm

Commits
-------

c2e84c610a Setup queues once in AMQP

Author: fabpot

Tests/Transport/ConnectionTest.php

@@ -421,6 +421,24 @@ public function testItCanDisableTheSetup()
         $connection->publish('body');
     }
 
+    public function testItSetupQueuesOnce()
+    {
+        $factory = new TestAmqpFactory(
+            $amqpConnection = $this->createMock(\AMQPConnection::class),
+            $amqpChannel = $this->createMock(\AMQPChannel::class),
+            $amqpQueue = $this->createMock(\AMQPQueue::class),
+            $amqpExchange = $this->createMock(\AMQPExchange::class)
+        );
+
+        $amqpExchange->expects($this->once())->method('declareExchange');
+        $amqpQueue->expects($this->once())->method('declareQueue');
+        $amqpQueue->expects($this->once())->method('bind');
+
+        $connection = Connection::fromDsn('amqp://localhost', ['auto_setup' => true], $factory);
+        $connection->publish('body');
+        $connection->publish('body');
+    }
+
     public function testSetChannelPrefetchWhenSetup()
     {
         $factory = new TestAmqpFactory(

Transport/Connection.php

@@ -79,6 +79,8 @@ class Connection
     private $exchangeOptions;
     private $queuesOptions;
     private $amqpFactory;
+    private $autoSetupExchange;
+    private $autoSetup;
 
     /**
      * @var \AMQPChannel|null
@@ -112,6 +114,7 @@ public function __construct(array $connectionOptions, array $exchangeOptions, ar
                 'queue_name_pattern' => 'delay_%exchange_name%_%routing_key%_%delay%',
             ],
         ], $connectionOptions);
+        $this->autoSetupExchange = $this->autoSetup = $connectionOptions['auto_setup'] ?? true;
         $this->exchangeOptions = $exchangeOptions;
         $this->queuesOptions = $queuesOptions;
         $this->amqpFactory = $amqpFactory ?: new AmqpFactory();
@@ -207,6 +210,9 @@ public static function fromDsn(string $dsn, array $options = [], AmqpFactory $am
         $exchangeOptions = $amqpOptions['exchange'];
         $queuesOptions = $amqpOptions['queues'];
         unset($amqpOptions['queues'], $amqpOptions['exchange']);
+        if (isset($amqpOptions['auto_setup'])) {
+            $amqpOptions['auto_setup'] = filter_var($amqpOptions['auto_setup'], \FILTER_VALIDATE_BOOLEAN);
+        }
 
         $queuesOptions = array_map(function ($queueOptions) {
             if (!\is_array($queueOptions)) {
@@ -285,7 +291,7 @@ public function publish(string $body, array $headers = [], int $delayInMs = 0, A
             return;
         }
 
-        if ($this->shouldSetup()) {
+        if ($this->autoSetupExchange) {
             $this->setupExchangeAndQueues();
         }
 
@@ -347,7 +353,7 @@ private function publishOnExchange(\AMQPExchange $exchange, string $body, string
 
     private function setupDelay(int $delay, ?string $routingKey)
     {
-        if ($this->shouldSetup()) {
+        if ($this->autoSetup) {
             $this->setup(); // setup delay exchange and normal exchange for delay queue to DLX messages to
         }
 
@@ -418,23 +424,12 @@ public function get(string $queueName): ?\AMQPEnvelope
     {
         $this->clearWhenDisconnected();
 
-        if ($this->shouldSetup()) {
+        if ($this->autoSetupExchange) {
             $this->setupExchangeAndQueues();
         }
 
-        try {
-            if (false !== $message = $this->queue($queueName)->get()) {
-                return $message;
-            }
-        } catch (\AMQPQueueException $e) {
-            if (404 === $e->getCode() && $this->shouldSetup()) {
-                // If we get a 404 for the queue, it means we need to set up the exchange & queue.
-                $this->setupExchangeAndQueues();
-
-                return $this->get($queueName);
-            }
-
-            throw $e;
+        if (false !== $message = $this->queue($queueName)->get()) {
+            return $message;
         }
 
         return null;
@@ -452,8 +447,11 @@ public function nack(\AMQPEnvelope $message, string $queueName, int $flags = \AM
 
     public function setup(): void
     {
-        $this->setupExchangeAndQueues();
+        if ($this->autoSetupExchange) {
+            $this->setupExchangeAndQueues();
+        }
         $this->getDelayExchange()->declareExchange();
+        $this->autoSetup = false;
     }
 
     private function setupExchangeAndQueues(): void
@@ -466,6 +464,7 @@ private function setupExchangeAndQueues(): void
                 $this->queue($queueName)->bind($this->exchangeOptions['name'], $bindingKey, $queueConfig['binding_arguments'] ?? []);
             }
         }
+        $this->autoSetupExchange = false;
     }
 
     /**
@@ -558,19 +557,6 @@ private function clearWhenDisconnected(): void
         }
     }
 
-    private function shouldSetup(): bool
-    {
-        if (!\array_key_exists('auto_setup', $this->connectionOptions)) {
-            return true;
-        }
-
-        if (\in_array($this->connectionOptions['auto_setup'], [false, 'false'], true)) {
-            return false;
-        }
-
-        return true;
-    }
-
     private function getDefaultPublishRoutingKey(): ?string
     {
         return $this->exchangeOptions['default_publish_routing_key'] ?? null;