Add support to set queue options, either globally using queue.php or by implementing an interface

CHANGELOG.md

@@ -7,6 +7,7 @@ The format is based on [Keep a Changelog][keepachangelog] and this project adher
 ### Added
 
 - gRPC client support
+- Added support to set queue options globally or per job [#158]
 
 ## Unreleased
 

src/Queue/Contract/HasQueueOptions.php

@@ -0,0 +1,8 @@
+<?php
+
+namespace Spiral\RoadRunnerLaravel\Queue\Contract;
+
+interface HasQueueOptions
+{
+    function queueOptions(): array;
+}

src/Queue/RoadRunnerConnector.php

@@ -26,6 +26,7 @@ public function connect(array $config): Queue
             new Jobs($rpc),
             $rpc,
             $config['queue'],
+            $config['options'] ?? [],
         );
     }
 }

src/Queue/RoadRunnerQueue.php

@@ -12,14 +12,20 @@
 use RoadRunner\Jobs\DTO\V1\Stats;
 use Spiral\Goridge\RPC\RPCInterface;
 use Spiral\RoadRunner\Jobs\Jobs;
+use Spiral\RoadRunner\Jobs\KafkaOptions;
+use Spiral\RoadRunner\Jobs\Options;
+use Spiral\RoadRunner\Jobs\OptionsInterface;
+use Spiral\RoadRunner\Jobs\Queue\Driver;
 use Spiral\RoadRunner\Jobs\QueueInterface;
+use Spiral\RoadRunnerLaravel\Queue\Contract\HasQueueOptions;
 
 final class RoadRunnerQueue extends Queue implements QueueContract
 {
     public function __construct(
         private readonly Jobs $jobs,
         private readonly RPCInterface $rpc,
         private readonly string $default = 'default',
+        private readonly array $defaultOptions = [],
     ) {}
 
     public function push($job, $data = '', $queue = null): string
@@ -29,13 +35,13 @@ public function push($job, $data = '', $queue = null): string
             $this->createPayload($job, $queue, $data),
             $queue,
             null,
-            fn($payload, $queue) => $this->pushRaw($payload, $queue),
+            fn($payload, $queue) => $this->pushRaw($payload, $queue, $this->getJobOverrideOptions($job)),
         );
     }
 
     public function pushRaw($payload, $queue = null, array $options = []): string
     {
-        $queue = $this->getQueue($queue);
+        $queue = $this->getQueue($queue, $options);
 
         $task = $queue->dispatch(
             $queue
@@ -52,7 +58,7 @@ public function later($delay, $job, $data = '', $queue = null): string
             $this->createPayload($job, $queue, $data),
             $queue,
             $delay,
-            fn($payload, $queue) => $this->laterRaw($delay, $payload, $queue),
+            fn($payload, $queue) => $this->laterRaw($delay, $payload, $queue, $this->getJobOverrideOptions($job)),
         );
     }
 
@@ -88,8 +94,9 @@ private function laterRaw(
         \DateTimeInterface|\DateInterval|int $delay,
         array $payload,
         ?string $queue = null,
+        array $options = []
     ): string {
-        $queue = $this->getQueue($queue);
+        $queue = $this->getQueue($queue, $options);
 
         $task = $queue->dispatch(
             $queue
@@ -101,9 +108,9 @@ private function laterRaw(
         return $task->getId();
     }
 
-    private function getQueue(?string $queue = null): QueueInterface
+    private function getQueue(?string $queue = null, array $options = []): QueueInterface
     {
-        $queue = $this->jobs->connect($queue ?? $this->default);
+        $queue = $this->jobs->connect($queue ?? $this->default, $this->getQueueOptions($options));
 
         if (!$this->getStats($queue->getName())->getReady()) {
             $queue->resume();
@@ -127,4 +134,29 @@ private function getStats(?string $queue = null): Stat
 
         return new Stat();
     }
+
+    private function getJobOverrideOptions(string|object $job): array
+    {
+        if ($job instanceof HasQueueOptions) {
+            return $job->queueOptions();
+        }
+
+        return [];
+    }
+
+    private function getQueueOptions(array $overrides = []): OptionsInterface
+    {
+        $config = array_merge($this->defaultOptions, $overrides);
+        $options = new Options(
+            $config['delay'] ?? 0,
+            $config['priority'] ?? 0,
+            $config['auto_ack'] ?? false
+        );
+
+        return match ($config['driver'] ?? null) {
+            Driver::Kafka => KafkaOptions::from($options)
+                ->withTopic($config['topic'] ?? 'default'),
+            default => $options,
+        };
+    }
 }