feat: allow queue listeners to be unique

Author: calebdw

src/Illuminate/Bus/UniqueLock.php

@@ -36,8 +36,8 @@ public function acquire($job)
             : ($job->uniqueFor ?? 0);
 
         $cache = method_exists($job, 'uniqueVia')
-            ? $job->uniqueVia()
-            : $this->cache;
+            ? ($job->uniqueVia() ?? $this->cache)
+            : ($job->uniqueVia ?? $this->cache);
 
         return (bool) $cache->lock($this->getKey($job), $uniqueFor)->get();
     }
@@ -51,8 +51,8 @@ public function acquire($job)
     public function release($job)
     {
         $cache = method_exists($job, 'uniqueVia')
-            ? $job->uniqueVia()
-            : $this->cache;
+            ? ($job->uniqueVia() ?? $this->cache)
+            : ($job->uniqueVia ?? $this->cache);
 
         $cache->lock($this->getKey($job))->forceRelease();
     }

src/Illuminate/Events/CallQueuedListener.php

@@ -4,6 +4,7 @@
 
 use Illuminate\Bus\Queueable;
 use Illuminate\Container\Container;
+use Illuminate\Contracts\Cache\Repository as Cache;
 use Illuminate\Contracts\Queue\Job;
 use Illuminate\Contracts\Queue\ShouldQueue;
 use Illuminate\Queue\InteractsWithQueue;
@@ -82,6 +83,31 @@ class CallQueuedListener implements ShouldQueue
      */
     public $shouldBeEncrypted = false;
 
+    /**
+     * Indicates if the listener should be unique.
+     */
+    public bool $shouldBeUnique = false;
+
+    /**
+     * Indicates if the listener should be unique until processing begins.
+     */
+    public bool $shouldBeUniqueUntilProcessing = false;
+
+    /**
+     * The unique ID of the listener.
+     */
+    public mixed $uniqueId = null;
+
+    /**
+     * The number of seconds the unique lock should be maintained.
+     */
+    public ?int $uniqueFor = null;
+
+    /**
+     * The cache store used to manage unique locks.
+     */
+    public ?Cache $uniqueVia = null;
+
     /**
      * Create a new job instance.
      *
@@ -172,6 +198,46 @@ public function displayName()
         return $this->class;
     }
 
+    /**
+     * Determine if the listener should be unique.
+     */
+    public function shouldBeUnique(): bool
+    {
+        return $this->shouldBeUnique;
+    }
+
+    /**
+     * Determine if the listener should be unique until processing begins.
+     */
+    public function shouldBeUniqueUntilProcessing(): bool
+    {
+        return $this->shouldBeUniqueUntilProcessing;
+    }
+
+    /**
+     * Get the unique ID for the listener.
+     */
+    public function uniqueId(): mixed
+    {
+        return $this->uniqueId;
+    }
+
+    /**
+     * Get the number of seconds the unique lock should be maintained.
+     */
+    public function uniqueFor(): ?int
+    {
+        return $this->uniqueFor;
+    }
+
+    /**
+     * Get the cache store used to manage unique locks.
+     */
+    public function uniqueVia(): ?Cache
+    {
+        return $this->uniqueVia;
+    }
+
     /**
      * Prepare the instance for cloning.
      *

src/Illuminate/Events/Dispatcher.php

@@ -4,14 +4,18 @@
 
 use Closure;
 use Exception;
+use Illuminate\Bus\UniqueLock;
 use Illuminate\Container\Container;
 use Illuminate\Contracts\Broadcasting\Factory as BroadcastFactory;
 use Illuminate\Contracts\Broadcasting\ShouldBroadcast;
+use Illuminate\Contracts\Cache\Repository as Cache;
 use Illuminate\Contracts\Container\Container as ContainerContract;
 use Illuminate\Contracts\Events\Dispatcher as DispatcherContract;
 use Illuminate\Contracts\Events\ShouldDispatchAfterCommit;
 use Illuminate\Contracts\Events\ShouldHandleEventsAfterCommit;
 use Illuminate\Contracts\Queue\ShouldBeEncrypted;
+use Illuminate\Contracts\Queue\ShouldBeUnique;
+use Illuminate\Contracts\Queue\ShouldBeUniqueUntilProcessing;
 use Illuminate\Contracts\Queue\ShouldQueue;
 use Illuminate\Contracts\Queue\ShouldQueueAfterCommit;
 use Illuminate\Support\Arr;
@@ -649,6 +653,11 @@ protected function queueHandler($class, $method, $arguments)
     {
         [$listener, $job] = $this->createListenerAndJob($class, $method, $arguments);
 
+        if ($job->shouldBeUnique &&
+            ! (new UniqueLock($this->container->make(Cache::class)))->acquire($job)) {
+            return;
+        }
+
         $connection = $this->resolveQueue()->connection(method_exists($listener, 'viaConnection')
             ? (isset($arguments[0]) ? $listener->viaConnection($arguments[0]) : $listener->viaConnection())
             : $listener->connection ?? null);
@@ -720,6 +729,21 @@ protected function propagateListenerOptions($listener, $job)
                 method_exists($listener, 'middleware') ? $listener->middleware(...$data) : [],
                 $listener->middleware ?? []
             ));
+
+            $job->shouldBeUnique = $listener instanceof ShouldBeUnique;
+            $job->shouldBeUniqueUntilProcessing = $listener instanceof ShouldBeUniqueUntilProcessing;
+
+            if ($job->shouldBeUnique) {
+                $job->uniqueId = method_exists($listener, 'uniqueId')
+                    ? $listener->uniqueId(...$data)
+                    : ($listener->uniqueId ?? null);
+                $job->uniqueFor = method_exists($listener, 'uniqueFor')
+                    ? $listener->uniqueFor(...$data)
+                    : ($listener->uniqueFor ?? 0);
+                $job->uniqueVia = method_exists($listener, 'uniqueVia')
+                    ? $listener->uniqueVia(...$data)
+                    : null;
+            }
         });
     }
 

src/Illuminate/Queue/CallQueuedHandler.php

@@ -14,6 +14,7 @@
 use Illuminate\Contracts\Queue\ShouldBeUnique;
 use Illuminate\Contracts\Queue\ShouldBeUniqueUntilProcessing;
 use Illuminate\Database\Eloquent\ModelNotFoundException;
+use Illuminate\Events\CallQueuedListener;
 use Illuminate\Log\Context\Repository as ContextRepository;
 use Illuminate\Pipeline\Pipeline;
 use Illuminate\Queue\Attributes\DeleteWhenMissingModels;
@@ -67,7 +68,7 @@ public function call(Job $job, array $data)
 
         $this->dispatchThroughMiddleware($job, $command);
 
-        if (! $job->isReleased() && ! $command instanceof ShouldBeUniqueUntilProcessing) {
+        if (! $job->isReleased() && ! $this->commandShouldBeUniqueUntilProcessing($command)) {
             $this->ensureUniqueJobLockIsReleased($command);
         }
 
@@ -120,12 +121,12 @@ protected function dispatchThroughMiddleware(Job $job, $command)
         return (new Pipeline($this->container))->send($command)
             ->through(array_merge(method_exists($command, 'middleware') ? $command->middleware() : [], $command->middleware ?? []))
             ->finally(function ($command) use (&$lockReleased) {
-                if (! $lockReleased && $command instanceof ShouldBeUniqueUntilProcessing && ! $command->job->isReleased()) {
+                if (! $lockReleased && $this->commandShouldBeUniqueUntilProcessing($command) && ! $command->job->isReleased()) {
                     $this->ensureUniqueJobLockIsReleased($command);
                 }
             })
             ->then(function ($command) use ($job, &$lockReleased) {
-                if ($command instanceof ShouldBeUniqueUntilProcessing) {
+                if ($this->commandShouldBeUniqueUntilProcessing($command)) {
                     $this->ensureUniqueJobLockIsReleased($command);
 
                     $lockReleased = true;
@@ -212,11 +213,29 @@ protected function ensureSuccessfulBatchJobIsRecorded($command)
      */
     protected function ensureUniqueJobLockIsReleased($command)
     {
-        if ($command instanceof ShouldBeUnique) {
+        if ($this->commandShouldBeUnique($command)) {
             (new UniqueLock($this->container->make(Cache::class)))->release($command);
         }
     }
 
+    /**
+     * Determine if the given command should be unique.
+     */
+    protected function commandShouldBeUnique(mixed $command): bool
+    {
+        return $command instanceof ShouldBeUnique ||
+            ($command instanceof CallQueuedListener && $command->shouldBeUnique());
+    }
+
+    /**
+     * Determine if the given command should be unique until processing begins.
+     */
+    protected function commandShouldBeUniqueUntilProcessing(mixed $command): bool
+    {
+        return $command instanceof ShouldBeUniqueUntilProcessing ||
+            ($command instanceof CallQueuedListener && $command->shouldBeUniqueUntilProcessing());
+    }
+
     /**
      * Handle a model not found exception.
      *
@@ -294,7 +313,7 @@ public function failed(array $data, $e, string $uuid, ?Job $job = null)
             $command = $this->setJobInstanceIfNecessary($job, $command);
         }
 
-        if (! $command instanceof ShouldBeUniqueUntilProcessing) {
+        if (! $this->commandShouldBeUniqueUntilProcessing($command)) {
             $this->ensureUniqueJobLockIsReleased($command);
         }