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

README.md

@@ -99,7 +99,14 @@ provider:
     environment:
         APP_ENV: production
         QUEUE_CONNECTION: sqs
-        SQS_QUEUE: ${construct:jobs.queueUrl}
+        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
+    iamRoleStatements:
+        # Allows our code to interact with SQS
+        -   Effect: Allow
+            Action: [sqs:SendMessage, sqs:DeleteMessage]
+            Resource: !GetAtt AlertQueue.Arn
 
 functions:
     ...

composer.json

@@ -16,11 +16,11 @@
         }
     },
     "require": {
-        "php": ">=7.3",
+        "php": "^7.3|^8.0",
         "ext-json": "*",
         "bref/bref": "^1.0",
-        "illuminate/queue": "^6.0|^7.0|^8.0",
-        "illuminate/support": "^6.0|^7.0|^8.0",
+        "illuminate/queue": "^6.0|^7.0|^8.0|^9.0",
+        "illuminate/support": "^6.0|^7.0|^8.0|^9.0",
         "aws/aws-sdk-php": "^3.134"
     },
     "require-dev": {

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 = 3,
+            $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);
+    }
+}