Merge branch 'main' into feature/replcae-laminas-mime

Author: albertcht

composer.json

@@ -122,6 +122,7 @@
         "tijsverkoyen/css-to-inline-styles": "^2.2.5",
         "symfony/process": "^6.2",
         "dragonmantank/cron-expression": "^3.3.2",
+        "friendsofhyperf/redis-subscriber": "~3.1.0",
         "hypervel/laminas-mime": "^0.1.0"
     },
     "replace": {

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.');
+        }
+    }
+}

src/queue/src/Console/ListenCommand.php

@@ -36,16 +36,13 @@ class ListenCommand extends Command
      */
     protected string $description = 'Listen to a given queue';
 
-    /**
-     * The queue listener instance.
-     */
-    protected Listener $listener;
-
     /**
      * Create a new queue listen command.
      */
-    public function __construct(Listener $listener)
-    {
+    public function __construct(
+        protected ConfigInterface $config,
+        protected Listener $listener
+    ) {
         parent::__construct();
 
         $this->setOutputHandler($this->listener = $listener);
@@ -77,9 +74,9 @@ public function handle()
      */
     protected function getQueue(?string $connection): string
     {
-        $connection = $connection ?: $this->app->get(ConfigInterface::class)->get('queue.default');
+        $connection = $connection ?: $this->config->get('queue.default');
 
-        return $this->input->getOption('queue') ?: $this->app->get(ConfigInterface::class)->get(
+        return $this->input->getOption('queue') ?: $this->config->get(
             "queue.connections.{$connection}.queue",
             'default'
         );

src/queue/src/Console/MonitorCommand.php

@@ -38,7 +38,8 @@ class MonitorCommand extends Command
      */
     public function __construct(
         protected Factory $manager,
-        protected EventDispatcherInterface $events
+        protected EventDispatcherInterface $events,
+        protected ConfigInterface $config,
     ) {
         parent::__construct();
     }
@@ -66,7 +67,7 @@ protected function parseQueues($queues): Collection
 
             if (! isset($queue)) {
                 $queue = $connection;
-                $connection = $this->app->get(ConfigInterface::class)->get('queue.default');
+                $connection = $this->config->get('queue.default');
             }
 
             return [

src/queue/src/Console/RetryCommand.php

@@ -14,6 +14,7 @@
 use Hypervel\Queue\Events\JobRetryRequested;
 use Hypervel\Queue\Failed\FailedJobProviderInterface;
 use Hypervel\Support\Traits\HasLaravelStyleCommand;
+use Psr\Container\ContainerInterface;
 use Psr\EventDispatcher\EventDispatcherInterface;
 use RuntimeException;
 use stdClass;
@@ -39,6 +40,7 @@ class RetryCommand extends Command
      * Create a new queue restart command.
      */
     public function __construct(
+        protected ContainerInterface $app,
         protected FailedJobProviderInterface $failer
     ) {
         parent::__construct();