Merge pull request #38 from stackkit/bugfix/unlimited-max-attempts

Add support for 'Max retry duration' and allow for unlimited tasks

Author: marickvantuil

src/CloudTasksJob.php

@@ -2,6 +2,7 @@
 
 namespace Stackkit\LaravelGoogleCloudTasksQueue;
 
+use Google\Cloud\Tasks\V2\CloudTasksClient;
 use Illuminate\Container\Container;
 use Illuminate\Queue\Jobs\Job as LaravelJob;
 use Illuminate\Contracts\Queue\Job as JobContract;
@@ -11,11 +12,18 @@ class CloudTasksJob extends LaravelJob implements JobContract
     private $job;
     private $attempts;
     private $maxTries;
+    public $retryUntil = null;
 
-    public function __construct($job)
+    /**
+     * @var CloudTasksQueue
+     */
+    private $cloudTasksQueue;
+
+    public function __construct($job, CloudTasksQueue $cloudTasksQueue)
     {
         $this->job = $job;
         $this->container = Container::getInstance();
+        $this->cloudTasksQueue = $cloudTasksQueue;
     }
 
     public function getJobId()
@@ -41,7 +49,7 @@ public function setAttempts($attempts)
     public function setMaxTries($maxTries)
     {
         if ((int) $maxTries === -1) {
-            $maxTries = null;
+            $maxTries = 0;
         }
 
         $this->maxTries = $maxTries;
@@ -56,4 +64,27 @@ public function setQueue($queue)
     {
         $this->queue = $queue;
     }
+
+    public function setRetryUntil($retryUntil)
+    {
+        $this->retryUntil = $retryUntil;
+    }
+
+    public function retryUntil()
+    {
+        return $this->retryUntil;
+    }
+
+    // timeoutAt was renamed to retryUntil in 8.x but we still support this.
+    public function timeoutAt()
+    {
+        return $this->retryUntil;
+    }
+
+    public function delete()
+    {
+        parent::delete();
+
+        $this->cloudTasksQueue->delete($this);
+    }
 }

src/CloudTasksQueue.php

@@ -94,6 +94,20 @@ private function createHttpRequest()
         return app(HttpRequest::class);
     }
 
+    public function delete(CloudTasksJob $job)
+    {
+        $config = $this->config;
+
+        $taskName = $this->client->taskName(
+            $config['project'],
+            $config['location'],
+            $job->getQueue(),
+            request()->header('X-Cloudtasks-Taskname')
+        );
+
+        $this->client->deleteTask($taskName);
+    }
+
     /**
      * @return Task
      */

src/TaskHandler.php

@@ -2,7 +2,9 @@
 
 namespace Stackkit\LaravelGoogleCloudTasksQueue;
 
+use Google\Cloud\Tasks\V2\Attempt;
 use Google\Cloud\Tasks\V2\CloudTasksClient;
+use Google\Cloud\Tasks\V2\RetryConfig;
 use Illuminate\Http\Request;
 use Illuminate\Queue\Events\JobFailed;
 use Illuminate\Queue\Worker;
@@ -14,6 +16,16 @@ class TaskHandler
     private $publicKey;
     private $config;
 
+    /**
+     * @var CloudTasksQueue
+     */
+    private $queue;
+
+    /**
+     * @var RetryConfig
+     */
+    private $retryConfig = null;
+
     public function __construct(CloudTasksClient $client, Request $request, OpenIdVerificator $publicKey)
     {
         $this->client = $client;
@@ -31,6 +43,8 @@ public function handle($task = null)
 
         $this->loadQueueConnectionConfiguration($task);
 
+        $this->setQueue();
+
         $this->authorizeRequest();
 
         $this->listenForEvents();
@@ -48,6 +62,11 @@ private function loadQueueConnectionConfiguration($task)
         );
     }
 
+    private function setQueue()
+    {
+        $this->queue = new CloudTasksQueue($this->config, $this->client);
+    }
+
     /**
      * @throws CloudTasksException
      */
@@ -122,29 +141,63 @@ private function listenForEvents()
      */
     private function handleTask($task)
     {
-        $job = new CloudTasksJob($task);
+        $job = new CloudTasksJob($task, $this->queue);
+
+        $this->loadQueueRetryConfig();
 
         $job->setAttempts(request()->header('X-CloudTasks-TaskRetryCount') + 1);
         $job->setQueue(request()->header('X-Cloudtasks-Queuename'));
-        $job->setMaxTries($this->getQueueMaxTries($job));
+        $job->setMaxTries($this->retryConfig->getMaxAttempts());
+
+        // If the job is being attempted again we also check if a
+        // max retry duration has been set. If that duration
+        // has passed, it should stop trying altogether.
+        if ($job->attempts() > 1) {
+            $job->setRetryUntil($this->getRetryUntilTimestamp($job));
+        }
 
         $worker = $this->getQueueWorker();
 
         $worker->process($this->config['connection'], $job, new WorkerOptions());
     }
 
-    private function getQueueMaxTries(CloudTasksJob $job)
+    private function loadQueueRetryConfig()
     {
         $queueName = $this->client->queueName(
             $this->config['project'],
             $this->config['location'],
-            $job->getQueue()
+            request()->header('X-Cloudtasks-Queuename')
         );
 
-        return $this->client
-            ->getQueue($queueName)
-            ->getRetryConfig()
-            ->getMaxAttempts();
+        $this->retryConfig = $this->client->getQueue($queueName)->getRetryConfig();
+    }
+
+    private function getRetryUntilTimestamp(CloudTasksJob $job)
+    {
+        $task = $this->client->getTask(
+            $this->client->taskName(
+                $this->config['project'],
+                $this->config['location'],
+                $job->getQueue(),
+                request()->header('X-Cloudtasks-Taskname')
+            )
+        );
+
+        $attempt = $task->getFirstAttempt();
+
+        if (!$attempt instanceof Attempt) {
+            return null;
+        }
+
+        if (! $this->retryConfig->hasMaxRetryDuration()) {
+            return null;
+        }
+
+        $maxDurationInSeconds = $this->retryConfig->getMaxRetryDuration()->getSeconds();
+
+        $firstAttemptTimestamp = $attempt->getDispatchTime()->toDateTime()->getTimestamp();
+
+        return $firstAttemptTimestamp + $maxDurationInSeconds;
     }
 
     /**