[Messenger] Added new TransportException which is thrown if transport could not send a message

Author: lolmx

CHANGELOG.md

@@ -11,6 +11,11 @@ CHANGELOG
    changed from `Serializer` to `PhpSerializer` inside `AmqpReceiver`,
    `AmqpSender`, `AmqpTransport` and `AmqpTransportFactory`.
 
+ * Added `TransportException` to mark an exception transport-related
+
+ * [BC BREAK] If listening to exceptions while using `AmqpSender`, `\AMQPException` is
+   no longer thrown in favor of `TransportException`.
+
 4.2.0
 -----
 

Exception/TransportException.php

@@ -0,0 +1,21 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <[email protected]>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Exception;
+
+/**
+ * @author Eric Masoero <[email protected]>
+ *
+ * @experimental in 4.2
+ */
+class TransportException extends RuntimeException
+{
+}

Tests/Transport/AmqpExt/AmqpSenderTest.php

@@ -37,4 +37,22 @@ public function testItSendsTheEncodedMessage()
         $sender = new AmqpSender($connection, $serializer);
         $sender->send($envelope);
     }
+
+    /**
+     * @expectedException Symfony\Component\Messenger\Exception\TransportException
+     */
+    public function testItThrowsATransportExceptionIfItCannotSendTheMessage()
+    {
+        $envelope = new Envelope(new DummyMessage('Oy'));
+        $encoded = ['body' => '...', 'headers' => ['type' => DummyMessage::class]];
+
+        $serializer = $this->getMockBuilder(SerializerInterface::class)->getMock();
+        $serializer->method('encode')->with($envelope)->willReturnOnConsecutiveCalls($encoded);
+
+        $connection = $this->getMockBuilder(Connection::class)->disableOriginalConstructor()->getMock();
+        $connection->method('publish')->with($encoded['body'], $encoded['headers'])->willThrowException(new \AMQPException());
+
+        $sender = new AmqpSender($connection, $serializer);
+        $sender->send($envelope);
+    }
 }

Transport/AmqpExt/AmqpSender.php

@@ -12,6 +12,7 @@
 namespace Symfony\Component\Messenger\Transport\AmqpExt;
 
 use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Exception\TransportException;
 use Symfony\Component\Messenger\Transport\Sender\SenderInterface;
 use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;
 use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
@@ -41,7 +42,11 @@ public function send(Envelope $envelope): Envelope
     {
         $encodedMessage = $this->serializer->encode($envelope);
 
-        $this->connection->publish($encodedMessage['body'], $encodedMessage['headers']);
+        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');
+        }
 
         return $envelope;
     }