Merge pull request #497 from beyondcode/fix/app-connections-count

[fix] Redis connection counter didn't work properly

Author: rennokki

composer.json

@@ -36,11 +36,12 @@
         "illuminate/routing": "^6.0|^7.0",
         "illuminate/support": "^6.0|^7.0",
         "pusher/pusher-php-server": "^3.0|^4.0",
-        "react/dns": "^1.1",
+        "react/promise": "^2.0",
         "symfony/http-kernel": "^4.0|^5.0",
         "symfony/psr-http-message-bridge": "^1.1|^2.0"
     },
     "require-dev": {
+        "clue/block-react": "^1.4",
         "mockery/mockery": "^1.3",
         "orchestra/testbench-browser-kit": "^4.0|^5.0",
         "phpunit/phpunit": "^8.0|^9.0"

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,26 @@ public function channelMemberCounts($appId, array $channelNames): PromiseInterfa
 
         return new FulfilledPromise($results);
     }
+
+    /**
+     * Get the amount of unique connections.
+     *
+     * @param  mixed  $appId
+     * @return null|int
+     */
+    public function getLocalConnectionsCount($appId)
+    {
+        return null;
+    }
+
+    /**
+     * Get the amount of connections aggregated on multiple instances.
+     *
+     * @param  mixed  $appId
+     * @return null|int|\React\Promise\PromiseInterface
+     */
+    public function getGlobalConnectionsCount($appId)
+    {
+        return null;
+    }
 }

src/PubSub/Drivers/RedisClient.php

@@ -175,6 +175,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->publishClient->__call('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->publishClient->__call('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 +288,17 @@ public function channelMemberCounts($appId, array $channelNames): PromiseInterfa
             });
     }
 
+    /**
+     * Get the amount of connections aggregated on multiple instances.
+     *
+     * @param  mixed  $appId
+     * @return null|int|\React\Promise\PromiseInterface
+     */
+    public function getGlobalConnectionsCount($appId)
+    {
+        return $this->publishClient->hget($this->getTopicName($appId), 'connections');
+    }
+
     /**
      * Handle a message received from Redis on a specific channel.
      *
@@ -321,8 +362,8 @@ public function onMessage(string $redisChannel, string $payload)
      */
     protected function getConnectionUri()
     {
-        $name = config('websockets.replication.redis.connection') ?: 'default';
-        $config = config('database.redis')[$name];
+        $name = config('websockets.replication.redis.connection', 'default');
+        $config = config("database.redis.{$name}");
 
         $host = $config['host'];
         $port = $config['port'] ?: 6379;
@@ -377,13 +418,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,20 @@ 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
+     */
+    public function getLocalConnectionsCount($appId);
+
+    /**
+     * Get the amount of connections aggregated on multiple instances.
+     *
+     * @param  mixed  $appId
+     * @return null|int|\React\Promise\PromiseInterface
+     */
+    public function getGlobalConnectionsCount($appId);
 }

src/Statistics/Logger/MemoryStatisticsLogger.php

@@ -105,7 +105,7 @@ public function save()
 
             $this->createRecord($statistic, $appId);
 
-            $currentConnectionCount = $this->channelManager->getConnectionCount($appId);
+            $currentConnectionCount = $this->channelManager->getGlobalConnectionsCount($appId);
 
             $statistic->reset($currentConnectionCount);
         }