[amqp-bunny] throw only queue interop exception on Producer::send();

Author: makasim

pkg/amqp-bunny/AmqpProducer.php

@@ -5,11 +5,13 @@
 use Bunny\Channel;
 use Enqueue\AmqpTools\DelayStrategyAware;
 use Enqueue\AmqpTools\DelayStrategyAwareTrait;
+use Interop\Amqp\AmqpDestination as InteropAmqpDestination;
 use Interop\Amqp\AmqpMessage as InteropAmqpMessage;
 use Interop\Amqp\AmqpProducer as InteropAmqpProducer;
 use Interop\Amqp\AmqpQueue as InteropAmqpQueue;
 use Interop\Amqp\AmqpTopic as InteropAmqpTopic;
 use Interop\Queue\DeliveryDelayNotSupportedException;
+use Interop\Queue\Exception;
 use Interop\Queue\InvalidDestinationException;
 use Interop\Queue\InvalidMessageException;
 use Interop\Queue\PsrDestination;
@@ -56,55 +58,24 @@ public function __construct(Channel $channel, AmqpContext $context)
     }
 
     /**
+     * {@inheritdoc}
+     *
      * @param InteropAmqpTopic|InteropAmqpQueue $destination
      * @param InteropAmqpMessage                $message
      */
     public function send(PsrDestination $destination, PsrMessage $message)
     {
         $destination instanceof PsrTopic
             ? InvalidDestinationException::assertDestinationInstanceOf($destination, InteropAmqpTopic::class)
-            : InvalidDestinationException::assertDestinationInstanceOf($destination, InteropAmqpQueue::class);
+            : InvalidDestinationException::assertDestinationInstanceOf($destination, InteropAmqpQueue::class)
+        ;
 
         InvalidMessageException::assertMessageInstanceOf($message, InteropAmqpMessage::class);
 
-        if (null !== $this->priority && null === $message->getPriority()) {
-            $message->setPriority($this->priority);
-        }
-
-        if (null !== $this->timeToLive && null === $message->getExpiration()) {
-            $message->setExpiration($this->timeToLive);
-        }
-
-        $amqpProperties = $message->getHeaders();
-
-        if (array_key_exists('timestamp', $amqpProperties) && null !== $amqpProperties['timestamp']) {
-            $amqpProperties['timestamp'] = \DateTime::createFromFormat('U', $amqpProperties['timestamp']);
-        }
-
-        if ($appProperties = $message->getProperties()) {
-            $amqpProperties['application_headers'] = $appProperties;
-        }
-
-        if ($this->deliveryDelay) {
-            $this->delayStrategy->delayMessage($this->context, $destination, $message, $this->deliveryDelay);
-        } elseif ($destination instanceof InteropAmqpTopic) {
-            $this->channel->publish(
-                $message->getBody(),
-                $amqpProperties,
-                $destination->getTopicName(),
-                $message->getRoutingKey(),
-                (bool) ($message->getFlags() & InteropAmqpMessage::FLAG_MANDATORY),
-                (bool) ($message->getFlags() & InteropAmqpMessage::FLAG_IMMEDIATE)
-            );
-        } else {
-            $this->channel->publish(
-                $message->getBody(),
-                $amqpProperties,
-                '',
-                $destination->getQueueName(),
-                (bool) ($message->getFlags() & InteropAmqpMessage::FLAG_MANDATORY),
-                (bool) ($message->getFlags() & InteropAmqpMessage::FLAG_IMMEDIATE)
-            );
+        try {
+            $this->doSend($destination, $message);
+        } catch (\Exception $e) {
+            throw new Exception($e->getMessage(), $e->getCode(), $e);
         }
     }
 
@@ -163,4 +134,47 @@ public function getTimeToLive()
     {
         return $this->timeToLive;
     }
+
+    private function doSend(InteropAmqpDestination $destination, InteropAmqpMessage $message)
+    {
+        if (null !== $this->priority && null === $message->getPriority()) {
+            $message->setPriority($this->priority);
+        }
+
+        if (null !== $this->timeToLive && null === $message->getExpiration()) {
+            $message->setExpiration($this->timeToLive);
+        }
+
+        $amqpProperties = $message->getHeaders();
+
+        if (array_key_exists('timestamp', $amqpProperties) && null !== $amqpProperties['timestamp']) {
+            $amqpProperties['timestamp'] = \DateTime::createFromFormat('U', $amqpProperties['timestamp']);
+        }
+
+        if ($appProperties = $message->getProperties()) {
+            $amqpProperties['application_headers'] = $appProperties;
+        }
+
+        if ($this->deliveryDelay) {
+            $this->delayStrategy->delayMessage($this->context, $destination, $message, $this->deliveryDelay);
+        } elseif ($destination instanceof InteropAmqpTopic) {
+            $this->channel->publish(
+                $message->getBody(),
+                $amqpProperties,
+                $destination->getTopicName(),
+                $message->getRoutingKey(),
+                (bool) ($message->getFlags() & InteropAmqpMessage::FLAG_MANDATORY),
+                (bool) ($message->getFlags() & InteropAmqpMessage::FLAG_IMMEDIATE)
+            );
+        } else {
+            $this->channel->publish(
+                $message->getBody(),
+                $amqpProperties,
+                '',
+                $destination->getQueueName(),
+                (bool) ($message->getFlags() & InteropAmqpMessage::FLAG_MANDATORY),
+                (bool) ($message->getFlags() & InteropAmqpMessage::FLAG_IMMEDIATE)
+            );
+        }
+    }
 }