Store job attempt in job payload & use Laravel's built-in queue worker

LICENSE

@@ -19,3 +19,27 @@ AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 THE SOFTWARE.
+
+vapor-core is distributed under the following license:
+
+The MIT License (MIT)
+
+Copyright (c) Taylor Otwell
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.

README.md

@@ -86,6 +86,7 @@ provider:
     ...
     environment:
         APP_ENV: production
+        QUEUE_CONNECTION: sqs
         SQS_QUEUE: !Ref AlertQueue
         # If you create the queue manually, the `SQS_QUEUE` variable can be defined like this:
         # SQS_QUEUE: https://sqs.us-east-1.amazonaws.com/your-account-id/my-queue

src/Queue/LaravelSqsHandler.php

@@ -7,35 +7,28 @@
 use Bref\Event\Sqs\SqsEvent;
 use Bref\Event\Sqs\SqsHandler;
 use Illuminate\Container\Container;
-use Illuminate\Contracts\Debug\ExceptionHandler;
-use Illuminate\Contracts\Events\Dispatcher;
-use Illuminate\Queue\Events\JobExceptionOccurred;
-use Illuminate\Queue\Events\JobProcessed;
-use Illuminate\Queue\Events\JobProcessing;
-use Illuminate\Queue\Jobs\SqsJob;
 use Illuminate\Queue\QueueManager;
 use Illuminate\Queue\SqsQueue;
-use Throwable;
+use Illuminate\Queue\WorkerOptions;
 
 /**
  * SQS handler for AWS Lambda that integrates with Laravel Queue.
  */
 class LaravelSqsHandler extends SqsHandler
 {
-    /** @var Container */
+    /** @var Container $container */
     private $container;
+
     /** @var SqsClient */
     private $sqs;
-    /** @var Dispatcher */
-    private $events;
-    /** @var ExceptionHandler */
-    private $exceptions;
+
     /** @var string */
     private $connectionName;
+
     /** @var string */
     private $queue;
 
-    public function __construct(Container $container, Dispatcher $events, ExceptionHandler $exceptions, string $connection, string $queue)
+    public function __construct(Container $container, string $connection, string $queue)
     {
         $this->container = $container;
         $queueManager = $container->get('queue');
@@ -45,95 +38,71 @@ public function __construct(Container $container, Dispatcher $events, ExceptionH
             throw new \RuntimeException("The '$connection' connection is not a SQS connection in the Laravel config");
         }
         $this->sqs = $queueConnector->getSqs();
-        $this->events = $events;
-        $this->exceptions = $exceptions;
         $this->connectionName = $connection;
         $this->queue = $queue;
     }
 
     public function handleSqs(SqsEvent $event, Context $context): void
     {
+        $worker = $this->container->makeWith(Worker::class, [
+            'isDownForMaintenance' => function () {
+                return false;
+            },
+        ]);
+
+        \assert($worker instanceof Worker);
+
         foreach ($event->getRecords() as $sqsRecord) {
-            $recordData = $sqsRecord->toArray();
-            $jobData = [
-                'MessageId' => $recordData['messageId'],
-                'ReceiptHandle' => $recordData['receiptHandle'],
-                'Attributes' => $recordData['attributes'],
-                'Body' => $recordData['body'],
-            ];
-
-            $job = new SqsJob(
-                $this->container,
-                $this->sqs,
-                $jobData,
+            $message = $this->normalizeMessage($sqsRecord->toArray());
+
+            $worker->runSqsJob(
+                $this->buildJob($message),
                 $this->connectionName,
-                $this->queue,
+                $this->getWorkerOptions()
             );
-
-            $this->process($this->connectionName, $job);
         }
     }
 
-    /**
-     * @see \Illuminate\Queue\Worker::process()
-     */
-    private function process(string $connectionName, SqsJob $job): void
+    protected function normalizeMessage(array $message): array
     {
-        try {
-            // First we will raise the before job event and determine if the job has already ran
-            // over its maximum attempt limits, which could primarily happen when this job is
-            // continually timing out and not actually throwing any exceptions from itself.
-            $this->raiseBeforeJobEvent($connectionName, $job);
-
-            // Here we will fire off the job and let it process. We will catch any exceptions so
-            // they can be reported to the developers logs, etc. Once the job is finished the
-            // proper events will be fired to let any listeners know this job has finished.
-            $job->fire();
-
-            $this->raiseAfterJobEvent($connectionName, $job);
-        } catch (Throwable $e) {
-            // Fire off an exception event
-            $this->raiseExceptionOccurredJobEvent($connectionName, $job, $e);
-
-            // Report exception to defined log channel
-            $this->exceptions->report($e);
-
-            // Rethrow the exception to let SQS handle it
-            throw $e;
-        }
+        return [
+            'MessageId' => $message['messageId'],
+            'ReceiptHandle' => $message['receiptHandle'],
+            'Body' => $message['body'],
+            'Attributes' => $message['attributes'],
+            'MessageAttributes' => $message['messageAttributes'],
+        ];
     }
 
-    /**
-     * @see \Illuminate\Queue\Worker::raiseBeforeJobEvent()
-     */
-    private function raiseBeforeJobEvent(string $connectionName, SqsJob $job): void
+    protected function buildJob(array $message): SqsJob
     {
-        $this->events->dispatch(new JobProcessing(
-            $connectionName,
-            $job
-        ));
+        return new SqsJob(
+            $this->container,
+            $this->sqs,
+            $message,
+            $this->connectionName,
+            $this->queue,
+        );
     }
 
-    /**
-     * @see \Illuminate\Queue\Worker::raiseAfterJobEvent()
-     */
-    private function raiseAfterJobEvent(string $connectionName, SqsJob $job): void
+    protected function getWorkerOptions(): WorkerOptions
     {
-        $this->events->dispatch(new JobProcessed(
-            $connectionName,
-            $job
-        ));
-    }
+        $options = [
+            $backoff = 0,
+            $memory = 512,
+            $timeout = 0,
+            $sleep = 0,
+            $maxTries = 0,
+            $force = false,
+            $stopWhenEmpty = false,
+            $maxJobs = 0,
+            $maxTime = 0,
+        ];
 
-    /**
-     * @see \Illuminate\Queue\Worker::raiseExceptionOccurredJobEvent()
-     */
-    private function raiseExceptionOccurredJobEvent(string $connectionName, SqsJob $job, Throwable $e): void
-    {
-        $this->events->dispatch(new JobExceptionOccurred(
-            $connectionName,
-            $job,
-            $e
-        ));
+        if (property_exists(WorkerOptions::class, 'name')) {
+            $options = array_merge(['default'], $options);
+        }
+
+        return new WorkerOptions(...$options);
     }
 }

src/Queue/SqsJob.php

@@ -0,0 +1,43 @@
+<?php declare(strict_types=1);
+
+namespace Bref\LaravelBridge\Queue;
+
+use Illuminate\Queue\Jobs\SqsJob as LaravelSqsJob;
+
+/**
+ * @author Taylor Otwell <[email protected]>
+ * @copyright Copyright (c) 2019, Taylor Otwell
+ */
+class SqsJob extends LaravelSqsJob
+{
+    /**
+     * {@inheritDoc}
+     */
+    public function release($delay = 0)
+    {
+        $this->released = true;
+
+        $payload = $this->payload();
+
+        $payload['attempts'] = ($payload['attempts'] ?? 0) + 1;
+
+        $this->sqs->deleteMessage([
+            'QueueUrl' => $this->queue,
+            'ReceiptHandle' => $this->job['ReceiptHandle'],
+        ]);
+
+        $this->sqs->sendMessage([
+            'QueueUrl' => $this->queue,
+            'MessageBody' => json_encode($payload),
+            'DelaySeconds' => $this->secondsUntil($delay),
+        ]);
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function attempts()
+    {
+        return ($this->payload()['attempts'] ?? 0) + 1;
+    }
+}

src/Queue/Worker.php

@@ -0,0 +1,15 @@
+<?php declare(strict_types=1);
+
+namespace Bref\LaravelBridge\Queue;
+
+use Illuminate\Contracts\Queue\Job;
+use Illuminate\Queue\Worker as LaravelWorker;
+use Illuminate\Queue\WorkerOptions;
+
+class Worker extends LaravelWorker
+{
+    public function runSqsJob(Job $job, string $connectionName, WorkerOptions $options): void
+    {
+        $this->runJob($job, $connectionName, $options);
+    }
+}