[Messenger] fix manual amqp setup when autosetup disabled

Author: Tobion

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

@@ -80,7 +80,7 @@ class Connection
     private $queuesOptions;
     private $amqpFactory;
     private $autoSetupExchange;
-    private $autoSetup;
+    private $autoSetupDelayExchange;
 
     /**
      * @var \AMQPChannel|null
@@ -114,7 +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->autoSetupExchange = $this->autoSetupDelayExchange = $connectionOptions['auto_setup'] ?? true;
         $this->exchangeOptions = $exchangeOptions;
         $this->queuesOptions = $queuesOptions;
         $this->amqpFactory = $amqpFactory ?? new AmqpFactory();
@@ -288,16 +288,16 @@ public function publish(string $body, array $headers = [], int $delayInMs = 0, A
     {
         $this->clearWhenDisconnected();
 
+        if ($this->autoSetupExchange) {
+            $this->setupExchangeAndQueues(); // also setup normal exchange for delayed messages so delay queue can DLX messages to it
+        }
+
         if (0 !== $delayInMs) {
             $this->publishWithDelay($body, $headers, $delayInMs, $amqpStamp);
 
             return;
         }
 
-        if ($this->autoSetupExchange) {
-            $this->setupExchangeAndQueues();
-        }
-
         $this->publishOnExchange(
             $this->exchange(),
             $body,
@@ -356,8 +356,8 @@ private function publishOnExchange(\AMQPExchange $exchange, string $body, string
 
     private function setupDelay(int $delay, ?string $routingKey)
     {
-        if ($this->autoSetup) {
-            $this->setup(); // setup delay exchange and normal exchange for delay queue to DLX messages to
+        if ($this->autoSetupDelayExchange) {
+            $this->setupDelayExchange();
         }
 
         $queue = $this->createDelayQueue($delay, $routingKey);
@@ -450,11 +450,8 @@ public function nack(\AMQPEnvelope $message, string $queueName, int $flags = \AM
 
     public function setup(): void
     {
-        if ($this->autoSetupExchange) {
-            $this->setupExchangeAndQueues();
-        }
-        $this->getDelayExchange()->declareExchange();
-        $this->autoSetup = false;
+        $this->setupExchangeAndQueues();
+        $this->setupDelayExchange();
     }
 
     private function setupExchangeAndQueues(): void
@@ -470,6 +467,12 @@ private function setupExchangeAndQueues(): void
         $this->autoSetupExchange = false;
     }
 
+    private function setupDelayExchange(): void
+    {
+        $this->getDelayExchange()->declareExchange();
+        $this->autoSetupDelayExchange = false;
+    }
+
     /**
      * @return string[]
      */