Added ReloadChannelListener to reload timeout or failed channels automatically. (#2459)

* Added `ReloadChannelListener` to reload timeout or failed channels automatically.

* Added test cases.

Author: limingxinleo

src/Listener/ReloadChannelListener.php

@@ -0,0 +1,65 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf.
+ *
+ * @link     https://www.hyperf.io
+ * @document https://hyperf.wiki
+ * @contact  group@hyperf.io
+ * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE
+ */
+namespace Hyperf\AsyncQueue\Listener;
+
+use Hyperf\AsyncQueue\Event\QueueLength;
+use Hyperf\Contract\StdoutLoggerInterface;
+use Hyperf\Event\Contract\ListenerInterface;
+
+class ReloadChannelListener implements ListenerInterface
+{
+    /**
+     * @var StdoutLoggerInterface
+     */
+    protected $logger;
+
+    /**
+     * @var string[]
+     */
+    protected $channels = [
+        'timeout',
+    ];
+
+    public function __construct(StdoutLoggerInterface $logger)
+    {
+        $this->logger = $logger;
+    }
+
+    public function listen(): array
+    {
+        return [
+            QueueLength::class,
+        ];
+    }
+
+    /**
+     * @param QueueLength $event
+     */
+    public function process(object $event)
+    {
+        if (! $event instanceof QueueLength) {
+            return;
+        }
+
+        if (! in_array($event->key, $this->channels)) {
+            return;
+        }
+
+        if ($event->length == 0) {
+            return;
+        }
+
+        $event->driver->reload($event->key);
+
+        $this->logger->info(sprintf('%s channel reload %d messages to waiting channel success.', $event->key, $event->length));
+    }
+}

tests/ListenerTest.php

@@ -0,0 +1,45 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf.
+ *
+ * @link     https://www.hyperf.io
+ * @document https://hyperf.wiki
+ * @contact  group@hyperf.io
+ * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE
+ */
+namespace HyperfTest\AsyncQueue;
+
+use Hyperf\AsyncQueue\Driver\DriverInterface;
+use Hyperf\AsyncQueue\Event\QueueLength;
+use Hyperf\AsyncQueue\Listener\ReloadChannelListener;
+use Hyperf\Contract\StdoutLoggerInterface;
+use Mockery;
+use PHPUnit\Framework\TestCase;
+
+/**
+ * @internal
+ * @coversNothing
+ */
+class ListenerTest extends TestCase
+{
+    protected function tearDown()
+    {
+        Mockery::close();
+    }
+
+    public function testReloadChannelListener()
+    {
+        $logger = Mockery::mock(StdoutLoggerInterface::class);
+        $logger->shouldReceive('info')->withAnyArgs()->once()->andReturnUsing(function ($message) {
+            $this->assertSame('timeout channel reload 10 messages to waiting channel success.', $message);
+        });
+        $driver = Mockery::mock(DriverInterface::class);
+        $driver->shouldReceive('reload')->withAnyArgs()->andReturn(0);
+        $listener = new ReloadChannelListener($logger);
+        $listener->process(new QueueLength($driver, 'failed', 10));
+        $listener->process(new QueueLength($driver, 'timeout', 10));
+        $listener->process(new QueueLength($driver, 'timeout', 0));
+    }
+}