bug symfony#59508 [Messenger] [AMQP] Improve AMQP connection issues (AurelienPillevesse)

This PR was merged into the 7.1 branch.

Discussion
----------

[Messenger] [AMQP] Improve AMQP connection issues

| Q             | A
| ------------- | ---
| Branch?       | 7.1
| Bug fix?      | yes
| New feature?  | no
| Deprecations? | no
| License       | MIT

Appy this fix symfony#53892 for :
- `ack` (https://github.com/php-amqp/php-amqp/blob/latest/stubs/AMQPQueue.php#L50)
- `nack` (https://github.com/php-amqp/php-amqp/blob/latest/stubs/AMQPQueue.php#L238)

Because `\AMQPConnectionException` can also happen here

Commits
-------

5356023 improve amqp connection issues

Author: nicolas-grekas

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);
         }