retry on missed heartbeats (#24)

packaged · Jul 7, 2021 · c7b9c26 · c7b9c26
1 parent 0c21b0b
commit c7b9c26
Showing 1 changed file with 56 additions and 42 deletions.
diff --git a/src/Provider/Amqp/AmqpQueueProvider.php b/src/Provider/Amqp/AmqpQueueProvider.php
@@ -11,6 +11,7 @@
 use PhpAmqpLib\Connection\AbstractConnection;
 use PhpAmqpLib\Connection\AMQPStreamConnection;
 use PhpAmqpLib\Connection\Heartbeat\PCNTLHeartbeatSender;
+use PhpAmqpLib\Exception\AMQPHeartbeatMissedException;
 use PhpAmqpLib\Exception\AMQPProtocolChannelException;
 use PhpAmqpLib\Exception\AMQPRuntimeException;
 use PhpAmqpLib\Exception\AMQPTimeoutException;
@@ -145,63 +146,71 @@ public function pushBatch(array $batch, $persistent = null)
 
     while($needRetry)
     {
-      $needRetry = false;
-
-      $this->_refreshConnection(self::CONN_PUSH);
-      $ch = $this->_getChannel(self::CONN_PUSH);
-
-      if($needDeclare)
+      try
       {
-        $this->_log("Auto-declaring exchange and queue");
-        $declareAttempts++;
-        $this->declareExchange();
-        $this->declareQueue();
-        $this->bindQueue();
-      }
+        $needRetry = false;
 
-      $exchangeName = $this->_getExchangeName();
-      $routingKey = $this->_getRoutingKey();
+        $this->_refreshConnection(self::CONN_PUSH);
+        $ch = $this->_getChannel(self::CONN_PUSH);
 
-      if($mandatory && $returnCallback)
-      {
-        $ch->set_return_listener($returnCallback);
-      }
+        if($needDeclare)
+        {
+          $this->_log("Auto-declaring exchange and queue");
+          $declareAttempts++;
+          $this->declareExchange();
+          $this->declareQueue();
+          $this->bindQueue();
+        }
 
-      foreach($batch as $data)
-      {
-        $ch->batch_basic_publish(
-          $this->_getMessage($data, $persistent),
-          $exchangeName,
-          $routingKey,
-          $mandatory
-        );
-      }
+        $exchangeName = $this->_getExchangeName();
+        $routingKey = $this->_getRoutingKey();
 
-      $ch->publish_batch();
+        if($mandatory && $returnCallback)
+        {
+          $ch->set_return_listener($returnCallback);
+        }
 
-      if($publishConfirm || $mandatory)
-      {
-        try
+        foreach($batch as $data)
         {
-          $ch->wait_for_pending_acks_returns($this->_getPushTimeout());
+          $ch->batch_basic_publish(
+            $this->_getMessage($data, $persistent),
+            $exchangeName,
+            $routingKey,
+            $mandatory
+          );
         }
-        catch(Exception $e)
+
+        $ch->publish_batch();
+
+        if($publishConfirm || $mandatory)
         {
-          $this->disconnect(self::CONN_PUSH);
-          if($autoDeclare
-            && ($declareAttempts < $declareRetryLimit)
-            && ($e->getCode() == 404)
-          )
+          try
           {
-            $needRetry = true;
-            $needDeclare = true;
+            $ch->wait_for_pending_acks_returns($this->_getPushTimeout());
           }
-          else
+          catch(Exception $e)
           {
-            throw $e;
+            $this->disconnect(self::CONN_PUSH);
+            if($autoDeclare
+              && ($declareAttempts < $declareRetryLimit)
+              && ($e->getCode() == 404)
+            )
+            {
+              $needRetry = true;
+              $needDeclare = true;
+            }
+            else
+            {
+              throw $e;
+            }
           }
         }
       }
+      catch(AMQPHeartbeatMissedException $e)
+      {
+        $this->disconnect(self::CONN_PUSH);
+        $needRetry = true;
+      }
     }
     return $this;
   }
@@ -261,6 +270,11 @@ public function consume(callable $callback)
             $this->_fixedConsumerCallback
           );
         }
+        catch(AMQPHeartbeatMissedException $e)
+        {
+          $this->_disconnect(self::CONN_CONSUME);
+          $retry = true;
+        }
         catch(AMQPProtocolChannelException $e)
         {
           if(($e->getCode() == 404)