Improve run command (#168)

* Allow to process messages from different channels

* Add "maximum" option to the "run" console command

* Improve messages

* Fix test

* Apply fixes from StyleCI

* Fix typecasting bug

---------

Co-authored-by: StyleCI Bot <bot@styleci.io>

config/di.php

@@ -6,6 +6,7 @@
 use Yiisoft\Queue\Cli\LoopInterface;
 use Yiisoft\Queue\Cli\SignalLoop;
 use Yiisoft\Queue\Cli\SimpleLoop;
+use Yiisoft\Queue\Command\RunCommand;
 use Yiisoft\Queue\Middleware\Consume\ConsumeMiddlewareDispatcher;
 use Yiisoft\Queue\Middleware\Consume\MiddlewareFactoryConsume;
 use Yiisoft\Queue\Middleware\Consume\MiddlewareFactoryConsumeInterface;
@@ -52,4 +53,9 @@
     FailureMiddlewareDispatcher::class => [
         '__construct()' => ['middlewareDefinitions' => $params['yiisoft/queue']['middlewares-fail']],
     ],
+    RunCommand::class => [
+        '__construct()' => [
+            'channels' => array_keys($params['yiisoft/yii-queue']['channel-definitions']),
+        ],
+    ],
 ];

config/params.php

@@ -2,6 +2,7 @@
 
 declare(strict_types=1);
 
+use Yiisoft\Queue\Adapter\AdapterInterface;
 use Yiisoft\Queue\Command\ListenCommand;
 use Yiisoft\Queue\Command\RunCommand;
 use Yiisoft\Queue\Debug\QueueCollector;
@@ -19,7 +20,9 @@
     ],
     'yiisoft/queue' => [
         'handlers' => [],
-        'channel-definitions' => [],
+        'channel-definitions' => [
+            QueueFactoryInterface::DEFAULT_CHANNEL_NAME => AdapterInterface::class,
+        ],
         'middlewares-push' => [],
         'middlewares-consume' => [],
         'middlewares-fail' => [],

src/Command/RunCommand.php

@@ -7,16 +7,17 @@
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputArgument;
 use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
-use Yiisoft\Queue\QueueFactory;
 use Yiisoft\Queue\QueueFactoryInterface;
 
 final class RunCommand extends Command
 {
     protected static $defaultName = 'queue:run';
-    protected static $defaultDescription = 'Runs all the existing messages in the queue. Exits once messages are over.';
+    protected static $defaultDescription = 'Runs all the existing messages in the given queues. ' .
+        'Exits once messages are over.';
 
-    public function __construct(private QueueFactoryInterface $queueFactory)
+    public function __construct(private QueueFactoryInterface $queueFactory, private array $channels)
     {
         parent::__construct();
     }
@@ -25,17 +26,31 @@ public function configure(): void
     {
         $this->addArgument(
             'channel',
-            InputArgument::OPTIONAL,
-            'Queue channel name to connect to',
-            QueueFactory::DEFAULT_CHANNEL_NAME
-        );
+            InputArgument::OPTIONAL | InputArgument::IS_ARRAY,
+            'Queue channel name list to connect to.',
+            $this->channels,
+        )
+            ->addOption(
+                'maximum',
+                'm',
+                InputOption::VALUE_REQUIRED,
+                'Maximum number of messages to process in each channel. Default is 0 (no limits).',
+                0,
+            )
+            ->addUsage('[channel1 [channel2 [...]]] --maximum 100');
     }
 
     protected function execute(InputInterface $input, OutputInterface $output): int
     {
-        $this->queueFactory
-            ->get($input->getArgument('channel'))
-            ->run();
+        /** @var string $channel */
+        foreach ($input->getArgument('channel') as $channel) {
+            $output->write("Processing channel $channel... ");
+            $count = $this->queueFactory
+                ->get($channel)
+                ->run((int)$input->getOption('maximum'));
+
+            $output->writeln("Messages processed: $count.");
+        }
 
         return 0;
     }

src/Debug/QueueDecorator.php

@@ -35,9 +35,9 @@ public function push(
         return $message;
     }
 
-    public function run(int $max = 0): void
+    public function run(int $max = 0): int
     {
-        $this->queue->run($max);
+        return $this->queue->run($max);
     }
 
     public function listen(): void

src/Middleware/FailureHandling/Implementation/ExponentialDelayMiddleware.php

@@ -101,7 +101,11 @@ private function getDelay(MessageInterface $message): float
         $meta = $message->getMetadata();
         $key = self::META_KEY_DELAY . "-$this->id";
 
-        $delayOriginal = (float) ($meta[$key] ?? 0 ?: $this->delayInitial);
+        $delayOriginal = (float) ($meta[$key] ?? 0);
+        if ($delayOriginal <= 0) {
+            $delayOriginal = $this->delayInitial;
+        }
+
         $result = $delayOriginal * $this->exponent;
 
         return min($result, $this->delayMaximum);

src/Queue.php

@@ -67,7 +67,7 @@ public function push(
         return $message;
     }
 
-    public function run(int $max = 0): void
+    public function run(int $max = 0): int
     {
         $this->checkAdapter();
 
@@ -90,6 +90,8 @@ public function run(int $max = 0): void
             'Processed {count} queue messages.',
             ['count' => $count]
         );
+
+        return $count;
     }
 
     public function listen(): void

src/QueueInterface.php

@@ -30,8 +30,10 @@ public function push(MessageInterface $message, MiddlewarePushInterface|callable
      * Execute all existing jobs and exit
      *
      * @param int $max
+     *
+     * @return int How many messages were processed
      */
-    public function run(int $max = 0): void;
+    public function run(int $max = 0): int;
 
     /**
      * Listen to the queue and execute jobs as they come

tests/App/DummyQueue.php

@@ -24,8 +24,9 @@ public function push(
         return $message;
     }
 
-    public function run(int $max = 0): void
+    public function run(int $max = 0): int
     {
+        throw new Exception('`run()` method is not implemented yet.');
     }
 
     public function listen(): void

tests/Unit/Command/RunCommandTest.php

@@ -15,7 +15,7 @@ final class RunCommandTest extends TestCase
 {
     public function testConfigure(): void
     {
-        $command = new RunCommand($this->createMock(QueueFactoryInterface::class));
+        $command = new RunCommand($this->createMock(QueueFactoryInterface::class), []);
         $channelArgument = $command->getNativeDefinition()->getArgument('channel');
         $this->assertEquals('channel', $channelArgument->getName());
     }
@@ -28,7 +28,7 @@ public function testExecute(): void
         $queueFactory->method('get')->willReturn($queue);
         $input = new StringInput('channel');
 
-        $command = new RunCommand($queueFactory);
+        $command = new RunCommand($queueFactory, []);
         $exitCode = $command->run($input, $this->createMock(OutputInterface::class));
 
         $this->assertEquals(0, $exitCode);