Merge branch 'fix/refactor-creation-of-connections-and-queues' into feature/support-octane-reconnect

Author: adm-bome

src/Queue/RabbitMQQueue.php

@@ -84,7 +84,7 @@ public function size($queue = null): int
         }
 
         // create a temporary channel, so the main channel will not be closed on exception
-        $channel = $this->getConnection()->channel();
+        $channel = $this->createChannel();
         [, $size] = $channel->queue_declare($queue, true);
         $channel->close();
 
@@ -277,7 +277,7 @@ public function setConnection(AbstractConnection $connection): RabbitMQQueue
     public function getChannel($forceNew = false): AMQPChannel
     {
         if (! $this->channel || $forceNew) {
-            $this->channel = $this->getConnection()->channel();
+            $this->channel = $this->createChannel();
         }
 
         return $this->channel;
@@ -333,7 +333,7 @@ public function isExchangeExists(string $exchange): bool
 
         try {
             // create a temporary channel, so the main channel will not be closed on exception
-            $channel = $this->getConnection()->channel();
+            $channel = $this->createChannel();
             $channel->exchange_declare($exchange, '', true);
             $channel->close();
 
@@ -405,12 +405,20 @@ public function deleteExchange(string $name, bool $unused = false): void
      */
     public function isQueueExists(string $name = null): bool
     {
+        $queueName = $this->getQueue($name);
+
+        if ($this->isQueueDeclared($queueName)) {
+            return true;
+        }
+
         try {
             // create a temporary channel, so the main channel will not be closed on exception
-            $channel = $this->getConnection()->channel();
-            $channel->queue_declare($this->getQueue($name), true);
+            $channel = $this->createChannel();
+            $channel->queue_declare($queueName, true);
             $channel->close();
 
+            $this->queues[] = $queueName;
+
             return true;
         } catch (AMQPProtocolChannelException $exception) {
             if ($exception->amqp_reply_code === 404) {
@@ -457,6 +465,9 @@ public function deleteQueue(string $name, bool $if_unused = false, bool $if_empt
             return;
         }
 
+        $idx = array_search($name, $this->queues);
+        unset($this->queues[$idx]);
+
         $this->getChannel()->queue_delete($name, $if_unused, $if_empty);
     }
 
@@ -482,7 +493,7 @@ public function bindQueue(string $queue, string $exchange, string $routingKey =
     public function purge(string $queue = null): void
     {
         // create a temporary channel, so the main channel will not be closed on exception
-        $channel = $this->getConnection()->channel();
+        $channel = $this->createChannel();
         $channel->queue_purge($this->getQueue($queue));
         $channel->close();
     }
@@ -739,4 +750,9 @@ protected function publishBatch(): void
     {
         $this->getChannel()->publish_batch();
     }
+
+    protected function createChannel(): AMQPChannel
+    {
+        return $this->getConnection()->channel();
+    }
 }