Added argument queue for method flush and reload.

Author: limingxinleo

src/Command/FlushFailedMessageCommand.php

@@ -18,6 +18,7 @@
 use Symfony\Component\Console\Command\Command as SymfonyCommand;
 use Symfony\Component\Console\Input\InputArgument;
 use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
 
 /**
@@ -39,10 +40,12 @@ public function __construct(ContainerInterface $container)
     public function execute(InputInterface $input, OutputInterface $output)
     {
         $name = $input->getArgument('name');
+        $queue = $input->getOption('queue');
+
         $factory = $this->container->get(DriverFactory::class);
         $driver = $factory->get($name);
 
-        $driver->flush();
+        $driver->flush($queue);
 
         $output->writeln('<fg=red>Flush all message from failed queue.</>');
     }
@@ -51,5 +54,6 @@ protected function configure()
     {
         $this->setDescription('Delete all message from failed queue.');
         $this->addArgument('name', InputArgument::OPTIONAL, 'The name of queue.', 'default');
+        $this->addOption('queue', 'Q', InputOption::VALUE_OPTIONAL, 'The channel name of queue.');
     }
 }

src/Command/ReloadFailedMessageCommand.php

@@ -18,6 +18,7 @@
 use Symfony\Component\Console\Command\Command as SymfonyCommand;
 use Symfony\Component\Console\Input\InputArgument;
 use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
 
 /**
@@ -39,10 +40,12 @@ public function __construct(ContainerInterface $container)
     public function execute(InputInterface $input, OutputInterface $output)
     {
         $name = $input->getArgument('name');
+        $queue = $input->getOption('queue');
+
         $factory = $this->container->get(DriverFactory::class);
         $driver = $factory->get($name);
 
-        $num = $driver->reload();
+        $num = $driver->reload($queue);
 
         $output->writeln(sprintf('<fg=green>Reload %d failed message into waiting queue.</>', $num));
     }
@@ -51,5 +54,6 @@ protected function configure()
     {
         $this->setDescription('Reload all failed message into waiting queue.');
         $this->addArgument('name', InputArgument::OPTIONAL, 'The name of queue.', 'default');
+        $this->addOption('queue', 'Q', InputOption::VALUE_OPTIONAL, 'The channel name of queue.');
     }
 }

src/Driver/ChannelConfig.php

@@ -12,6 +12,8 @@
 
 namespace Hyperf\AsyncQueue\Driver;
 
+use Hyperf\AsyncQueue\Exception\InvalidQueueException;
+
 class ChannelConfig
 {
     /**
@@ -64,6 +66,15 @@ public function __construct(string $channel)
         $this->timeout = "{$channel}:timeout";
     }
 
+    public function get(string $queue)
+    {
+        if (isset($this->{$queue}) && is_string($this->{$queue})) {
+            return $this->{$queue};
+        }
+
+        throw new InvalidQueueException(sprintf('Queue %s is not exist.', $queue));
+    }
+
     /**
      * @return string
      */

src/Driver/DriverInterface.php

@@ -53,12 +53,12 @@ public function consume(): void;
     /**
      * Reload failed message into waiting queue.
      */
-    public function reload(): int;
+    public function reload(string $queue = null): int;
 
     /**
      * Delete all failed message from failed queue.
      */
-    public function flush(): bool;
+    public function flush(string $queue = null): bool;
 
     /**
      * Return info for current queue.

src/Driver/RedisDriver.php

@@ -12,6 +12,7 @@
 
 namespace Hyperf\AsyncQueue\Driver;
 
+use Hyperf\AsyncQueue\Exception\InvalidQueueException;
 use Hyperf\AsyncQueue\JobInterface;
 use Hyperf\AsyncQueue\Message;
 use Hyperf\AsyncQueue\MessageInterface;
@@ -113,18 +114,32 @@ public function fail($data): bool
         return false;
     }
 
-    public function reload(): int
+    public function reload(string $queue = null): int
     {
+        $channel = $this->channel->getFailed();
+        if ($queue) {
+            if (! in_array($queue, ['timeout', 'failed'])) {
+                throw new InvalidQueueException(sprintf('Queue %s is not supported.', $queue));
+            }
+
+            $channel = $this->channel->get($queue);
+        }
+
         $num = 0;
-        while ($this->redis->rpoplpush($this->channel->getFailed(), $this->channel->getWaiting())) {
+        while ($this->redis->rpoplpush($channel, $this->channel->getWaiting())) {
             ++$num;
         }
         return $num;
     }
 
-    public function flush(): bool
+    public function flush(string $queue = null): bool
     {
-        return (bool) $this->redis->delete($this->channel->getFailed());
+        $channel = $this->channel->getFailed();
+        if ($queue) {
+            $channel = $this->channel->get($queue);
+        }
+
+        return (bool) $this->redis->delete($channel);
     }
 
     public function info(): array

src/Exception/InvalidQueueException.php

@@ -0,0 +1,17 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf.
+ *
+ * @link     https://hyperf.io
+ * @document https://doc.hyperf.io
+ * @contact  group@hyperf.io
+ * @license  https://github.com/hyperf-cloud/hyperf/blob/master/LICENSE
+ */
+
+namespace Hyperf\AsyncQueue\Exception;
+
+class InvalidQueueException extends \RuntimeException
+{
+}