Merge pull request vyuldashev#167 from net53/master

Fix chained jobs attempts

Author: vyuldashev

src/Queue/Jobs/RabbitMQJob.php

@@ -107,7 +107,6 @@ public function release($delay = 0)
         parent::release($delay);
 
         $this->delete();
-        $this->connection->setAttempts($this->attempts() + 1);
 
         $body = $this->payload();
 
@@ -122,11 +121,7 @@ public function release($delay = 0)
 
         $data = $body['data'];
 
-        if ($delay > 0) {
-            $this->connection->later($delay, $job, $data, $this->getQueue());
-        } else {
-            $this->connection->push($job, $data, $this->getQueue());
-        }
+        $this->connection->release($delay, $job, $data, $this->getQueue(), $this->attempts() + 1);
     }
 
     /**

src/Queue/RabbitMQQueue.php

@@ -15,11 +15,6 @@
 
 class RabbitMQQueue extends Queue implements QueueContract
 {
-    /**
-     * Used for retry logic, to set the retries on the message metadata instead of the message body.
-     */
-    const ATTEMPT_COUNT_HEADERS_KEY = 'attempts_count';
-
     protected $sleepOnError;
 
     protected $queueOptions;
@@ -32,7 +27,6 @@ class RabbitMQQueue extends Queue implements QueueContract
      * @var AmqpContext
      */
     private $context;
-    private $retryAfter;
     private $correlationId;
 
     public function __construct(AmqpContext $context, array $config)
@@ -78,8 +72,8 @@ public function pushRaw($payload, $queueName = null, array $options = [])
             $message->setContentType('application/json');
             $message->setDeliveryMode(AmqpMessage::DELIVERY_MODE_PERSISTENT);
 
-            if ($this->retryAfter !== null) {
-                $message->setProperty(self::ATTEMPT_COUNT_HEADERS_KEY, $this->retryAfter);
+            if (isset($options['attempts'])) {
+                $message->setProperty(RabbitMQJob::ATTEMPT_COUNT_HEADERS_KEY, $options['attempts']);
             }
 
             $producer = $this->context->createProducer();
@@ -103,6 +97,24 @@ public function later($delay, $job, $data = '', $queue = null)
         return $this->pushRaw($this->createPayload($job, $data), $queue, ['delay' => $this->secondsUntil($delay)]);
     }
 
+    /**
+     * Release a reserved job back onto the queue.
+     *
+     * @param  \DateTimeInterface|\DateInterval|int  $delay
+     * @param  string|object  $job
+     * @param  mixed   $data
+     * @param  string  $queue
+     * @param  int  $attempts
+     * @return mixed
+     */
+    public function release($delay, $job, $data, $queue, $attempts = 0)
+    {
+        return $this->pushRaw($this->createPayload($job, $data), $queue, [
+            'delay' => $this->secondsUntil($delay),
+            'attempts' => $attempts
+        ]);
+    }
+
     /** @inheritdoc */
     public function pop($queueName = null)
     {
@@ -122,18 +134,6 @@ public function pop($queueName = null)
         return null;
     }
 
-    /**
-     * Sets the attempts member variable to be used in message generation.
-     *
-     * @param int $count
-     *
-     * @return void
-     */
-    public function setAttempts(int $count)
-    {
-        $this->retryAfter = $count;
-    }
-
     /**
      * Retrieves the correlation id, or a unique id.
      *

tests/Queue/RabbitMQQueueTest.php

@@ -34,15 +34,6 @@ public function testCouldBeConstructedWithExpectedArguments()
         new RabbitMQQueue($this->createAmqpContext(), $this->createDummyConfig());
     }
 
-    public function testShouldAllowSetAttempts()
-    {
-        $queue = new RabbitMQQueue($this->createAmqpContext(), $this->createDummyConfig());
-
-        $queue->setAttempts(123);
-
-        $this->assertAttributeSame(123, 'retryAfter', $queue);
-    }
-
     public function testShouldGenerateNewCorrelationIdIfNotSet()
     {
         $queue = new RabbitMQQueue($this->createAmqpContext(), $this->createDummyConfig());
@@ -128,7 +119,7 @@ public function testShouldSendExpectedMessageOnPushRaw()
                 $this->assertSame('application/json', $message->getContentType());
                 $this->assertSame(AmqpMessage::DELIVERY_MODE_PERSISTENT, $message->getDeliveryMode());
                 $this->assertNotEmpty($message->getCorrelationId());
-                $this->assertNull($message->getProperty(RabbitMQQueue::ATTEMPT_COUNT_HEADERS_KEY));
+                $this->assertNull($message->getProperty(RabbitMQJob::ATTEMPT_COUNT_HEADERS_KEY));
             })
         ;
         $producer
@@ -167,7 +158,7 @@ public function testShouldSendExpectedMessageOnPushRaw()
         $queue->pushRaw('thePayload', $expectedQueueName);
     }
 
-    public function testShouldSetAttemptCountHeaderIfNotNull()
+    public function testShouldSetAttemptCountPropIfNotNull()
     {
         $expectedAttempts = 54321;
 
@@ -179,7 +170,7 @@ public function testShouldSetAttemptCountHeaderIfNotNull()
             ->method('send')
             ->with($this->identicalTo($topic), $this->isInstanceOf(AmqpMessage::class))
             ->willReturnCallback(function ($actualTopic, AmqpMessage $message) use ($expectedAttempts) {
-                $this->assertSame($expectedAttempts, $message->getProperty(RabbitMQQueue::ATTEMPT_COUNT_HEADERS_KEY));
+                $this->assertSame($expectedAttempts, $message->getProperty(RabbitMQJob::ATTEMPT_COUNT_HEADERS_KEY));
             })
         ;
         $producer
@@ -212,9 +203,8 @@ public function testShouldSetAttemptCountHeaderIfNotNull()
 
         $queue = new RabbitMQQueue($context, $this->createDummyConfig());
         $queue->setContainer($this->createDummyContainer());
-        $queue->setAttempts($expectedAttempts);
 
-        $queue->pushRaw('thePayload', 'aQueue');
+        $queue->pushRaw('thePayload', 'aQueue', ['attempts' => $expectedAttempts]);
     }
 
     public function testShouldSetDeliveryDelayIfDelayOptionPresent()