improve amqp connection issues

Author: AurelienPillevesse

src/Symfony/Component/Messenger/Bridge/Amqp/Transport/AmqpReceiver.php

@@ -92,10 +92,16 @@ public function ack(Envelope $envelope): void
         try {
             $stamp = $this->findAmqpStamp($envelope);
 
-            $this->connection->ack(
-                $stamp->getAmqpEnvelope(),
-                $stamp->getQueueName()
-            );
+            $this->connection->ack($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 +130,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);
         }