[amqp-ext] AmqpProducer::send must throw only interop exception.

Author: makasim

pkg/amqp-ext/AmqpProducer.php

@@ -4,11 +4,13 @@
 
 use Enqueue\AmqpTools\DelayStrategyAware;
 use Enqueue\AmqpTools\DelayStrategyAwareTrait;
+use Interop\Amqp\AmqpDestination;
 use Interop\Amqp\AmqpMessage;
 use Interop\Amqp\AmqpProducer as InteropAmqpProducer;
 use Interop\Amqp\AmqpQueue;
 use Interop\Amqp\AmqpTopic;
 use Interop\Queue\DeliveryDelayNotSupportedException;
+use Interop\Queue\Exception;
 use Interop\Queue\InvalidDestinationException;
 use Interop\Queue\InvalidMessageException;
 use Interop\Queue\PsrDestination;
@@ -64,51 +66,14 @@ public function send(PsrDestination $destination, PsrMessage $message)
     {
         $destination instanceof PsrTopic
             ? InvalidDestinationException::assertDestinationInstanceOf($destination, AmqpTopic::class)
-            : InvalidDestinationException::assertDestinationInstanceOf($destination, AmqpQueue::class)
-        ;
+            : InvalidDestinationException::assertDestinationInstanceOf($destination, AmqpQueue::class);
 
         InvalidMessageException::assertMessageInstanceOf($message, AmqpMessage::class);
 
-        if (null !== $this->priority && null === $message->getPriority()) {
-            $message->setPriority($this->priority);
-        }
-
-        if (null !== $this->timeToLive && null === $message->getExpiration()) {
-            $message->setExpiration($this->timeToLive);
-        }
-
-        $amqpAttributes = $message->getHeaders();
-
-        if ($message->getProperties()) {
-            $amqpAttributes['headers'] = $message->getProperties();
-        }
-
-        if ($this->deliveryDelay) {
-            $this->delayStrategy->delayMessage($this->context, $destination, $message, $this->deliveryDelay);
-        } elseif ($destination instanceof AmqpTopic) {
-            $amqpExchange = new \AMQPExchange($this->amqpChannel);
-            $amqpExchange->setType($destination->getType());
-            $amqpExchange->setName($destination->getTopicName());
-            $amqpExchange->setFlags(Flags::convertTopicFlags($destination->getFlags()));
-            $amqpExchange->setArguments($destination->getArguments());
-
-            $amqpExchange->publish(
-                $message->getBody(),
-                $message->getRoutingKey(),
-                Flags::convertMessageFlags($message->getFlags()),
-                $amqpAttributes
-            );
-        } else {
-            $amqpExchange = new \AMQPExchange($this->amqpChannel);
-            $amqpExchange->setType(AMQP_EX_TYPE_DIRECT);
-            $amqpExchange->setName('');
-
-            $amqpExchange->publish(
-                $message->getBody(),
-                $destination->getQueueName(),
-                Flags::convertMessageFlags($message->getFlags()),
-                $amqpAttributes
-            );
+        try {
+            $this->doSend($destination, $message);
+        } catch (\Exception $e) {
+            throw new Exception($e->getMessage(), $e->getCode(), $e);
         }
     }
 
@@ -167,4 +132,50 @@ public function getTimeToLive()
     {
         return $this->timeToLive;
     }
+
+    private function doSend(AmqpDestination $destination, AmqpMessage $message)
+    {
+        if (null !== $this->priority && null === $message->getPriority()) {
+            $message->setPriority($this->priority);
+        }
+
+        if (null !== $this->timeToLive && null === $message->getExpiration()) {
+            $message->setExpiration($this->timeToLive);
+        }
+
+        $amqpAttributes = $message->getHeaders();
+
+        if ($message->getProperties()) {
+            $amqpAttributes['headers'] = $message->getProperties();
+        }
+
+        if ($this->deliveryDelay) {
+            $this->delayStrategy->delayMessage($this->context, $destination, $message, $this->deliveryDelay);
+        } elseif ($destination instanceof AmqpTopic) {
+            $amqpExchange = new \AMQPExchange($this->amqpChannel);
+            $amqpExchange->setType($destination->getType());
+            $amqpExchange->setName($destination->getTopicName());
+            $amqpExchange->setFlags(Flags::convertTopicFlags($destination->getFlags()));
+            $amqpExchange->setArguments($destination->getArguments());
+
+            $amqpExchange->publish(
+                $message->getBody(),
+                $message->getRoutingKey(),
+                Flags::convertMessageFlags($message->getFlags()),
+                $amqpAttributes
+            );
+        } else {
+            /** @var AmqpQueue $destination */
+            $amqpExchange = new \AMQPExchange($this->amqpChannel);
+            $amqpExchange->setType(AMQP_EX_TYPE_DIRECT);
+            $amqpExchange->setName('');
+
+            $amqpExchange->publish(
+                $message->getBody(),
+                $destination->getQueueName(),
+                Flags::convertMessageFlags($message->getFlags()),
+                $amqpAttributes
+            );
+        }
+    }
 }