Merge pull request #26 from r2luna/issue-18

issue 18

Author: r2luna

README.md

@@ -150,6 +150,171 @@ class SendWelcomeNotifications extends Task implements ShouldQueue
 }
 ```
 
+#### Setting a Specific Queue with `#[OnQueue]`
+
+> [!WARNING]
+> Do **not** declare `public string $queue = 'my-queue'` on a Task — this causes a PHP fatal error because Laravel's `Queueable` trait already declares `$queue` without a type hint.
+
+Use the `#[OnQueue('queue-name')]` attribute to assign a specific queue to a Task or Process:
+
+```php
+use Brain\Attributes\OnQueue;
+use Brain\Task;
+use Illuminate\Contracts\Queue\ShouldQueue;
+
+#[OnQueue('emails')]
+class SendWelcomeNotifications extends Task implements ShouldQueue
+{
+    public function handle(): self
+    {
+        // This task will run on the "emails" queue
+        return $this;
+    }
+}
+```
+
+When applied to a **Process**, the attribute does two things:
+
+1. The Process itself is dispatched to that queue (if it implements `ShouldQueue`)
+2. All queued child tasks **inherit** the Process queue — unless the task defines its own `#[OnQueue]`
+
+```php
+use Brain\Attributes\OnQueue;
+use Brain\Process;
+use Illuminate\Contracts\Queue\ShouldQueue;
+
+#[OnQueue('strava')]
+class SyncActivitiesProcess extends Process implements ShouldQueue
+{
+    protected array $tasks = [
+        FetchActivities::class,      // ShouldQueue → runs on "strava" (inherited)
+        SaveActivities::class,       // sync task   → unaffected
+        NotifyUser::class,           // has #[OnQueue('emails')] → runs on "emails" (own queue wins)
+    ];
+}
+```
+
+#### Queue Execution Flows
+
+Below are the three most common queue configurations.
+
+**1. Process + all Tasks on the same queue**
+
+The process and every queued task run on `strava`:
+
+```mermaid
+flowchart LR
+    D((Dispatch)) --> Q[strava queue]
+
+    subgraph Q[strava queue]
+        direction LR
+        P[SyncActivitiesProcess] --> T1[FetchActivities]
+        T1 --> T2[TransformData]
+        T2 --> T3[SaveActivities]
+    end
+
+    style Q fill:#1a1a2e,stroke:#e94560,color:#eee
+    style P fill:#0f3460,stroke:#e94560,color:#eee
+    style T1 fill:#16213e,stroke:#0f3460,color:#eee
+    style T2 fill:#16213e,stroke:#0f3460,color:#eee
+    style T3 fill:#16213e,stroke:#0f3460,color:#eee
+```
+
+```php
+#[OnQueue('strava')]
+class SyncActivitiesProcess extends Process implements ShouldQueue
+{
+    protected array $tasks = [
+        FetchActivities::class,    // implements ShouldQueue
+        TransformData::class,      // implements ShouldQueue
+        SaveActivities::class,     // implements ShouldQueue
+    ];
+}
+```
+
+**2. Process on a queue, Tasks run synchronously inside it**
+
+The process is queued, but internally its tasks run one after another in the same job:
+
+```mermaid
+flowchart LR
+    D((Dispatch)) --> Q
+
+    subgraph Q[strava queue]
+        subgraph P[SyncActivitiesProcess]
+            direction LR
+            T1[FetchActivities] --> T2[TransformData] --> T3[SaveActivities]
+        end
+    end
+
+    style Q fill:#1a1a2e,stroke:#e94560,color:#eee
+    style P fill:#0f3460,stroke:#e94560,color:#eee
+    style T1 fill:#16213e,stroke:#0f3460,color:#eee
+    style T2 fill:#16213e,stroke:#0f3460,color:#eee
+    style T3 fill:#16213e,stroke:#0f3460,color:#eee
+```
+
+```php
+#[OnQueue('strava')]
+class SyncActivitiesProcess extends Process implements ShouldQueue
+{
+    protected array $tasks = [
+        FetchActivities::class,    // sync (no ShouldQueue)
+        TransformData::class,      // sync
+        SaveActivities::class,     // sync
+    ];
+}
+```
+
+**3. Process on a queue, mixed sync and queued Tasks**
+
+The process is queued. Some tasks run synchronously inside the process job, others are dispatched to their own queue jobs:
+
+```mermaid
+flowchart LR
+    D((Dispatch)) --> Q
+
+    subgraph Q[strava queue]
+        subgraph P[SyncActivitiesProcess]
+            direction LR
+            T1[FetchActivities\nsync] --> T2[TransformData\nsync]
+        end
+        T2 --> T3[SaveActivities\nqueued]
+        T3 --> T4[NotifyUser\nqueued]
+    end
+
+    T4 -.-> E[emails queue]
+
+    subgraph E[emails queue]
+        T4real[NotifyUser]
+    end
+
+    style Q fill:#1a1a2e,stroke:#e94560,color:#eee
+    style E fill:#1a1a2e,stroke:#53a653,color:#eee
+    style P fill:#0f3460,stroke:#e94560,color:#eee
+    style T1 fill:#16213e,stroke:#0f3460,color:#eee
+    style T2 fill:#16213e,stroke:#0f3460,color:#eee
+    style T3 fill:#16213e,stroke:#0f3460,color:#eee
+    style T4 fill:#16213e,stroke:#0f3460,color:#eee
+    style T4real fill:#16213e,stroke:#53a653,color:#eee
+```
+
+```php
+#[OnQueue('strava')]
+class SyncActivitiesProcess extends Process implements ShouldQueue
+{
+    protected array $tasks = [
+        FetchActivities::class,    // sync — runs inside the process job
+        TransformData::class,      // sync — runs inside the process job
+        SaveActivities::class,     // implements ShouldQueue → dispatched to "strava"
+        NotifyUser::class,         // implements ShouldQueue + #[OnQueue('emails')] → dispatched to "emails"
+    ];
+}
+```
+
+> [!TIP]
+> A task's own `#[OnQueue]` always takes precedence over the Process-level queue.
+
 #### Conditional Run Tasks in a Process
 
 You can use a protected function `runIf()` to conditionally run a task.

src/Attributes/OnQueue.php

@@ -0,0 +1,13 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Brain\Attributes;
+
+use Attribute;
+
+#[Attribute(Attribute::TARGET_CLASS)]
+class OnQueue
+{
+    public function __construct(public string $queue) {}
+}

src/Console/BrainMap.php

@@ -4,6 +4,7 @@
 
 namespace Brain\Console;
 
+use Brain\Attributes\OnQueue;
 use Brain\Process;
 use Brain\Task;
 use Exception;
@@ -159,9 +160,13 @@ private function loadProcessesFor(string $domainPath): array
                 $chainValue = $chainProperty->getValue(new $reflection->name([]));
                 $value = $value->getPathname();
 
+                $onQueueAttr = $reflection->getAttributes(OnQueue::class);
+                $queueName = $onQueueAttr !== [] ? $onQueueAttr[0]->newInstance()->queue : null;
+
                 return [
                     'name' => basename($value, '.php'),
                     'chain' => $chainValue,
+                    'onQueue' => $queueName,
                     'tasks' => $this->getProcessesTasks($reflection),
                 ];
             })
@@ -211,10 +216,14 @@ private function getTask(SplFileInfo|string $task): array
         $reflection = $this->getReflectionClass($task);
         $isProcess = $reflection->isSubclassOf(Process::class);
 
+        $onQueueAttr = $reflection->getAttributes(OnQueue::class);
+        $queueName = $onQueueAttr !== [] ? $onQueueAttr[0]->newInstance()->queue : null;
+
         $data = [
             'name' => $reflection->getShortName(),
             'fullName' => $reflection->name,
             'queue' => $reflection->implementsInterface(ShouldQueue::class),
+            'onQueue' => $queueName,
             'type' => $isProcess ? 'process' : 'task',
             'properties' => $this->getPropertiesFor($reflection),
         ];

src/Process.php

@@ -4,13 +4,15 @@
 
 namespace Brain;
 
+use Brain\Attributes\OnQueue;
 use Brain\Processes\Events\Error;
 use Brain\Processes\Events\Processed;
 use Brain\Processes\Events\Processing;
 use Brain\Tasks\Events\Cancelled;
 use Brain\Tasks\Events\Error as TasksError;
 use Brain\Tasks\Events\Skipped;
 use Exception;
+use Illuminate\Bus\Queueable;
 use Illuminate\Contracts\Queue\ShouldQueue;
 use Illuminate\Foundation\Bus\Dispatchable;
 use Illuminate\Support\Facades\Bus;
@@ -29,6 +31,7 @@
 class Process
 {
     use Dispatchable;
+    use Queueable;
 
     /**
      * When running the process we will assign am uuid
@@ -66,6 +69,13 @@ public function __construct(
         $this->name = (new ReflectionClass($this))->getName();
 
         Context::add('process', [$this->name, $this->uuid]);
+
+        $onQueue = (new ReflectionClass(static::class))
+            ->getAttributes(OnQueue::class);
+
+        if ($onQueue !== []) {
+            $this->onQueue($onQueue[0]->newInstance()->queue);
+        }
     }
 
     /**
@@ -140,8 +150,13 @@ public function handle(): object|array|null
      */
     private function runInChain(?object $payload = null): ?object
     {
-        Bus::chain($this->getChainedTasks())
-            ->dispatch();
+        $chain = Bus::chain($this->getChainedTasks());
+
+        if (($queue = $this->resolveQueue()) !== null && ($queue = $this->resolveQueue()) !== '' && ($queue = $this->resolveQueue()) !== '0') {
+            $chain->onQueue($queue);
+        }
+
+        $chain->dispatch();
 
         return $payload;
     }
@@ -194,7 +209,14 @@ private function run(array|object|null $payload): ?object
                 }
 
                 if ($reflectionClass->implementsInterface(ShouldQueue::class)) {
-                    $task::dispatch($payload);
+                    $processQueue = $this->resolveQueue();
+                    $instance = new $task($payload);
+
+                    if ($instance->queue === null && $processQueue !== null) {
+                        $instance->onQueue($processQueue);
+                    }
+
+                    dispatch($instance);
 
                     continue;
                 }
@@ -242,6 +264,21 @@ private function run(array|object|null $payload): ?object
         return $payload;
     }
 
+    /**
+     * Resolve the queue name from the #[OnQueue] attribute on this Process class.
+     */
+    private function resolveQueue(): ?string
+    {
+        $attributes = (new ReflectionClass(static::class))
+            ->getAttributes(OnQueue::class);
+
+        if ($attributes === []) {
+            return null;
+        }
+
+        return $attributes[0]->newInstance()->queue;
+    }
+
     /**
      * Fire Event for the Listeners save all the info
      * in the database, and we track what is happening to

src/Task.php

@@ -4,6 +4,7 @@
 
 namespace Brain;
 
+use Brain\Attributes\OnQueue;
 use Brain\Exceptions\InvalidPayload;
 use Brain\Tasks\Events\Processed;
 use Brain\Tasks\Events\Processing;
@@ -57,6 +58,13 @@ public function __construct(
         if ($runIn = $this->runIn()) {
             $this->delay($runIn);
         }
+
+        $onQueue = (new ReflectionClass(static::class))
+            ->getAttributes(OnQueue::class);
+
+        if ($onQueue !== []) {
+            $this->onQueue($onQueue[0]->newInstance()->queue);
+        }
     }
 
     /**