Use custom ID and fix attempts count

Author: kainxspirits

src/Jobs/PubSubJob.php

@@ -40,6 +40,8 @@ public function __construct(Container $container, PubSubQueue $pubsub, Message $
         $this->queue = $queue;
         $this->container = $container;
         $this->connectionName = $connectionName;
+
+        $this->decoded = $this->payload();
     }
 
     /**
@@ -49,7 +51,7 @@ public function __construct(Container $container, PubSubQueue $pubsub, Message $
      */
     public function getJobId()
     {
-        return $this->job->id();
+        return $this->decoded['id'] ?? null;
     }
 
     /**
@@ -81,7 +83,7 @@ public function delete()
      */
     public function attempts()
     {
-        return (int) $this->job->attributes('attempts') ?? 0;
+        return ((int) $this->job->attribute('attempts') ?? 0) + 1;
     }
 
     /**
@@ -94,7 +96,8 @@ public function release($delay = 0)
     {
         parent::release($delay);
 
-        $attempts = $this->attempts() + 1;
+        $attempts = $this->attempts();
+
         $this->pubsub->acknowledgeAndPublish(
             $this->job,
             $this->queue,

src/PubSubQueue.php

@@ -3,6 +3,7 @@
 namespace Kainxspirits\PubSubQueue;
 
 use Illuminate\Queue\Queue;
+use Illuminate\Support\Str;
 use Google\Cloud\PubSub\Topic;
 use Google\Cloud\PubSub\Message;
 use Google\Cloud\PubSub\PubSubClient;
@@ -62,7 +63,7 @@ public function size($queue = null)
      */
     public function push($job, $data = '', $queue = null)
     {
-        return $this->pushRaw($this->createPayload($job, $queue, $data), $queue);
+        return $this->pushRaw($this->createPayload($job, $this->getQueue($queue), $data), $queue);
     }
 
     /**
@@ -86,7 +87,11 @@ public function pushRaw($payload, $queue = null, array $options = [])
             $publish['attributes'] = $options;
         }
 
-        return $topic->publish($publish);
+        $topic->publish($publish);
+
+        $decoded_payload = json_decode(base64_decode($payload), true);
+
+        return $decoded_payload['id'];
     }
 
     /**
@@ -102,7 +107,7 @@ public function pushRaw($payload, $queue = null, array $options = [])
     public function later($delay, $job, $data = '', $queue = null)
     {
         return $this->pushRaw(
-            $this->createPayload($job, $queue, $data),
+            $this->createPayload($job, $this->getQueue($queue), $data),
             $queue,
             ['available_at' => $this->availableAt($delay)]
         );
@@ -116,7 +121,7 @@ public function later($delay, $job, $data = '', $queue = null)
      */
     public function pop($queue = null)
     {
-        $topic = $this->getTopic($queue);
+        $topic = $this->getTopic($this->getQueue($queue));
 
         if (! $topic->exists()) {
             return;
@@ -134,7 +139,7 @@ public function pop($queue = null)
                 $this,
                 $messages[0],
                 $this->connectionName,
-                $queue
+                $this->getQueue($queue)
             );
         }
     }
@@ -153,10 +158,10 @@ public function bulk($jobs, $data = '', $queue = null)
         $payloads = [];
 
         foreach ((array) $jobs as $job) {
-            $payloads[] = ['data' => $this->createPayload($job, $queue, $data)];
+            $payloads[] = ['data' => $this->createPayload($job, $this->getQueue($queue), $data)];
         }
 
-        $topic = $this->getTopic($queue, true);
+        $topic = $this->getTopic($this->getQueue($queue), true);
 
         $this->subscribeToTopic($topic);
 
@@ -171,7 +176,7 @@ public function bulk($jobs, $data = '', $queue = null)
      */
     public function acknowledge(Message $message, $queue = null)
     {
-        $subscription = $this->getTopic($queue)->subscription($this->getSubscriberName());
+        $subscription = $this->getTopic($this->getQueue($queue))->subscription($this->getSubscriberName());
         $subscription->acknowledge($message);
     }
 
@@ -185,7 +190,7 @@ public function acknowledge(Message $message, $queue = null)
      */
     public function acknowledgeAndPublish(Message $message, $queue = null, $options = [], $delay = 0)
     {
-        $topic = $this->getTopic($queue);
+        $topic = $this->getTopic($this->getQueue($queue));
         $subscription = $topic->subscription($this->getSubscriberName());
 
         $subscription->acknowledge($message);
@@ -201,15 +206,37 @@ public function acknowledgeAndPublish(Message $message, $queue = null, $options
     }
 
     /**
-     * {@inheritdoc}
+     * Create a payload string from the given job and data.
+     *
+     * @param  string  $job
+     * @param  string  $queue
+     * @param  mixed   $data
+     * @return string
+     *
+     * @throws \Illuminate\Queue\InvalidPayloadException
      */
-    protected function createPayload($job, $queue = null, $data = '')
+    protected function createPayload($job, $queue, $data = '')
     {
-        $payload = parent::createPayload($job, $queue, $data);
+        $payload = parent::createPayload($job, $this->getQueue($queue), $data);
 
         return base64_encode($payload);
     }
 
+    /**
+     * Create a payload array from the given job and data.
+     *
+     * @param  mixed  $job
+     * @param  string  $queue
+     * @param  mixed  $data
+     * @return array
+     */
+    protected function createPayloadArray($job, $queue, $data = '')
+    {
+        return array_merge(parent::createPayloadArray($job, $this->getQueue($queue), $data), [
+            'id' => $this->getRandomId(),
+        ]);
+    }
+
     /**
      * Get the current topic.
      *
@@ -220,7 +247,7 @@ protected function createPayload($job, $queue = null, $data = '')
      */
     public function getTopic($queue, $create = false)
     {
-        $queue = $queue ?: $this->default;
+        $queue = $this->getQueue($queue);
         $topic = $this->pubsub->topic($queue);
 
         if (! $topic->exists() && $create) {
@@ -269,4 +296,25 @@ public function getPubSub()
     {
         return $this->pubsub;
     }
+
+    /**
+     * Get the queue or return the default.
+     *
+     * @param  string|null  $queue
+     * @return string
+     */
+    public function getQueue($queue)
+    {
+        return $queue ?: $this->default;
+    }
+
+    /**
+     * Get a random ID string.
+     *
+     * @return string
+     */
+    protected function getRandomId()
+    {
+        return Str::random(32);
+    }
 }