Updated exception message in AmqpSender, updated AmqpReceiver to throw new TransportException

Author: lolmx

Tests/Transport/AmqpExt/AmqpReceiverTest.php

@@ -101,6 +101,83 @@ public function testItRejectsTheMessageIfTheExceptionIsARejectMessageExceptionIn
             throw new WillNeverWorkException('Well...');
         });
     }
+
+    /**
+     * @expectedException \Symfony\Component\Messenger\Exception\TransportException
+     */
+    public function testItThrowsATransportExceptionIfItCannotAcknowledgeMessage()
+    {
+        $serializer = new Serializer(
+            new SerializerComponent\Serializer([new ObjectNormalizer()], ['json' => new JsonEncoder()])
+        );
+
+        $envelope = $this->getMockBuilder(\AMQPEnvelope::class)->getMock();
+        $envelope->method('getBody')->willReturn('{"message": "Hi"}');
+        $envelope->method('getHeaders')->willReturn([
+            'type' => DummyMessage::class,
+        ]);
+
+        $connection = $this->getMockBuilder(Connection::class)->disableOriginalConstructor()->getMock();
+        $connection->method('get')->willReturn($envelope);
+
+        $connection->method('ack')->with($envelope)->willThrowException(new \AMQPException());
+
+        $receiver = new AmqpReceiver($connection, $serializer);
+        $receiver->receive(function (?Envelope $envelope) use ($receiver) {
+            $receiver->stop();
+        });
+    }
+
+    /**
+     * @expectedException \Symfony\Component\Messenger\Exception\TransportException
+     */
+    public function testItThrowsATransportExceptionIfItCannotRejectMessage()
+    {
+        $serializer = new Serializer(
+            new SerializerComponent\Serializer([new ObjectNormalizer()], ['json' => new JsonEncoder()])
+        );
+
+        $envelope = $this->getMockBuilder(\AMQPEnvelope::class)->getMock();
+        $envelope->method('getBody')->willReturn('{"message": "Hi"}');
+        $envelope->method('getHeaders')->willReturn([
+            'type' => DummyMessage::class,
+        ]);
+
+        $connection = $this->getMockBuilder(Connection::class)->disableOriginalConstructor()->getMock();
+        $connection->method('get')->willReturn($envelope);
+        $connection->method('reject')->with($envelope)->willThrowException(new \AMQPException());
+
+        $receiver = new AmqpReceiver($connection, $serializer);
+        $receiver->receive(function () {
+            throw new WillNeverWorkException('Well...');
+        });
+    }
+
+    /**
+     * @expectedException \Symfony\Component\Messenger\Exception\TransportException
+     */
+    public function testItThrowsATransportExceptionIfItCannotNonAcknowledgeMessage()
+    {
+        $serializer = new Serializer(
+            new SerializerComponent\Serializer([new ObjectNormalizer()], ['json' => new JsonEncoder()])
+        );
+
+        $envelope = $this->getMockBuilder(\AMQPEnvelope::class)->getMock();
+        $envelope->method('getBody')->willReturn('{"message": "Hi"}');
+        $envelope->method('getHeaders')->willReturn([
+            'type' => DummyMessage::class,
+        ]);
+
+        $connection = $this->getMockBuilder(Connection::class)->disableOriginalConstructor()->getMock();
+        $connection->method('get')->willReturn($envelope);
+
+        $connection->method('nack')->with($envelope)->willThrowException(new \AMQPException());
+
+        $receiver = new AmqpReceiver($connection, $serializer);
+        $receiver->receive(function () {
+            throw new InterruptException('Well...');
+        });
+    }
 }
 
 class InterruptException extends \Exception

Transport/AmqpExt/AmqpReceiver.php

@@ -11,6 +11,7 @@
 
 namespace Symfony\Component\Messenger\Transport\AmqpExt;
 
+use Symfony\Component\Messenger\Exception\TransportException;
 use Symfony\Component\Messenger\Transport\AmqpExt\Exception\RejectMessageExceptionInterface;
 use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
 use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;
@@ -61,11 +62,21 @@ public function receive(callable $handler): void
 
                 $this->connection->ack($AMQPEnvelope);
             } catch (RejectMessageExceptionInterface $e) {
-                $this->connection->reject($AMQPEnvelope);
+                try {
+                    $this->connection->reject($AMQPEnvelope);
+                } catch (\AMQPException $exception) {
+                    throw new TransportException($exception->getMessage(), 0, $exception);
+                }
 
                 throw $e;
+            } catch (\AMQPException $e) {
+                throw new TransportException($e->getMessage(), 0, $e);
             } catch (\Throwable $e) {
-                $this->connection->nack($AMQPEnvelope, AMQP_REQUEUE);
+                try {
+                    $this->connection->nack($AMQPEnvelope, AMQP_REQUEUE);
+                } catch (\AMQPException $exception) {
+                    throw new TransportException($exception->getMessage(), 0, $exception);
+                }
 
                 throw $e;
             } finally {

Transport/AmqpExt/AmqpSender.php

@@ -45,7 +45,7 @@ public function send(Envelope $envelope): Envelope
         try {
             $this->connection->publish($encodedMessage['body'], $encodedMessage['headers']);
         } catch (\AMQPException $e) {
-            throw new TransportException('Current transport was not able to send given message, please try again', 0, $e);
+            throw new TransportException($e->getMessage(), 0, $e);
         }
 
         return $envelope;