improve amqp connection issues

AurelienPillevesse · AurelienPillevesse · commit c4a8995c47af · 2025-01-15T10:19:29.000+01:00
diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/Transport/AmqpReceiver.php b/src/Symfony/Component/Messenger/Bridge/Amqp/Transport/AmqpReceiver.php
@@ -96,6 +96,18 @@ public function ack(Envelope $envelope): void
                 $stamp->getAmqpEnvelope(),
                 $stamp->getQueueName()
             );
+        } catch (\AMQPConnectionException) {
+            try {
+                $stamp = $this->findAmqpStamp($envelope);
+
+                $this->connection->queue($stamp->getQueueName())->getConnection()->reconnect();
+                $this->connection->ack(
+                    $stamp->getAmqpEnvelope(),
+                    $stamp->getQueueName()
+                );
+            } catch (\AMQPException $exception) {
+                throw new TransportException($exception->getMessage(), 0, $exception);
+            }
         } catch (\AMQPException $exception) {
             throw new TransportException($exception->getMessage(), 0, $exception);
         }
@@ -124,6 +136,13 @@ private function rejectAmqpEnvelope(\AMQPEnvelope $amqpEnvelope, string $queueNa
     {
         try {
             $this->connection->nack($amqpEnvelope, $queueName, \AMQP_NOPARAM);
+        } catch (\AMQPConnectionException) {
+            try {
+                $this->connection->queue($queueName)->getConnection()->reconnect();
+                $this->connection->nack($amqpEnvelope, $queueName, \AMQP_NOPARAM);
+            } catch (\AMQPException $exception) {
+                throw new TransportException($exception->getMessage(), 0, $exception);
+            }
         } catch (\AMQPException $exception) {
             throw new TransportException($exception->getMessage(), 0, $exception);
         }
