channels

Author: asika32764

packages/queue/src/AbstractRunner.php

@@ -87,6 +87,8 @@ public function loop(string|array $channel): void
     {
         gc_enable();
 
+        $isAllChannels = in_array('*', (array) $channel, true);
+
         // Last Restart
         $this->lastRestart = (int) new DateTimeImmutable('now')->format('U');
 
@@ -113,8 +115,18 @@ public function loop(string|array $channel): void
             $this->registerSignals();
 
             if (($this->options->force ?? null) || $this->canLoop()) {
+                $channels = $channel;
+
+                if ($isAllChannels) {
+                    $channels = iterator_to_array($this->queue->getChannels());
+                }
+
+                if ($this->options->shuffleChannels && is_array($channels)) {
+                    shuffle($channels);
+                }
+
                 try {
-                    $this->next($channel);
+                    $this->next($channels);
                 } catch (UnrecoverableException $e) {
                     $this->stop('[STOP] Unrecoverable error: ' . $e->getMessage(), 1, true);
                 } catch (Exception $exception) {

packages/queue/src/Queue.php

@@ -7,6 +7,7 @@
 use InvalidArgumentException;
 use JsonException;
 use Windwalker\DI\Definition\DefinitionInterface;
+use Windwalker\Queue\Driver\ChannelAwareDriverInterface;
 use Windwalker\Queue\Driver\QueueDriverInterface;
 use Windwalker\Queue\Job\ClosureJob;
 use Windwalker\Utilities\Classes\ObjectBuilderAwareTrait;
@@ -157,6 +158,18 @@ public function defer(mixed $message, int $delay = 0): void
         $this->driver->defer($message);
     }
 
+    /**
+     * @return  iterable<string>
+     */
+    public function getChannels(): iterable
+    {
+        if (!$this->driver instanceof ChannelAwareDriverInterface) {
+            throw new \DomainException('Driver does not support get channels.');
+        }
+
+        return $this->driver->getChannels();
+    }
+
     /**
      * getMessage
      *

packages/queue/src/RunnerOptions.php

@@ -16,6 +16,7 @@ public function __construct(
         public int $timeout = 60,
         public int $maxRuns = 0,
         public int $lifetime = 0,
+        public bool $shuffleChannels = false,
         public bool $stopWhenEmpty = false,
         public ?string $restartSignal = null,
         public ?\Closure $controllerFactory = null,

packages/queue/src/Worker.php

@@ -4,24 +4,12 @@
 
 namespace Windwalker\Queue;
 
-use DateTimeImmutable;
-use Exception;
-use Psr\Log\AbstractLogger;
-use Psr\Log\LoggerInterface;
-use Psr\Log\NullLogger;
 use Throwable;
-use Windwalker\Event\EventAwareInterface;
-use Windwalker\Event\EventAwareTrait;
 use Windwalker\Queue\Attributes\JobBackoff;
 use Windwalker\Queue\Attributes\JobFailed;
 use Windwalker\Queue\Event\AfterJobRunEvent;
 use Windwalker\Queue\Event\BeforeJobRunEvent;
-use Windwalker\Queue\Event\DebugOutputEvent;
 use Windwalker\Queue\Event\JobFailureEvent;
-use Windwalker\Queue\Event\LoopEndEvent;
-use Windwalker\Queue\Event\LoopFailureEvent;
-use Windwalker\Queue\Event\LoopStartEvent;
-use Windwalker\Queue\Event\StopEvent;
 use Windwalker\Queue\Exception\MaxAttemptsExceededException;
 use Windwalker\Queue\Job\JobController;