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

Author: makasim

pkg/amqp-lib/AmqpProducer.php

@@ -4,11 +4,13 @@
 
 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;
@@ -58,6 +60,8 @@ public function __construct(AMQPChannel $channel, AmqpContext $context)
     }
 
     /**
+     * {@inheritdoc}
+     *
      * @param InteropAmqpTopic|InteropAmqpQueue $destination
      * @param InteropAmqpMessage                $message
      */
@@ -70,40 +74,10 @@ public function send(PsrDestination $destination, PsrMessage $message)
 
         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 ($appProperties = $message->getProperties()) {
-            $amqpProperties['application_headers'] = new AMQPTable($appProperties);
-        }
-
-        $amqpMessage = new LibAMQPMessage($message->getBody(), $amqpProperties);
-
-        if ($this->deliveryDelay) {
-            $this->delayStrategy->delayMessage($this->context, $destination, $message, $this->deliveryDelay);
-        } elseif ($destination instanceof InteropAmqpTopic) {
-            $this->channel->basic_publish(
-                $amqpMessage,
-                $destination->getTopicName(),
-                $message->getRoutingKey(),
-                (bool) ($message->getFlags() & InteropAmqpMessage::FLAG_MANDATORY),
-                (bool) ($message->getFlags() & InteropAmqpMessage::FLAG_IMMEDIATE)
-            );
-        } else {
-            $this->channel->basic_publish(
-                $amqpMessage,
-                '',
-                $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);
         }
     }
 
@@ -162,4 +136,43 @@ 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 ($appProperties = $message->getProperties()) {
+            $amqpProperties['application_headers'] = new AMQPTable($appProperties);
+        }
+
+        $amqpMessage = new LibAMQPMessage($message->getBody(), $amqpProperties);
+
+        if ($this->deliveryDelay) {
+            $this->delayStrategy->delayMessage($this->context, $destination, $message, $this->deliveryDelay);
+        } elseif ($destination instanceof InteropAmqpTopic) {
+            $this->channel->basic_publish(
+                $amqpMessage,
+                $destination->getTopicName(),
+                $message->getRoutingKey(),
+                (bool) ($message->getFlags() & InteropAmqpMessage::FLAG_MANDATORY),
+                (bool) ($message->getFlags() & InteropAmqpMessage::FLAG_IMMEDIATE)
+            );
+        } else {
+            $this->channel->basic_publish(
+                $amqpMessage,
+                '',
+                $destination->getQueueName(),
+                (bool) ($message->getFlags() & InteropAmqpMessage::FLAG_MANDATORY),
+                (bool) ($message->getFlags() & InteropAmqpMessage::FLAG_IMMEDIATE)
+            );
+        }
+    }
 }