[fix] Redis connection counter didn't work properly

config/websockets.php

@@ -42,21 +42,6 @@
 
         'app' => \BeyondCode\LaravelWebSockets\Apps\ConfigAppManager::class,
 
-        /*
-        |--------------------------------------------------------------------------
-        | Channel Manager
-        |--------------------------------------------------------------------------
-        |
-        | When users subscribe or unsubscribe from specific channels,
-        | the connections are stored to keep track of any interaction with the
-        | WebSocket server.
-        | You can however add your own implementation that will help the store
-        | of the channels alongside their connections.
-        |
-        */
-
-        'channel' => \BeyondCode\LaravelWebSockets\WebSockets\Channels\ChannelManagers\ArrayChannelManager::class,
-
     ],
 
     /*
@@ -191,6 +176,8 @@
 
             'statistics_logger' => \BeyondCode\LaravelWebSockets\Statistics\Logger\MemoryStatisticsLogger::class,
 
+            'channel_manager' => \BeyondCode\LaravelWebSockets\WebSockets\Channels\ChannelManagers\ArrayChannelManager::class,
+
         ],
 
         /*
@@ -214,6 +201,8 @@
 
             'statistics_logger' => \BeyondCode\LaravelWebSockets\Statistics\Logger\RedisStatisticsLogger::class,
 
+            'channel_manager' => \BeyondCode\LaravelWebSockets\WebSockets\Channels\ChannelManagers\RedisChannelManager::class,
+
         ],
 
     ],

src/PubSub/Drivers/LocalClient.php

@@ -66,6 +66,28 @@ public function unsubscribe($appId, string $channel): bool
         return true;
     }
 
+    /**
+     * Subscribe to the app's pubsub keyspace.
+     *
+     * @param  mixed  $appId
+     * @return bool
+     */
+    public function subscribeToApp($appId): bool
+    {
+        return true;
+    }
+
+    /**
+     * Unsubscribe from the app's pubsub keyspace.
+     *
+     * @param  mixed  $appId
+     * @return bool
+     */
+    public function unsubscribeFromApp($appId): bool
+    {
+        return true;
+    }
+
     /**
      * Add a member to a channel. To be called when they have
      * subscribed to the channel.
@@ -137,4 +159,15 @@ public function channelMemberCounts($appId, array $channelNames): PromiseInterfa
 
         return new FulfilledPromise($results);
     }
+
+    /**
+     * Get the amount of unique connections.
+     *
+     * @param  mixed  $appId
+     * @return null|int|\React\Promise\PromiseInterface
+     */
+    public function appConnectionsCount($appId)
+    {
+        return null;
+    }
 }

src/PubSub/Drivers/RedisClient.php

@@ -7,6 +7,7 @@
 use BeyondCode\LaravelWebSockets\WebSockets\Channels\ChannelManager;
 use Clue\React\Redis\Client;
 use Clue\React\Redis\Factory;
+use Illuminate\Support\Facades\Cache;
 use Illuminate\Support\Str;
 use React\EventLoop\LoopInterface;
 use React\Promise\PromiseInterface;
@@ -42,6 +43,13 @@ class RedisClient extends LocalClient
      */
     protected $subscribeClient;
 
+    /**
+     * The Redis manager instance.
+     *
+     * @var \Illuminate\Redis\RedisManager
+     */
+    protected $redis;
+
     /**
      * Mapping of subscribed channels, where the key is the channel name,
      * and the value is the amount of connections which are subscribed to
@@ -60,6 +68,7 @@ class RedisClient extends LocalClient
     public function __construct()
     {
         $this->serverId = Str::uuid()->toString();
+        $this->redis = Cache::getRedis();
     }
 
     /**
@@ -175,6 +184,36 @@ public function unsubscribe($appId, string $channel): bool
         return true;
     }
 
+    /**
+     * Subscribe to the app's pubsub keyspace.
+     *
+     * @param  mixed  $appId
+     * @return bool
+     */
+    public function subscribeToApp($appId): bool
+    {
+        $this->subscribeClient->__call('subscribe', [$this->getTopicName($appId)]);
+
+        $this->redis->hincrby($this->getTopicName($appId), 'connections', 1);
+
+        return true;
+    }
+
+    /**
+     * Unsubscribe from the app's pubsub keyspace.
+     *
+     * @param  mixed  $appId
+     * @return bool
+     */
+    public function unsubscribeFromApp($appId): bool
+    {
+        $this->subscribeClient->__call('unsubscribe', [$this->getTopicName($appId)]);
+
+        $this->redis->hincrby($this->getTopicName($appId), 'connections', -1);
+
+        return true;
+    }
+
     /**
      * Add a member to a channel. To be called when they have
      * subscribed to the channel.
@@ -187,7 +226,7 @@ public function unsubscribe($appId, string $channel): bool
      */
     public function joinChannel($appId, string $channel, string $socketId, string $data)
     {
-        $this->publishClient->__call('hset', [$this->getTopicName($appId, $channel), $socketId, $data]);
+        $this->redis->hset($this->getTopicName($appId, $channel), $socketId, $data);
 
         DashboardLogger::log($appId, DashboardLogger::TYPE_REPLICATOR_JOINED_CHANNEL, [
             'channel' => $channel,
@@ -209,7 +248,7 @@ public function joinChannel($appId, string $channel, string $socketId, string $d
      */
     public function leaveChannel($appId, string $channel, string $socketId)
     {
-        $this->publishClient->__call('hdel', [$this->getTopicName($appId, $channel), $socketId]);
+        $this->redis->hdel($this->getTopicName($appId, $channel), $socketId);
 
         DashboardLogger::log($appId, DashboardLogger::TYPE_REPLICATOR_LEFT_CHANNEL, [
             'channel' => $channel,
@@ -258,6 +297,19 @@ public function channelMemberCounts($appId, array $channelNames): PromiseInterfa
             });
     }
 
+    /**
+     * Get the amount of unique connections.
+     *
+     * @param  mixed  $appId
+     * @return null|int|\React\Promise\PromiseInterface
+     */
+    public function appConnectionsCount($appId)
+    {
+        // Use the in-built Redis manager to avoid async run.
+
+        return $this->redis->hget($this->getTopicName($appId), 'connections') ?: 0;
+    }
+
     /**
      * Handle a message received from Redis on a specific channel.
      *
@@ -377,13 +429,19 @@ public function getServerId()
      * app ID and channel name.
      *
      * @param  mixed  $appId
-     * @param  string  $channel
+     * @param  string|null  $channel
      * @return string
      */
-    protected function getTopicName($appId, string $channel): string
+    protected function getTopicName($appId, string $channel = null): string
     {
         $prefix = config('database.redis.options.prefix', null);
 
-        return "{$prefix}{$appId}:{$channel}";
+        $hash = "{$prefix}{$appId}";
+
+        if ($channel) {
+            $hash .= ":{$channel}";
+        }
+
+        return $hash;
     }
 }

src/PubSub/ReplicationInterface.php

@@ -45,6 +45,22 @@ public function subscribe($appId, string $channel): bool;
      */
     public function unsubscribe($appId, string $channel): bool;
 
+    /**
+     * Subscribe to the app's pubsub keyspace.
+     *
+     * @param  mixed  $appId
+     * @return bool
+     */
+    public function subscribeToApp($appId): bool;
+
+    /**
+     * Unsubscribe from the app's pubsub keyspace.
+     *
+     * @param  mixed  $appId
+     * @return bool
+     */
+    public function unsubscribeFromApp($appId): bool;
+
     /**
      * Add a member to a channel. To be called when they have
      * subscribed to the channel.
@@ -85,4 +101,12 @@ public function channelMembers($appId, string $channel): PromiseInterface;
      * @return PromiseInterface
      */
     public function channelMemberCounts($appId, array $channelNames): PromiseInterface;
+
+    /**
+     * Get the amount of unique connections.
+     *
+     * @param  mixed  $appId
+     * @return null|int|\React\Promise\PromiseInterface
+     */
+    public function appConnectionsCount($appId);
 }

src/WebSockets/Channels/ChannelManagers/RedisChannelManager.php

@@ -0,0 +1,36 @@
+<?php
+
+namespace BeyondCode\LaravelWebSockets\WebSockets\Channels\ChannelManagers;
+
+use BeyondCode\LaravelWebSockets\PubSub\ReplicationInterface;
+
+class RedisChannelManager extends ArrayChannelManager
+{
+    /**
+     * The replicator driver.
+     *
+     * @var \BeyondCode\LaravelWebSockets\PubSub\ReplicationInterface
+     */
+    protected $replicator;
+
+    /**
+     * Initialize the channel manager.
+     *
+     * @return void
+     */
+    public function __construct()
+    {
+        $this->replicator = app(ReplicationInterface::class);
+    }
+
+    /**
+     * Get the connections count on the app.
+     *
+     * @param  mixed  $appId
+     * @return int
+     */
+    public function getConnectionCount($appId): int
+    {
+        return $this->replicator->appConnectionsCount($appId);
+    }
+}

src/WebSockets/WebSocketHandler.php

@@ -5,6 +5,7 @@
 use BeyondCode\LaravelWebSockets\Apps\App;
 use BeyondCode\LaravelWebSockets\Dashboard\DashboardLogger;
 use BeyondCode\LaravelWebSockets\Facades\StatisticsLogger;
+use BeyondCode\LaravelWebSockets\PubSub\ReplicationInterface;
 use BeyondCode\LaravelWebSockets\QueryParameters;
 use BeyondCode\LaravelWebSockets\WebSockets\Channels\ChannelManager;
 use BeyondCode\LaravelWebSockets\WebSockets\Exceptions\ConnectionsOverCapacity;
@@ -26,6 +27,13 @@ class WebSocketHandler implements MessageComponentInterface
      */
     protected $channelManager;
 
+    /**
+     * The replicator client.
+     *
+     * @var ReplicationInterface
+     */
+    protected $replicator;
+
     /**
      * Initialize a new handler.
      *
@@ -35,6 +43,7 @@ class WebSocketHandler implements MessageComponentInterface
     public function __construct(ChannelManager $channelManager)
     {
         $this->channelManager = $channelManager;
+        $this->replicator = app(ReplicationInterface::class);
     }
 
     /**
@@ -83,6 +92,8 @@ public function onClose(ConnectionInterface $connection)
         ]);
 
         StatisticsLogger::disconnection($connection->app->id);
+
+        $this->replicator->unsubscribeFromApp($connection->app->id);
     }
 
     /**
@@ -99,6 +110,8 @@ public function onError(ConnectionInterface $connection, Exception $exception)
                 $exception->getPayload()
             ));
         }
+
+        $this->replicator->unsubscribeFromApp($connection->app->id);
     }
 
     /**
@@ -203,6 +216,8 @@ protected function establishConnection(ConnectionInterface $connection)
 
         StatisticsLogger::connection($connection->app->id);
 
+        $this->replicator->subscribeToApp($connection->app->id);
+
         return $this;
     }
 }

src/WebSocketsServiceProvider.php

@@ -59,9 +59,11 @@ public function register()
         });
 
         $this->app->singleton(ChannelManager::class, function () {
-            $channelManager = config('websockets.managers.channel', ArrayChannelManager::class);
+            $replicationDriver = config('websockets.replication.driver', 'local');
 
-            return new $channelManager;
+            $class = config("websockets.replication.{$replicationDriver}.channel_manager", ArrayChannelManager::class);
+
+            return new $class;
         });
 
         $this->app->singleton(AppManager::class, function () {
@@ -72,9 +74,10 @@ public function register()
             $driver = config('websockets.statistics.driver');
 
             return $this->app->make(
-                config('websockets.statistics')[$driver]['driver']
-                ??
-                \BeyondCode\LaravelWebSockets\Statistics\Drivers\DatabaseDriver::class
+                config(
+                    "websockets.statistics.{$driver}.driver",
+                    \BeyondCode\LaravelWebSockets\Statistics\Drivers\DatabaseDriver::class
+                )
             );
         });
     }