Added redis connection counter.

Author: rennokki

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.
@@ -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 () {

tests/ConnectionTest.php

@@ -7,6 +7,7 @@
 use BeyondCode\LaravelWebSockets\WebSockets\Exceptions\ConnectionsOverCapacity;
 use BeyondCode\LaravelWebSockets\WebSockets\Exceptions\OriginNotAllowed;
 use BeyondCode\LaravelWebSockets\WebSockets\Exceptions\UnknownAppKey;
+use Illuminate\Support\Facades\Cache;
 
 class ConnectionTest extends TestCase
 {
@@ -31,6 +32,25 @@ public function known_app_keys_can_connect()
     /** @test */
     public function app_can_not_exceed_maximum_capacity()
     {
+        $this->runOnlyOnLocalReplication();
+
+        $this->app['config']->set('websockets.apps.0.capacity', 2);
+
+        $this->getConnectedWebSocketConnection(['test-channel']);
+        $this->getConnectedWebSocketConnection(['test-channel']);
+        $this->expectException(ConnectionsOverCapacity::class);
+        $this->getConnectedWebSocketConnection(['test-channel']);
+    }
+
+    /** @test */
+    public function app_can_not_exceed_maximum_capacity_on_redis_replication()
+    {
+        $this->runOnlyOnRedisReplication();
+
+        $redis = Cache::getRedis();
+
+        $redis->hdel('laravel_database_1234', 'connections');
+
         $this->app['config']->set('websockets.apps.0.capacity', 2);
 
         $this->getConnectedWebSocketConnection(['test-channel']);