Merge pull request #7 from yuanxing008/dev

Fix Publisher Reconnection
yuanxing008 · Jun 22, 2020 · a16012e · a16012e
2 parents 961deda + 38cf13a
commit a16012e
Show file tree

Hide file tree

Showing 2 changed files with 48 additions and 11 deletions.
diff --git a/src/AMQPConnection.php b/src/AMQPConnection.php
@@ -10,14 +10,20 @@
 use PhpAmqpLib\Connection\AbstractConnection;
 use PhpAmqpLib\Connection\AMQPSocketConnection;
 use PhpAmqpLib\Connection\AMQPStreamConnection;
+use PhpAmqpLib\Exception\AMQPChannelClosedException;
+use PhpAmqpLib\Exception\AMQPConnectionClosedException;
+use Psr\Log\LoggerAwareInterface;
+use Psr\Log\LoggerAwareTrait;
 
 /**
  * Class AMQPConnection
  *
  * @package JokerProject\LaravelAliyunAmqp
  */
-class AMQPConnection
+class AMQPConnection implements LoggerAwareInterface
 {
+    use LoggerAwareTrait;
+
     /**
      * @const array Default connections parameters
      */
@@ -201,9 +207,35 @@ private function createConnectionByType($type)
      */
     public function reconnect()
     {
-        $this->getConnection()->channel()->close();
-        $this->channel = null;
-        $this->getConnection()->reconnect();
+        try {
+            if (!$this->connection->isConnected()) {
+                $this->connection = $this->getConnection();
+            }
+            if ($this->connection->channel()->is_open()) {
+                $this->connection->channel()->close();
+            }
+            $this->channel = null;
+            $this->getConnection()->reconnect();
+        } catch (AMQPChannelClosedException $e) {
+            $this->logger->info('channel was closed');
+        } catch (AMQPConnectionClosedException $e){
+            $this->getConnection()->reconnect();
+            $this->logger->info('connection was closed');
+        }
+    }
+
+    public function close()
+    {
+        if ($this->channel->is_open()) {
+            $this->channel->close();
+        }
+        if ($this->connection->isConnected()) {
+            try {
+                $this->connection->close();
+            } catch (\Exception $exception) {
+                throw $exception;
+            }
+        }
     }
 
     /**

diff --git a/src/Entity/QueueEntity.php b/src/Entity/QueueEntity.php
@@ -263,20 +263,25 @@ public function publish(string $message, string $routingKey = '')
         }
 
         try {
-            $this->getChannel()
-                ->basic_publish(
-                    new AMQPMessage($message),
-                    '',
-                    $this->attributes['name'],
-                    true
-                );
+            if (!$this->getChannel()->is_open()) {
+                $this->getConnection()->reconnect();
+            }
+            $channel = $this->getChannel();
+            $channel->basic_publish(
+                new AMQPMessage($message),
+                '',
+                $this->attributes['name'],
+                true
+            );
             $this->retryCount = 0;
+            $this->connection->close();
         } catch (AMQPChannelClosedException $exception) {
             $this->retryCount++;
             // Retry publishing with re-connect
             if ($this->retryCount < self::MAX_RETRIES) {
                 $this->getConnection()->reconnect();
                 $this->publish($message, $routingKey);
+                $this->connection->close();
                 return;
             }
             throw $exception;