SWR-18338: Close connections on switch

Author: pupsick

composer.json

@@ -34,7 +34,7 @@
     },
     "extra": {
         "branch-alias": {
-            "dev-master": "1.21-dev"
+            "dev-master": "1.22-dev"
         },
         "laravel": {
             "providers": [

src/Consumer.php

@@ -32,6 +32,8 @@ class Consumer extends Worker
     /** @var AMQPChannel */
     protected $channel;
 
+    protected $connection;
+
     /** @var object|null */
     protected $currentJob;
 

src/VhostsConsumers/AbstractVhostsConsumer.php

@@ -74,8 +74,7 @@ public function __construct(
         ExceptionHandler $exceptions,
         callable $isDownForMaintenance,
         callable $resetScope = null
-    )
-    {
+    ) {
         parent::__construct($manager, $events, $exceptions, $isDownForMaintenance, $resetScope);
     }
 
@@ -124,7 +123,7 @@ public function daemon($connectionName, $queue, WorkerOptions $options)
 
         $this->vhostDaemon($connectionName, $options);
     }
-    
+
     abstract protected function vhostDaemon($connectionName, WorkerOptions $options);
 
     /**
@@ -141,8 +140,7 @@ protected function getStopStatus(
         $startTime = 0,
         $jobsProcessed = 0,
         bool $hasJob = false
-    ): ?int
-    {
+    ): ?int {
         return match (true) {
             $this->shouldQuit => static::EXIT_SUCCESS,
             $this->memoryExceeded($options->memory) => static::EXIT_MEMORY_LIMIT,
@@ -566,16 +564,14 @@ protected function updateLastProcessedAt()
     /**
      * @return RabbitMQQueue
      */
-    protected function initConnection(): RabbitMQQueue 
+    protected function initConnection(): RabbitMQQueue
     {
-        // Close any existing connection/channel
-        if ($this->channel) {
+        if ($this->connection) {
             try {
-                $this->channel->close();
+                $this->connection->close();
             } catch (\Exception $e) {
-                // Ignore close errors
             }
-            $this->channel = null;
+            $this->connection = null;
         }
 
         $connection = $this->manager->connection(
@@ -584,6 +580,19 @@ protected function initConnection(): RabbitMQQueue
 
         try {
             $channel = $connection->getChannel();
+
+            $this->currentConnectionName = $connection->getConnectionName();
+
+            $this->connectionMutex->lock(self::MAIN_HANDLER_LOCK);
+            $channel->basic_qos(
+                $this->prefetchSize,
+                $this->prefetchCount,
+                false
+            );
+            $this->connectionMutex->unlock(self::MAIN_HANDLER_LOCK);
+
+            $this->channel = $channel;
+            $this->connection = $connection;
         } catch (AMQPConnectionClosedException $exception) {
             $this->logError('initConnection.exception', [
                 'message' => $exception->getMessage(),
@@ -601,18 +610,6 @@ protected function initConnection(): RabbitMQQueue
             return $this->initConnection();
         }
 
-        $this->currentConnectionName = $connection->getConnectionName();
-
-        $this->connectionMutex->lock(self::MAIN_HANDLER_LOCK);
-        $channel->basic_qos(
-            $this->prefetchSize,
-            $this->prefetchCount,
-            false
-        );
-        $this->connectionMutex->unlock(self::MAIN_HANDLER_LOCK);
-
-        $this->channel = $channel;
-
         return $connection;
     }
 
@@ -686,4 +683,4 @@ protected function log(string $message, array $data = [], bool $isError = false)
             $this->logger->info($logMessage, $data);
         }
     }
-}
+}