Divide queue and channel names (#259)

Co-authored-by: Sergei Predvoditelev <sergei@predvoditelev.ru>
Co-authored-by: Alexander Makarov <sam@rmcreative.ru>

Author: 3 people

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/ChannelNormalizer.php

@@ -25,7 +25,7 @@ final class ListenAllCommand extends Command
     public function __construct(
         private readonly QueueProviderInterface $queueProvider,
         private readonly LoopInterface $loop,
-        private readonly array $channels,
+        private readonly array $queues,
     ) {
         parent::__construct();
     }
@@ -36,36 +36,36 @@ public function __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(
                 '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');

src/Command/ListenAllCommand.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/ListenCommand.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.");