Fixes for Attempt/retry logic

Small cleanups and helper method adjustments

Author: dustindoiron

src/VladimirYuldashev/LaravelQueueRabbitMQ/Queue/Jobs/RabbitMQJob.php

@@ -11,12 +11,26 @@
 
 class RabbitMQJob extends Job implements JobContract
 {
+    /**
+     * Same as RabbitMQQueue, used for attempt counts
+     */
+    const ATTEMPT_COUNT_HEADERS_KEY = 'attempts_count';
 
     protected $connection;
     protected $channel;
     protected $queue;
     protected $message;
 
+    /**
+     * Creates a new instance of RabbitMQJob.
+     *
+     * @param Illuminate\Container\Container $container
+     * @param VladimirYuldashev\LaravelQueueRabbitMQ\Queue\RabbitMQQueue $connection
+     * @param PhpAmqpLib\Channel\AMQPChannel $channel
+     * @param string $queue
+     * @param PhpAmqpLib\Message\AMQPMessage $message
+     *
+     */
     public function __construct(
         Container $container,
         RabbitMQQueue $connection,
@@ -38,7 +52,7 @@ public function __construct(
      */
     public function fire()
     {
-        $this->resolveAndFire(json_decode($this->message->body, true));
+        $this->resolveAndFire($this->getParsedBody());
     }
 
     /**
@@ -51,6 +65,16 @@ public function getRawBody()
         return $this->message->body;
     }
 
+    /**
+     * Retrieves the parsed body for the job.
+     *
+     * @return array|false
+     */
+    public function getParsedBody()
+    {
+        return json_decode($this->getRawBody(), true);
+    }
+
     /**
      * Delete the job from the queue.
      *
@@ -59,12 +83,11 @@ public function getRawBody()
     public function delete()
     {
         parent::delete();
-
-        $this->channel->basic_ack($this->message->delivery_info['delivery_tag']);
+        $this->channel->basic_ack($this->message->get('delivery_tag'));
     }
 
     /**
-     * Get queue name
+     * Get the queue name.
      *
      * @return string
      */
@@ -83,16 +106,18 @@ public function getQueue()
     public function release($delay = 0)
     {
         $this->delete();
+        $this->setAttempts($this->attempts() + 1);
 
-        $body = $this->message->body;
-        $body = json_decode($body, true);
-
-        $attempts = $this->attempts();
+        $body = $this->getParsedBody();
 
-        $job = unserialize($body['data']['command']);
-
-        // write attempts to job
-        $job->attempts = $attempts + 1;
+        /**
+         * Some jobs don't have the command set, so fall back to just sending it the job name string
+         */
+        if (isset($body['data']['command']) === true) {
+            $job = unserialize($body['data']['command']);
+        } else {
+            $job = $this->getName();
+        }
 
         $data = $body['data'];
 
@@ -110,15 +135,28 @@ public function release($delay = 0)
      */
     public function attempts()
     {
-        $body = json_decode($this->message->body, true);
-        $job = unserialize($body['data']['command']);
-        if (is_object($job) && property_exists($job, 'attempts'))
-        {
-            return (int) $job->attempts;
+        if ($this->message->has('application_headers') === true) {
+            $headers = $this->message->get('application_headers')->getNativeData();
+
+            if (isset($headers[self::ATTEMPT_COUNT_HEADERS_KEY]) === true) {
+                return $headers[self::ATTEMPT_COUNT_HEADERS_KEY];
+            }
         }
         return 0;
     }
 
+    /**
+     * Sets the count of attempts at processing this job.
+     *
+     * @param int $count
+     *
+     * @return void
+     */
+    private function setAttempts($count)
+    {
+        $this->connection->setAttempts($count);
+    }
+
     /**
      * Get the job identifier.
      *
@@ -128,5 +166,4 @@ public function getJobId()
     {
         return $this->message->get('correlation_id');
     }
-
 }

src/VladimirYuldashev/LaravelQueueRabbitMQ/Queue/RabbitMQQueue.php

@@ -13,6 +13,10 @@
 
 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 $connection;
     protected $channel;
@@ -24,6 +28,11 @@ class RabbitMQQueue extends Queue implements QueueContract
     protected $configQueue;
     protected $configExchange;
 
+    /**
+     * @var int
+     */
+    private $attempts;
+
     /**
      * @param AMQPStreamConnection $amqpConnection
      * @param array $config
@@ -72,11 +81,18 @@ public function pushRaw($payload, $queue = null, array $options = [])
         } else {
             list($queue, $exchange) = $this->declareQueue($queue);
         }
-        // push job to a queue
-        $message = new AMQPMessage($payload, [
+
+        $headers = [
             'Content-Type' => 'application/json',
             'delivery_mode' => 2,
-        ]);
+        ];
+
+        if (isset($this->attempts) === true) {
+            $headers['application_headers'] = [self::ATTEMPT_COUNT_HEADERS_KEY => ['I', $this->attempts]];
+        }
+
+        // push job to a queue
+        $message = new AMQPMessage($payload, $headers);
 
         // push task to a queue
         $this->channel->basic_publish($message, $exchange, $queue);
@@ -222,4 +238,15 @@ private function declareDelayedQueue($destination, $delay)
         return [$name, $exchange];
     }
 
-}
+    /**
+     * Sets the attempts member variable to be used in message generation
+     *
+     * @param int $count
+     *
+     * @return void
+     */
+    public function setAttempts($count)
+    {
+        $this->attempts = $count;
+    }
+}