Implement presence channel storage in Redis

Author: francislavoie

src/HttpApi/Controllers/Controller.php

@@ -11,6 +11,7 @@
 use Illuminate\Http\JsonResponse;
 use GuzzleHttp\Psr7\ServerRequest;
 use Illuminate\Support\Collection;
+use React\Promise\PromiseInterface;
 use Ratchet\Http\HttpServerInterface;
 use Psr\Http\Message\RequestInterface;
 use BeyondCode\LaravelWebSockets\Apps\App;
@@ -30,7 +31,7 @@ abstract class Controller implements HttpServerInterface
     /** @var int */
     protected $contentLength;
 
-    /** @var \BeyondCode\LaravelWebSockets\WebSockets\Channels\ChannelManager */
+    /** @var ChannelManager */
     protected $channelManager;
 
     public function __construct(ChannelManager $channelManager)
@@ -92,8 +93,23 @@ protected function handleRequest(ConnectionInterface $connection)
             ->ensureValidAppId($laravelRequest->appId)
             ->ensureValidSignature($laravelRequest);
 
+        // Invoke the controller action
         $response = $this($laravelRequest);
 
+        // Allow for async IO in the controller action
+        if ($response instanceof PromiseInterface) {
+            $response->then(function($response) use ($connection) {
+                $this->sendAndClose($connection, $response);
+            });
+
+            return;
+        }
+
+        $this->sendAndClose($connection, $response);
+    }
+
+    protected function sendAndClose(ConnectionInterface $connection, $response)
+    {
         $connection->send(JsonResponse::create($response));
         $connection->close();
     }

src/HttpApi/Controllers/FetchChannelsController.php

@@ -5,6 +5,8 @@
 use Illuminate\Support\Str;
 use Illuminate\Http\Request;
 use Illuminate\Support\Collection;
+use React\Promise\PromiseInterface;
+use BeyondCode\LaravelWebSockets\PubSub\ReplicationInterface;
 use BeyondCode\LaravelWebSockets\WebSockets\Channels\PresenceChannel;
 
 class FetchChannelsController extends Controller
@@ -21,10 +23,39 @@ public function __invoke(Request $request)
             });
         }
 
+        if (config('websockets.replication.enabled') === true) {
+            // We want to get the channel user count all in one shot when
+            // using a replication backend rather than doing individual queries.
+            // To do so, we first collect the list of channel names.
+            $channelNames = $channels->map(function (PresenceChannel $channel) use ($request) {
+                return $channel->getChannelName();
+            })->toArray();
+
+            /** @var PromiseInterface $memberCounts */
+            // We ask the replication backend to get us the member count per channel
+            $memberCounts = app(ReplicationInterface::class)
+                ->channelMemberCounts($request->appId, $channelNames);
+
+            // We return a promise since the backend runs async. We get $counts back
+            // as a key-value array of channel names and their member count.
+            return $memberCounts->then(function (array $counts) use ($channels) {
+                return $this->collectUserCounts($channels, function(PresenceChannel $channel) use ($counts) {
+                    return $counts[$channel->getChannelName()];
+                });
+            });
+        }
+
+        return $this->collectUserCounts($channels, function(PresenceChannel $channel) {
+            return $channel->getUserCount();
+        });
+    }
+
+    protected function collectUserCounts(Collection $channels, callable $transformer)
+    {
         return [
-            'channels' => $channels->map(function ($channel) {
+            'channels' => $channels->map(function (PresenceChannel $channel) use ($transformer) {
                 return [
-                    'user_count' => count($channel->getUsers()),
+                    'user_count' => $transformer($channel),
                 ];
             })->toArray() ?: new \stdClass,
         ];

src/HttpApi/Controllers/FetchUsersController.php

@@ -4,6 +4,7 @@
 
 use Illuminate\Http\Request;
 use Illuminate\Support\Collection;
+use React\Promise\PromiseInterface;
 use Symfony\Component\HttpKernel\Exception\HttpException;
 use BeyondCode\LaravelWebSockets\WebSockets\Channels\PresenceChannel;
 
@@ -21,8 +22,21 @@ public function __invoke(Request $request)
             throw new HttpException(400, 'Invalid presence channel "'.$request->channelName.'"');
         }
 
+        $users = $channel->getUsers($request->appId);
+
+        if ($users instanceof PromiseInterface) {
+            return $users->then(function(array $users) {
+                return $this->collectUsers($users);
+            });
+        }
+
+        return $this->collectUsers($users);
+    }
+
+    protected function collectUsers(array $users)
+    {
         return [
-            'users' => Collection::make($channel->getUsers())->map(function ($user) {
+            'users' => Collection::make($users)->map(function ($user) {
                 return ['id' => $user->user_id];
             })->values(),
         ];

src/PubSub/Redis/RedisClient.php

@@ -7,6 +7,7 @@
 use Clue\React\Redis\Client;
 use Clue\React\Redis\Factory;
 use React\EventLoop\LoopInterface;
+use React\Promise\PromiseInterface;
 use BeyondCode\LaravelWebSockets\PubSub\ReplicationInterface;
 use BeyondCode\LaravelWebSockets\WebSockets\Channels\ChannelManager;
 
@@ -183,6 +184,72 @@ public function publish(string $appId, string $channel, stdClass $payload): bool
         return true;
     }
 
+    /**
+     * Add a member to a channel. To be called when they have
+     * subscribed to the channel.
+     *
+     * @param string $appId
+     * @param string $channel
+     * @param string $socketId
+     * @param string $data
+     */
+    public function joinChannel(string $appId, string $channel, string $socketId, string $data)
+    {
+        $this->publishClient->__call('hset', ["$appId:$channel", $socketId, $data]);
+    }
+
+    /**
+     * Remove a member from the channel. To be called when they have
+     * unsubscribed from the channel.
+     *
+     * @param string $appId
+     * @param string $channel
+     * @param string $socketId
+     */
+    public function leaveChannel(string $appId, string $channel, string $socketId)
+    {
+        $this->publishClient->__call('hdel', ["$appId:$channel", $socketId]);
+    }
+
+    /**
+     * Retrieve the full information about the members in a presence channel.
+     *
+     * @param string $appId
+     * @param string $channel
+     * @return PromiseInterface
+     */
+    public function channelMembers(string $appId, string $channel): PromiseInterface
+    {
+        return $this->publishClient->__call('hgetall', ["$appId:$channel"])
+            ->then(function($members) {
+                // The data is expected as objects, so we need to JSON decode
+                return array_walk($members, function($user) {
+                    return json_decode($user);
+                });
+            });
+    }
+
+    /**
+     * Get the amount of users subscribed for each presence channel.
+     *
+     * @param string $appId
+     * @param array $channelNames
+     * @return PromiseInterface
+     */
+    public function channelMemberCounts(string $appId, array $channelNames): PromiseInterface
+    {
+        $this->publishClient->__call('multi', []);
+
+        foreach ($channelNames as $channel) {
+            $this->publishClient->__call('hlen', ["$appId:$channel"]);
+        }
+
+        return $this->publishClient->__call('exec', [])
+            ->then(function($data) use ($channelNames) {
+                return array_combine($channelNames, $data);
+            });
+    }
+
     /**
      * Build the Redis connection URL from Laravel database config.
      *

src/PubSub/ReplicationInterface.php

@@ -4,6 +4,7 @@
 
 use stdClass;
 use React\EventLoop\LoopInterface;
+use React\Promise\PromiseInterface;
 
 interface ReplicationInterface
 {
@@ -40,4 +41,43 @@ public function subscribe(string $appId, string $channel): bool;
      * @return bool
      */
     public function unsubscribe(string $appId, string $channel): bool;
+
+    /**
+     * Add a member to a channel. To be called when they have
+     * subscribed to the channel.
+     *
+     * @param string $appId
+     * @param string $channel
+     * @param string $socketId
+     * @param string $data
+     */
+    public function joinChannel(string $appId, string $channel, string $socketId, string $data);
+
+    /**
+     * Remove a member from the channel. To be called when they have
+     * unsubscribed from the channel.
+     *
+     * @param string $appId
+     * @param string $channel
+     * @param string $socketId
+     */
+    public function leaveChannel(string $appId, string $channel, string $socketId);
+
+    /**
+     * Retrieve the full information about the members in a presence channel.
+     *
+     * @param string $appId
+     * @param string $channel
+     * @return PromiseInterface
+     */
+    public function channelMembers(string $appId, string $channel): PromiseInterface;
+
+    /**
+     * Get the amount of users subscribed for each presence channel.
+     *
+     * @param string $appId
+     * @param array $channelNames
+     * @return PromiseInterface
+     */
+    public function channelMemberCounts(string $appId, array $channelNames): PromiseInterface;
 }

src/WebSockets/Channels/Channel.php

@@ -22,6 +22,11 @@ public function __construct(string $channelName)
         $this->channelName = $channelName;
     }
 
+    public function getChannelName(): string
+    {
+        return $this->channelName;
+    }
+
     public function hasConnections(): bool
     {
         return count($this->subscribedConnections) > 0;
@@ -32,6 +37,9 @@ public function getSubscribedConnections(): array
         return $this->subscribedConnections;
     }
 
+    /**
+     * @throws InvalidSignature
+     */
     protected function verifySignature(ConnectionInterface $connection, stdClass $payload)
     {
         $signature = "{$connection->socketId}:{$this->channelName}";
@@ -40,12 +48,15 @@ protected function verifySignature(ConnectionInterface $connection, stdClass $pa
             $signature .= ":{$payload->channel_data}";
         }
 
-        if (Str::after($payload->auth, ':') !== hash_hmac('sha256', $signature, $connection->app->secret)) {
+        if (! hash_equals(
+            hash_hmac('sha256', $signature, $connection->app->secret),
+            Str::after($payload->auth, ':'))
+        ) {
             throw new InvalidSignature();
         }
     }
 
-    /*
+    /**
      * @link https://pusher.com/docs/pusher_protocol#presence-channel-events
      */
     public function subscribe(ConnectionInterface $connection, stdClass $payload)
@@ -128,7 +139,7 @@ public function broadcastToEveryoneExcept($payload, ?string $socketId = null)
         }
     }
 
-    public function toArray(): array
+    public function toArray()
     {
         return [
             'occupied' => count($this->subscribedConnections) > 0,