feat: support subscribe command in Redis

composer.json

@@ -121,7 +121,8 @@
         "symfony/mailer": "^6.2",
         "tijsverkoyen/css-to-inline-styles": "^2.2.5",
         "symfony/process": "^6.2",
-        "dragonmantank/cron-expression": "^3.3.2"
+        "dragonmantank/cron-expression": "^3.3.2",
+        "friendsofhyperf/redis-subscriber": "~3.1.0"
     },
     "replace": {
         "hypervel/auth": "self.version",

src/core/composer.json

@@ -30,7 +30,8 @@
         "hyperf/database": "~3.1.0",
         "hyperf/http-message": "~3.1.0",
         "hyperf/context": "~3.1.0",
-        "hyperf/redis": "~3.1.0"
+        "hyperf/redis": "~3.1.0",
+        "friendsofhyperf/redis-subscriber": "~3.1.0"
     },
     "require-dev": {
         "fakerphp/faker": "^2.0"

src/core/src/Redis/Redis.php

@@ -4,8 +4,8 @@
 
 namespace Hypervel\Redis;
 
+use Closure;
 use Hyperf\Redis\Redis as HyperfRedis;
-use Hyperf\Redis\RedisFactory;
 use Hyperf\Redis\RedisProxy;
 use Hypervel\Context\ApplicationContext;
 
@@ -20,4 +20,22 @@ public function connection(string $name = 'default'): RedisProxy
             ->get(RedisFactory::class)
             ->get($name);
     }
+
+    /**
+     * Subscribe to a set of given channels for messages.
+     */
+    public function subscribe(array|string $channels, Closure $callback): void
+    {
+        $this->connection()
+            ->subscribe($channels, $callback);
+    }
+
+    /**
+     * Subscribe to a set of given channels with wildcards.
+     */
+    public function psubscribe(array|string $channels, Closure $callback): void
+    {
+        $this->connection()
+            ->psubscribe($channels, $callback);
+    }
 }

src/core/src/Redis/RedisFactory.php

@@ -0,0 +1,37 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Hypervel\Redis;
+
+use Hyperf\Contract\ConfigInterface;
+use Hyperf\Redis\Exception\InvalidRedisProxyException;
+
+use function Hyperf\Support\make;
+
+class RedisFactory
+{
+    /**
+     * @var RedisProxy[]
+     */
+    protected array $proxies = [];
+
+    public function __construct(ConfigInterface $config)
+    {
+        $redisConfig = $config->get('redis');
+
+        foreach ($redisConfig as $poolName => $item) {
+            $this->proxies[$poolName] = make(RedisProxy::class, ['pool' => $poolName]);
+        }
+    }
+
+    public function get(string $poolName): RedisProxy
+    {
+        $proxy = $this->proxies[$poolName] ?? null;
+        if (! $proxy instanceof RedisProxy) {
+            throw new InvalidRedisProxyException('Invalid Redis proxy.');
+        }
+
+        return $proxy;
+    }
+}

src/core/src/Redis/RedisProxy.php

@@ -0,0 +1,36 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Hypervel\Redis;
+
+use Closure;
+use Hyperf\Redis\RedisProxy as HyperfRedisProxy;
+
+class RedisProxy extends HyperfRedisProxy
+{
+    /**
+     * Subscribe to a set of given channels for messages.
+     */
+    public function subscribe(array|string $channels, Closure $callback): void
+    {
+        $this->getSubscriber()
+            ->subscribe($channels, $callback);
+    }
+
+    /**
+     * Subscribe to a set of given channels with wildcards.
+     */
+    public function psubscribe(array|string $channels, Closure $callback): void
+    {
+        $this->getSubscriber()
+            ->psubscribe($channels, $callback);
+    }
+
+    protected function getSubscriber(): Subscriber
+    {
+        return new Subscriber(
+            $this->factory->getPool($this->poolName)->getConfig()
+        );
+    }
+}

src/core/src/Redis/Subscriber.php

@@ -0,0 +1,66 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Hypervel\Redis;
+
+use Closure;
+use FriendsOfHyperf\Redis\Subscriber\Exception\SocketException;
+use FriendsOfHyperf\Redis\Subscriber\Subscriber as RedisSubscriber;
+use Hyperf\Contract\StdoutLoggerInterface;
+use Hypervel\Context\ApplicationContext;
+use Hypervel\Support\Arr;
+
+class Subscriber
+{
+    public function __construct(
+        protected array $config,
+        protected ?RedisSubscriber $subscriber = null
+    ) {
+        $this->subscriber = $subscriber ?: $this->createSubscriber($config);
+    }
+
+    protected function createSubscriber(array $config): RedisSubscriber
+    {
+        return new RedisSubscriber(
+            $config['host'] ?? 'localhost',
+            $config['port'] ?? 6379,
+            $config['auth'] ?? '',
+            $config['timeout'] ?? 5,
+            $config['options']['prefix'] ?? '',
+            ApplicationContext::getContainer()->get(StdoutLoggerInterface::class),
+        );
+    }
+
+    /**
+     * Subscribe to a set of given channels for messages.
+     */
+    public function subscribe(array|string $channels, Closure $callback): void
+    {
+        $this->subscriber->subscribe(...Arr::wrap($channels));
+
+        while ($data = $this->subscriber->channel()->pop()) {
+            $callback($data->payload, $data->channel);
+        }
+
+        if (! $this->subscriber->closed) {
+            throw new SocketException('Redis connection is disconnected abnormally.');
+        }
+    }
+
+    /**
+     * Subscribe to a set of given channels with wildcards.
+     */
+    public function psubscribe(array|string $channels, Closure $callback): void
+    {
+        $this->subscriber->psubscribe(...Arr::wrap($channels));
+
+        while ($data = $this->subscriber->channel()->pop()) {
+            $callback($data->payload, $data->channel);
+        }
+
+        if (! $this->subscriber->closed) {
+            throw new SocketException('Redis connection is disconnected abnormally.');
+        }
+    }
+}

tests/Core/RedisSubscriberTest.php

@@ -0,0 +1,181 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Hypervel\Tests\Core;
+
+use FriendsOfHyperf\Redis\Subscriber\Exception\SocketException;
+use FriendsOfHyperf\Redis\Subscriber\Subscriber as RedisSubscriber;
+use Hypervel\Redis\Subscriber;
+use Hypervel\Tests\TestCase;
+use Mockery as m;
+use stdClass;
+
+/**
+ * @internal
+ * @covers \Hypervel\Redis\Subscriber
+ */
+class RedisSubscriberTest extends TestCase
+{
+    public function testSubscribe()
+    {
+        // Arrange
+        $config = ['host' => 'localhost'];
+        $channels = ['channel1', 'channel2'];
+        $mockRedisSubscriber = m::mock(RedisSubscriber::class);
+        $mockChannel = m::mock('stdClass');
+
+        // Set up the channel to return a message once and then null
+        $message = new stdClass();
+        $message->payload = 'test-payload';
+        $message->channel = 'channel1';
+
+        $mockChannel->shouldReceive('pop')
+            ->once()
+            ->andReturn($message);
+        $mockChannel->shouldReceive('pop')
+            ->once()
+            ->andReturnNull();
+
+        $mockRedisSubscriber->shouldReceive('subscribe')
+            ->once()
+            ->with(...$channels);
+
+        $mockRedisSubscriber->shouldReceive('channel')
+            ->twice()
+            ->andReturn($mockChannel);
+
+        $mockRedisSubscriber->closed = true;
+
+        $subscriber = new Subscriber($config, $mockRedisSubscriber);
+
+        $callbackCalled = false;
+        $callback = function ($payload, $channel) use (&$callbackCalled, $message) {
+            $callbackCalled = true;
+            $this->assertEquals($message->payload, $payload);
+            $this->assertEquals($message->channel, $channel);
+        };
+
+        // Act
+        $subscriber->subscribe($channels, $callback);
+
+        // Assert
+        $this->assertTrue($callbackCalled);
+    }
+
+    public function testSubscribeThrowsExceptionWhenConnectionClosedAbnormally()
+    {
+        // Arrange
+        $config = ['host' => 'localhost'];
+        $channels = ['channel1'];
+        $mockRedisSubscriber = m::mock(RedisSubscriber::class);
+        $mockChannel = m::mock('stdClass');
+
+        $mockChannel->shouldReceive('pop')
+            ->once()
+            ->andReturnNull();
+
+        $mockRedisSubscriber->shouldReceive('subscribe')
+            ->once()
+            ->with(...$channels);
+
+        $mockRedisSubscriber->shouldReceive('channel')
+            ->once()
+            ->andReturn($mockChannel);
+
+        $mockRedisSubscriber->closed = false;
+
+        $subscriber = new Subscriber($config, $mockRedisSubscriber);
+
+        $callback = function () {
+            // Empty callback
+        };
+
+        // Assert & Act
+        $this->expectException(SocketException::class);
+        $this->expectExceptionMessage('Redis connection is disconnected abnormally.');
+
+        $subscriber->subscribe($channels, $callback);
+    }
+
+    public function testPsubscribe()
+    {
+        // Arrange
+        $config = ['host' => 'localhost'];
+        $patterns = ['channel*'];
+        $mockRedisSubscriber = m::mock(RedisSubscriber::class);
+        $mockChannel = m::mock('stdClass');
+
+        // Set up the channel to return a message once and then null
+        $message = new stdClass();
+        $message->payload = 'test-payload';
+        $message->channel = 'channel1';
+
+        $mockChannel->shouldReceive('pop')
+            ->once()
+            ->andReturn($message);
+        $mockChannel->shouldReceive('pop')
+            ->once()
+            ->andReturnNull();
+
+        $mockRedisSubscriber->shouldReceive('psubscribe')
+            ->once()
+            ->with(...$patterns);
+
+        $mockRedisSubscriber->shouldReceive('channel')
+            ->twice()
+            ->andReturn($mockChannel);
+
+        $mockRedisSubscriber->closed = true;
+
+        $subscriber = new Subscriber($config, $mockRedisSubscriber);
+
+        $callbackCalled = false;
+        $callback = function ($payload, $channel) use (&$callbackCalled, $message) {
+            $callbackCalled = true;
+            $this->assertEquals($message->payload, $payload);
+            $this->assertEquals($message->channel, $channel);
+        };
+
+        // Act
+        $subscriber->psubscribe($patterns, $callback);
+
+        // Assert
+        $this->assertTrue($callbackCalled);
+    }
+
+    public function testPsubscribeThrowsExceptionWhenConnectionClosedAbnormally()
+    {
+        // Arrange
+        $config = ['host' => 'localhost'];
+        $patterns = ['channel*'];
+        $mockRedisSubscriber = m::mock(RedisSubscriber::class);
+        $mockChannel = m::mock('stdClass');
+
+        $mockChannel->shouldReceive('pop')
+            ->once()
+            ->andReturnNull();
+
+        $mockRedisSubscriber->shouldReceive('psubscribe')
+            ->once()
+            ->with(...$patterns);
+
+        $mockRedisSubscriber->shouldReceive('channel')
+            ->once()
+            ->andReturn($mockChannel);
+
+        $mockRedisSubscriber->closed = false;
+
+        $subscriber = new Subscriber($config, $mockRedisSubscriber);
+
+        $callback = function () {
+            // Empty callback
+        };
+
+        // Assert & Act
+        $this->expectException(SocketException::class);
+        $this->expectExceptionMessage('Redis connection is disconnected abnormally.');
+
+        $subscriber->psubscribe($patterns, $callback);
+    }
+}