Divide queue and channel names

Makefile

@@ -1,6 +1,6 @@
 .DEFAULT_GOAL := help
 
-PHP_VERSION ?= 8.5
+PHP_VERSION ?= 8.4
 -include .env.local
 
 DOCKER_RUN := docker run --rm -it \

README.md

@@ -185,10 +185,9 @@ use Yiisoft\Queue\Adapter\SynchronousAdapter;
 
 [
     'channel1' => new SynchronousAdapter(),
-    'channel2' => static fn(SynchronousAdapter $adapter) => $adapter->withChannel('channel2'),
-    'channel3' => [
+    'channel2' => new SynchronousAdapter(), // a second instance for a different queue processing pipeline
+    'channel3' => [ // use a yiisoft/factory syntax for adapter creation
         'class' => SynchronousAdapter::class,
-        '__constructor' => ['channel' => 'channel3'],
     ],
 ]
 ```

config/di.php

@@ -31,7 +31,7 @@
 return [
     AdapterFactoryQueueProvider::class => [
         '__construct()' => [
-            'definitions' => $params['yiisoft/queue']['channels'],
+            'definitions' => $params['yiisoft/queue']['queues'],
         ],
     ],
     QueueProviderInterface::class => AdapterFactoryQueueProvider::class,
@@ -61,12 +61,12 @@
     MessageSerializerInterface::class => JsonMessageSerializer::class,
     RunCommand::class => [
         '__construct()' => [
-            'channels' => array_keys($params['yiisoft/queue']['channels']),
+            'queues' => array_keys($params['yiisoft/queue']['queues']),
         ],
     ],
     ListenAllCommand::class => [
         '__construct()' => [
-            'channels' => array_keys($params['yiisoft/queue']['channels']),
+            'queues' => array_keys($params['yiisoft/queue']['queues']),
         ],
     ],
 ];

config/params.php

@@ -22,8 +22,8 @@
     ],
     'yiisoft/queue' => [
         'handlers' => [],
-        'channels' => [
-            QueueProviderInterface::DEFAULT_CHANNEL => AdapterInterface::class,
+        'queues' => [
+            QueueProviderInterface::DEFAULT_QUEUE => AdapterInterface::class,
         ],
         'middlewares-push' => [],
         'middlewares-consume' => [],

src/Adapter/AdapterInterface.php

@@ -4,7 +4,6 @@
 
 namespace Yiisoft\Queue\Adapter;
 
-use BackedEnum;
 use InvalidArgumentException;
 use Yiisoft\Queue\JobStatus;
 use Yiisoft\Queue\Message\MessageInterface;
@@ -38,8 +37,4 @@ public function push(MessageInterface $message): MessageInterface;
      * @param callable(MessageInterface): bool $handlerCallback The handler which will handle messages. Returns false if it cannot continue handling messages.
      */
     public function subscribe(callable $handlerCallback): void;
-
-    public function withChannel(string|BackedEnum $channel): self;
-
-    public function getChannel(): string;
 }

src/Adapter/SynchronousAdapter.php

@@ -4,12 +4,9 @@
 
 namespace Yiisoft\Queue\Adapter;
 
-use BackedEnum;
 use InvalidArgumentException;
-use Yiisoft\Queue\ChannelNormalizer;
 use Yiisoft\Queue\JobStatus;
 use Yiisoft\Queue\Message\MessageInterface;
-use Yiisoft\Queue\Provider\QueueProviderInterface;
 use Yiisoft\Queue\QueueInterface;
 use Yiisoft\Queue\Worker\WorkerInterface;
 use Yiisoft\Queue\Message\IdEnvelope;
@@ -20,15 +17,11 @@ final class SynchronousAdapter implements AdapterInterface
 {
     private array $messages = [];
     private int $current = 0;
-    private string $channel;
 
     public function __construct(
         private readonly WorkerInterface $worker,
         private readonly QueueInterface $queue,
-        string|BackedEnum $channel = QueueProviderInterface::DEFAULT_CHANNEL,
-    ) {
-        $this->channel = ChannelNormalizer::normalize($channel);
-    }
+    ) {}
 
     public function __destruct()
     {
@@ -80,24 +73,4 @@ public function subscribe(callable $handlerCallback): void
     {
         $this->runExisting($handlerCallback);
     }
-
-    public function withChannel(string|BackedEnum $channel): self
-    {
-        $channel = ChannelNormalizer::normalize($channel);
-
-        if ($channel === $this->channel) {
-            return $this;
-        }
-
-        $new = clone $this;
-        $new->channel = $channel;
-        $new->messages = [];
-
-        return $new;
-    }
-
-    public function getChannel(): string
-    {
-        return $this->channel;
-    }
 }

src/Command/ListenAllCommand.php

@@ -25,7 +25,7 @@
     public function __construct(
         private readonly QueueProviderInterface $queueProvider,
         private readonly LoopInterface $loop,
-        private readonly array $channels,
+        private readonly array $queues,
     ) {
         parent::__construct();
     }
@@ -36,47 +36,47 @@
     public function configure(): void
     {
         $this->addArgument(
-            'channel',
+            'queue',
             InputArgument::OPTIONAL | InputArgument::IS_ARRAY,
-            'Queue channel name list to connect to',
-            $this->channels,
+            'Queue name list to connect to',
+            $this->queues,
         )
             ->addOption(
                 'pause',
                 'p',
                 InputOption::VALUE_REQUIRED,
-                'Pause between queue channel iterations in seconds. May save some CPU. Default: 1',
+                'Pause between queue iterations in seconds. May save some CPU. Default: 1',
                 1,
             )
             ->addOption(
                 'maximum',
                 'm',
                 InputOption::VALUE_REQUIRED,
-                'Maximum number of messages to process in each channel before switching to another channel. '
+                'Maximum number of messages to process in each queue before switching to another queue. '
                     . 'Default is 0 (no limits).',
                 0,
             );
 
-        $this->addUsage('[channel1 [channel2 [...]]] [--timeout=<timeout>] [--maximum=<maximum>]');
+        $this->addUsage('[queue1 [queue2 [...]]] [--timeout=<timeout>] [--maximum=<maximum>]');
     }
 
     protected function execute(InputInterface $input, OutputInterface $output): int
     {
         $queues = [];
-        /** @var string $channel */
-        foreach ($input->getArgument('channel') as $channel) {
-            $queues[] = $this->queueProvider->get($channel);
+        /** @var string $queue */
+        foreach ($input->getArgument('queue') as $queue) {
+            $queues[] = $this->queueProvider->get($queue);
         }
 
         $pauseSeconds = (int) $input->getOption('pause');
         if ($pauseSeconds < 0) {
             $pauseSeconds = 1;
         }
 
         while ($this->loop->canContinue()) {
             $hasMessages = false;
             foreach ($queues as $queue) {
                 $hasMessages = $queue->run((int) $input->getOption('maximum')) > 0 || $hasMessages;
             }
 
             if (!$hasMessages) {

src/Command/ListenCommand.php

@@ -26,17 +26,17 @@ public function __construct(
     public function configure(): void
     {
         $this->addArgument(
-            'channel',
+            'queue',
             InputArgument::OPTIONAL,
-            'Queue channel name to connect to',
-            QueueProviderInterface::DEFAULT_CHANNEL,
+            'Queue name to connect to',
+            QueueProviderInterface::DEFAULT_QUEUE,
         );
     }
 
     protected function execute(InputInterface $input, OutputInterface $output): int
     {
         $this->queueProvider
-            ->get($input->getArgument('channel'))
+            ->get($input->getArgument('queue'))
             ->listen();
 
         return 0;

src/Command/RunCommand.php

@@ -20,36 +20,36 @@ final class RunCommand extends Command
 {
     public function __construct(
         private readonly QueueProviderInterface $queueProvider,
-        private readonly array $channels,
+        private readonly array $queues,
     ) {
         parent::__construct();
     }
 
     public function configure(): void
     {
         $this->addArgument(
-            'channel',
+            'queue',
             InputArgument::OPTIONAL | InputArgument::IS_ARRAY,
-            'Queue channel name list to connect to.',
-            $this->channels,
+            'Queue name list to connect to.',
+            $this->queues,
         )
             ->addOption(
                 'maximum',
                 'm',
                 InputOption::VALUE_REQUIRED,
-                'Maximum number of messages to process in each channel. Default is 0 (no limits).',
+                'Maximum number of messages to process in each queue. Default is 0 (no limits).',
                 0,
             )
-            ->addUsage('[channel1 [channel2 [...]]] --maximum 100');
+            ->addUsage('[queue1 [queue2 [...]]] --maximum 100');
     }
 
     protected function execute(InputInterface $input, OutputInterface $output): int
     {
-        /** @var string $channel */
-        foreach ($input->getArgument('channel') as $channel) {
-            $output->write("Processing channel $channel... ");
+        /** @var string $queue */
+        foreach ($input->getArgument('queue') as $queue) {
+            $output->write("Processing queue $queue... ");
             $count = $this->queueProvider
-                ->get($channel)
+                ->get($queue)
                 ->run((int) $input->getOption('maximum'));
 
             $output->writeln("Messages processed: $count.");

src/Debug/QueueCollector.php

@@ -47,18 +47,15 @@ public function collectStatus(string $id, JobStatus $status): void
     }
 
     public function collectPush(
-        ?string $channel,
+        string $queueName,
         MessageInterface $message,
         string|array|callable|MiddlewarePushInterface ...$middlewareDefinitions,
     ): void {
         if (!$this->isActive()) {
             return;
         }
-        if ($channel === null) {
-            $channel = 'null';
-        }
 
-        $this->pushes[$channel][] = [
+        $this->pushes[$queueName][] = [
             'message' => $message,
             'middlewares' => $middlewareDefinitions,
         ];
@@ -69,7 +66,7 @@ public function collectWorkerProcessing(MessageInterface $message, QueueInterfac
         if (!$this->isActive()) {
             return;
         }
-        $this->processingMessages[$queue->getChannel()][] = $message;
+        $this->processingMessages[$queue->getName()][] = $message;
     }
 
     public function getSummary(): array

src/Debug/QueueDecorator.php

@@ -4,6 +4,7 @@
 
 namespace Yiisoft\Queue\Debug;
 
+use BackedEnum;
 use Yiisoft\Queue\Adapter\AdapterInterface;
 use Yiisoft\Queue\JobStatus;
 use Yiisoft\Queue\Message\MessageInterface;
@@ -30,7 +31,7 @@ public function push(
         string|array|callable|MiddlewarePushInterface ...$middlewareDefinitions,
     ): MessageInterface {
         $message = $this->queue->push($message, ...$middlewareDefinitions);
-        $this->collector->collectPush($this->queue->getChannel(), $message, ...$middlewareDefinitions);
+        $this->collector->collectPush($this->queue->getName(), $message, ...$middlewareDefinitions);
         return $message;
     }
 
@@ -44,13 +45,13 @@ public function listen(): void
         $this->queue->listen();
     }
 
-    public function withAdapter(AdapterInterface $adapter): static
+    public function withAdapter(AdapterInterface $adapter, string|BackedEnum|null $queueName = null): static
     {
-        return new self($this->queue->withAdapter($adapter), $this->collector);
+        return new self($this->queue->withAdapter($adapter, $queueName), $this->collector);
     }
 
-    public function getChannel(): string
+    public function getName(): string
     {
-        return $this->queue->getChannel();
+        return $this->queue->getName();
     }
 }

src/Debug/QueueProviderInterfaceProxy.php

@@ -15,14 +15,15 @@ public function __construct(
         private readonly QueueCollector $collector,
     ) {}
 
-    public function get(string|BackedEnum $channel): QueueInterface
+    public function get(string|BackedEnum $name): QueueInterface
     {
-        $queue = $this->queueProvider->get($channel);
+        $queue = $this->queueProvider->get($name);
+
         return new QueueDecorator($queue, $this->collector);
     }
 
-    public function has(string|BackedEnum $channel): bool
+    public function has(string|BackedEnum $name): bool
     {
-        return $this->queueProvider->has($channel);
+        return $this->queueProvider->has($name);
     }
 }

src/Middleware/FailureHandling/FailureMiddlewareDispatcher.php

@@ -37,18 +37,17 @@ public function dispatch(
         FailureHandlingRequest $request,
         MessageFailureHandlerInterface $finishHandler,
     ): FailureHandlingRequest {
-        /** @var string $channel It is always string in this context */
-        $channel = $request->getQueue()->getChannel();
-        if (!isset($this->middlewareDefinitions[$channel]) || $this->middlewareDefinitions[$channel] === []) {
-            $channel = self::DEFAULT_PIPELINE;
+        $queueName = $request->getQueue()->getName();
+        if (!isset($this->middlewareDefinitions[$queueName]) || $this->middlewareDefinitions[$queueName] === []) {
+            $queueName = self::DEFAULT_PIPELINE;
         }
-        $definitions = array_reverse($this->middlewareDefinitions[$channel]);
+        $definitions = array_reverse($this->middlewareDefinitions[$queueName]);
 
-        if (!isset($this->stack[$channel])) {
-            $this->stack[$channel] = new MiddlewareFailureStack($this->buildMiddlewares(...$definitions), $finishHandler);
+        if (!isset($this->stack[$queueName])) {
+            $this->stack[$queueName] = new MiddlewareFailureStack($this->buildMiddlewares(...$definitions), $finishHandler);
         }
 
-        return $this->stack[$channel]->handleFailure($request);
+        return $this->stack[$queueName]->handleFailure($request);
     }
 
     /**