fix retry strategies, add delay support.

symfony · makasim · Jun 27, 2017 · May 22, 2017 · Jul 5, 2017 · Jul 14, 2017
commit 43e82cca0d4666aa6c7a75ec7710321deb54a73e
diff --git a/src/Symfony/Component/Amqp/Broker.php b/src/Symfony/Component/Amqp/Broker.php
@@ -11,6 +11,8 @@
 
 namespace Symfony\Component\Amqp;
 
+use Enqueue\AmqpTools\DelayStrategyAware;
+use Enqueue\AmqpTools\RabbitMqDlxDelayStrategy;
 use Interop\Amqp\AmqpConsumer;
 use Interop\Amqp\AmqpContext;
 use Interop\Amqp\AmqpTopic;
@@ -394,7 +396,16 @@ public function publish($routingKey, $message, array $attributes = array())
         $amqpMessage->setRoutingKey($routingKey);
         $amqpMessage->setDeliveryMode(AmqpMessage::DELIVERY_MODE_PERSISTENT);
 
-        $this->context->createProducer()->send($topic, $amqpMessage);
+        $producer = $this->context->createProducer();
+
+        if (isset($attributes['delay']) && $producer instanceof DelayStrategyAware) {
+            $producer
+                ->setDelayStrategy(new RabbitMqDlxDelayStrategy())
+                ->setDeliveryDelay($attributes['delay'] * 1000)
+            ;
+        }
+
+        $producer->send($topic, $amqpMessage);
 
         return true;
     }
@@ -417,13 +428,7 @@ public function publish($routingKey, $message, array $attributes = array())
      */
     public function delay($routingKey, $message, $delay, array $attributes = array())
     {
-        $exchangeName = isset($attributes['exchange']) ? $attributes['exchange'] : self::DEFAULT_EXCHANGE;
-
-        $this->createDelayedQueue($routingKey, $delay, $exchangeName);
-
-        $attributes['exchange'] = self::DEAD_LETTER_EXCHANGE;
-        $attributes['headers']['queue-time'] = (string) $delay;
-        $attributes['headers']['exchange'] = (string) $exchangeName;
+        $attributes['delay'] = $delay;
 
         return $this->publish($routingKey, $message, $attributes);
     }
@@ -496,11 +501,11 @@ public function ack(AmqpMessage $message, $queueName = null)
      *
      * @return bool
      */
-    public function nack(AmqpMessage $message, $queueName = null)
+    public function nack(AmqpMessage $message, $queueName = null, $requeue = false)
     {
         $queueName = $queueName ?: $message->getRoutingKey();
 
-        $this->getQueueConsumer($queueName)->reject($message, false);
+        $this->getQueueConsumer($queueName)->reject($message, $requeue);
     }
 
     /**

diff --git a/src/Symfony/Component/Amqp/Helper/MessageExporter.php b/src/Symfony/Component/Amqp/Helper/MessageExporter.php
@@ -62,9 +62,9 @@ public function export($queueName, $ack = false)
         $phar = new \PharData($filename);
         foreach ($messages as $i => $message) {
             if ($ack) {
-                $this->broker->ack($message, null, $queueName);
+                $this->broker->ack($message, $queueName);
             } else {
-                $this->broker->nack($message, \AMQP_REQUEUE, $queueName);
+                $this->broker->nack($message, $queueName);
             }
             $buffer = '';
             foreach ($message->getHeaders() as $name => $value) {

diff --git a/src/Symfony/Component/Amqp/RetryStrategy/ConstantRetryStrategy.php b/src/Symfony/Component/Amqp/RetryStrategy/ConstantRetryStrategy.php
@@ -11,6 +11,7 @@
 
 namespace Symfony\Component\Amqp\RetryStrategy;
 
+use Interop\Amqp\AmqpMessage;
 use Symfony\Component\Amqp\Exception\InvalidArgumentException;
 
 /**
@@ -41,17 +42,17 @@ public function __construct($time, $max = 0)
     /**
      * {@inheritdoc}
      */
-    public function isRetryable(\AMQPEnvelope $msg)
+    public function isRetryable(AmqpMessage $msg)
     {
-        $retries = (int) $msg->getHeader('retries');
+        $retries = (int) $msg->getProperty('retries');
 
         return $this->max ? $retries < $this->max : true;
     }
 
     /**
      * {@inheritdoc}
      */
-    public function getWaitingTime(\AMQPEnvelope $msg)
+    public function getWaitingTime(AmqpMessage $msg)
     {
         return $this->time;
     }

diff --git a/src/Symfony/Component/Amqp/RetryStrategy/ExponentialRetryStrategy.php b/src/Symfony/Component/Amqp/RetryStrategy/ExponentialRetryStrategy.php
@@ -11,6 +11,8 @@
 
 namespace Symfony\Component\Amqp\RetryStrategy;
 
+use Interop\Amqp\AmqpMessage;
+
 /**
  * The retry mechanism is based on a truncated exponential backoff algorithm.
  *
@@ -35,23 +37,23 @@ public function __construct($max = 0, $offset = 0)
     /**
      * {@inheritdoc}
      */
-    public function isRetryable(\AMQPEnvelope $msg)
+    public function isRetryable(AmqpMessage $msg)
     {
         if (0 === $this->max) {
             return true;
         }
 
-        $retries = (int) $msg->getHeader('retries');
+        $retries = (int) $msg->getProperty('retries', 0);
 
         return $retries < $this->max;
     }
 
     /**
      * {@inheritdoc}
      */
-    public function getWaitingTime(\AMQPEnvelope $msg)
+    public function getWaitingTime(AmqpMessage $msg)
     {
-        $retries = (int) $msg->getHeader('retries');
+        $retries = (int) $msg->getProperty('retries', 0);
 
         return pow(2, $retries + $this->offset);
     }

diff --git a/src/Symfony/Component/Amqp/RetryStrategy/RetryStrategyInterface.php b/src/Symfony/Component/Amqp/RetryStrategy/RetryStrategyInterface.php
@@ -11,23 +11,25 @@
 
 namespace Symfony\Component\Amqp\RetryStrategy;
 
+use Interop\Amqp\AmqpMessage;
+
 /**
  * @author Fabien Potencier <fabien@symfony.com>
  * @author Grégoire Pineau <lyrixx@lyrixx.info>
  */
 interface RetryStrategyInterface
 {
     /**
-     * @param \AMQPEnvelope $msg
+     * @param AmqpMessage $msg
      *
      * @return bool
      */
-    public function isRetryable(\AMQPEnvelope $msg);
+    public function isRetryable(AmqpMessage $msg);
 
     /**
-     * @param \AMQPEnvelope $msg
+     * @param AmqpMessage $msg
      *
      * @return int
      */
-    public function getWaitingTime(\AMQPEnvelope $msg);
+    public function getWaitingTime(AmqpMessage $msg);
 }
diff --git a/src/Symfony/Component/Amqp/composer.json b/src/Symfony/Component/Amqp/composer.json
@@ -23,10 +23,12 @@
         "php": ">=5.5.9",
         "psr/log": "~1.0",
         "symfony/event-dispatcher": "^2.3|^3.0|^4.0",
-        "queue-interop/amqp-interop": "*@dev"
+        "queue-interop/amqp-interop": "^0.6",
+        "enqueue/amqp-tools": "^0.7"
     },
     "require-dev": {
-        "symfony/phpunit-bridge": "^3.3"
+        "symfony/phpunit-bridge": "^3.3",
+        "enqueue/amqp-bunny": "^0.7"
     },
     "autoload": {
         "psr-4": { "Symfony\\Component\\Amqp\\": "" },

diff --git a/src/Symfony/Component/Worker/MessageFetcher/AmqpMessageFetcher.php b/src/Symfony/Component/Worker/MessageFetcher/AmqpMessageFetcher.php
@@ -11,6 +11,7 @@
 
 namespace Symfony\Component\Worker\MessageFetcher;
 
+use Interop\Amqp\AmqpConsumer;
 use Symfony\Component\Amqp\Broker;
 use Symfony\Component\Worker\MessageCollection;
 
@@ -27,7 +28,7 @@ public function __construct(Broker $broker, $queueName, $autoAck = false)
     {
         $this->broker = $broker;
         $this->queueName = $queueName;
-        $this->flags = $autoAck ? \AMQP_AUTOACK : \AMQP_NOPARAM;
+        $this->flags = $autoAck ? AmqpConsumer::FLAG_NOACK : AmqpConsumer::FLAG_NOPARAM;
     }
 
     public function fetchMessages()