Merge pull request #140 from francislavoie/redis-replication

Redis as a replication backend for scalability

Author: rennokki

.gitignore

@@ -2,4 +2,5 @@ build
 composer.lock
 vendor
 coverage
-.phpunit.result.cache
+.phpunit.result.cache
+.idea/

composer.json

@@ -26,6 +26,7 @@
         "ext-json": "*",
         "cboden/ratchet": "^0.4.1",
         "clue/buzz-react": "^2.5",
+        "clue/redis-react": "^2.3",
         "facade/ignition-contracts": "^1.0",
         "guzzlehttp/psr7": "^1.5",
         "illuminate/broadcasting": "5.8.*|^6.0|^7.0",

config/websockets.php

@@ -141,6 +141,34 @@
 
     ],
 
+    /*
+    |--------------------------------------------------------------------------
+    | Broadcasting Replication
+    |--------------------------------------------------------------------------
+    |
+    | You can enable replication to publish and subscribe to
+    | messages across the driver.
+    |
+    | By default, it is disabled, but you can configure it to use drivers
+    | like Redis to ensure connection between multiple instances of
+    | WebSocket servers.
+    |
+    */
+
+    'replication' => [
+
+        'enabled' => false,
+
+        'driver' => 'redis',
+
+        'redis' => [
+
+            'connection' => 'default',
+
+        ],
+
+    ],
+
     'statistics' => [
 
         /*

src/Apps/ConfigAppProvider.php

@@ -19,7 +19,7 @@ public function all(): array
     {
         return $this->apps
             ->map(function (array $appAttributes) {
-                return $this->instanciate($appAttributes);
+                return $this->instantiate($appAttributes);
             })
             ->toArray();
     }
@@ -30,7 +30,7 @@ public function findById($appId): ?App
             ->apps
             ->firstWhere('id', $appId);
 
-        return $this->instanciate($appAttributes);
+        return $this->instantiate($appAttributes);
     }
 
     public function findByKey(string $appKey): ?App
@@ -39,7 +39,7 @@ public function findByKey(string $appKey): ?App
             ->apps
             ->firstWhere('key', $appKey);
 
-        return $this->instanciate($appAttributes);
+        return $this->instantiate($appAttributes);
     }
 
     public function findBySecret(string $appSecret): ?App
@@ -48,10 +48,10 @@ public function findBySecret(string $appSecret): ?App
             ->apps
             ->firstWhere('secret', $appSecret);
 
-        return $this->instanciate($appAttributes);
+        return $this->instantiate($appAttributes);
     }
 
-    protected function instanciate(?array $appAttributes): ?App
+    protected function instantiate(?array $appAttributes): ?App
     {
         if (! $appAttributes) {
             return null;

src/Console/StartWebSocketServer.php

@@ -4,6 +4,7 @@
 
 use BeyondCode\LaravelWebSockets\Facades\StatisticsLogger;
 use BeyondCode\LaravelWebSockets\Facades\WebSocketsRouter;
+use BeyondCode\LaravelWebSockets\PubSub\ReplicationInterface;
 use BeyondCode\LaravelWebSockets\Server\Logger\ConnectionLogger;
 use BeyondCode\LaravelWebSockets\Server\Logger\HttpLogger;
 use BeyondCode\LaravelWebSockets\Server\Logger\WebsocketsLogger;
@@ -49,6 +50,7 @@ public function handle()
             ->configureRestartTimer()
             ->registerEchoRoutes()
             ->registerCustomRoutes()
+            ->configurePubSubReplication()
             ->startWebSocketServer();
     }
 
@@ -61,7 +63,7 @@ protected function configureStatisticsLogger()
 
         $browser = new Browser($this->loop, $connector);
 
-        app()->singleton(StatisticsLoggerInterface::class, function () use ($browser) {
+        $this->laravel->singleton(StatisticsLoggerInterface::class, function () use ($browser) {
             $class = config('websockets.statistics.logger', \BeyondCode\LaravelWebSockets\Statistics\Logger::class);
 
             return new $class(app(ChannelManager::class), $browser);
@@ -76,7 +78,7 @@ protected function configureStatisticsLogger()
 
     protected function configureHttpLogger()
     {
-        app()->singleton(HttpLogger::class, function () {
+        $this->laravel->singleton(HttpLogger::class, function () {
             return (new HttpLogger($this->output))
                 ->enable($this->option('debug') ?: config('app.debug'))
                 ->verbose($this->output->isVerbose());
@@ -87,7 +89,7 @@ protected function configureHttpLogger()
 
     protected function configureMessageLogger()
     {
-        app()->singleton(WebsocketsLogger::class, function () {
+        $this->laravel->singleton(WebsocketsLogger::class, function () {
             return (new WebsocketsLogger($this->output))
                 ->enable($this->option('debug') ?: config('app.debug'))
                 ->verbose($this->output->isVerbose());
@@ -98,7 +100,7 @@ protected function configureMessageLogger()
 
     protected function configureConnectionLogger()
     {
-        app()->bind(ConnectionLogger::class, function () {
+        $this->laravel->bind(ConnectionLogger::class, function () {
             return (new ConnectionLogger($this->output))
                 ->enable(config('app.debug'))
                 ->verbose($this->output->isVerbose());
@@ -151,6 +153,13 @@ protected function startWebSocketServer()
             ->run();
     }
 
+    protected function configurePubSubReplication()
+    {
+        $this->laravel->get(ReplicationInterface::class)->boot($this->loop);
+
+        return $this;
+    }
+
     protected function getDnsResolver(): ResolverInterface
     {
         if (! config('websockets.statistics.perform_dns_lookup')) {

src/Facades/StatisticsLogger.php

@@ -5,7 +5,10 @@
 use BeyondCode\LaravelWebSockets\Statistics\Logger\StatisticsLogger as StatisticsLoggerInterface;
 use Illuminate\Support\Facades\Facade;
 
-/** @see \BeyondCode\LaravelWebSockets\Statistics\Logger\HttpStatisticsLogger */
+/**
+ * @see \BeyondCode\LaravelWebSockets\Statistics\Logger\HttpStatisticsLogger
+ * @mixin \BeyondCode\LaravelWebSockets\Statistics\Logger\HttpStatisticsLogger
+ */
 class StatisticsLogger extends Facade
 {
     protected static function getFacadeAccessor()

src/Facades/WebSocketsRouter.php

@@ -4,7 +4,10 @@
 
 use Illuminate\Support\Facades\Facade;
 
-/** @see \BeyondCode\LaravelWebSockets\Server\Router */
+/**
+ * @see \BeyondCode\LaravelWebSockets\Server\Router
+ * @mixin \BeyondCode\LaravelWebSockets\Server\Router
+ */
 class WebSocketsRouter extends Facade
 {
     protected static function getFacadeAccessor()

src/HttpApi/Controllers/Controller.php

@@ -16,6 +16,7 @@
 use Pusher\Pusher;
 use Ratchet\ConnectionInterface;
 use Ratchet\Http\HttpServerInterface;
+use React\Promise\PromiseInterface;
 use Symfony\Bridge\PsrHttpMessage\Factory\HttpFoundationFactory;
 use Symfony\Component\HttpKernel\Exception\HttpException;
 
@@ -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)
@@ -46,7 +47,11 @@ public function onOpen(ConnectionInterface $connection, RequestInterface $reques
 
         $this->requestBuffer = (string) $request->getBody();
 
-        $this->checkContentLength($connection);
+        if (! $this->verifyContentLength()) {
+            return;
+        }
+
+        $this->handleRequest($connection);
     }
 
     protected function findContentLength(array $headers): int
@@ -60,31 +65,53 @@ public function onMessage(ConnectionInterface $from, $msg)
     {
         $this->requestBuffer .= $msg;
 
-        $this->checkContentLength($from);
+        if (! $this->verifyContentLength()) {
+            return;
+        }
+
+        $this->handleRequest($from);
+    }
+
+    protected function verifyContentLength()
+    {
+        return strlen($this->requestBuffer) === $this->contentLength;
     }
 
-    protected function checkContentLength(ConnectionInterface $connection)
+    protected function handleRequest(ConnectionInterface $connection)
     {
-        if (strlen($this->requestBuffer) === $this->contentLength) {
-            $serverRequest = (new ServerRequest(
-                $this->request->getMethod(),
-                $this->request->getUri(),
-                $this->request->getHeaders(),
-                $this->requestBuffer,
-                $this->request->getProtocolVersion()
-            ))->withQueryParams(QueryParameters::create($this->request)->all());
+        $serverRequest = (new ServerRequest(
+            $this->request->getMethod(),
+            $this->request->getUri(),
+            $this->request->getHeaders(),
+            $this->requestBuffer,
+            $this->request->getProtocolVersion()
+        ))->withQueryParams(QueryParameters::create($this->request)->all());
+
+        $laravelRequest = Request::createFromBase((new HttpFoundationFactory)->createRequest($serverRequest));
 
-            $laravelRequest = Request::createFromBase((new HttpFoundationFactory)->createRequest($serverRequest));
+        $this
+            ->ensureValidAppId($laravelRequest->appId)
+            ->ensureValidSignature($laravelRequest);
 
-            $this
-                ->ensureValidAppId($laravelRequest->appId)
-                ->ensureValidSignature($laravelRequest);
+        // Invoke the controller action
+        $response = $this($laravelRequest);
 
-            $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);
+            });
 
-            $connection->send(JsonResponse::create($response));
-            $connection->close();
+            return;
         }
+
+        $this->sendAndClose($connection, $response);
+    }
+
+    protected function sendAndClose(ConnectionInterface $connection, $response)
+    {
+        $connection->send(JsonResponse::create($response));
+        $connection->close();
     }
 
     public function onClose(ConnectionInterface $connection)
@@ -122,7 +149,7 @@ protected function ensureValidSignature(Request $request)
         /*
          * The `auth_signature` & `body_md5` parameters are not included when calculating the `auth_signature` value.
          *
-         * The `appId`, `appKey` & `channelName` parameters are actually route paramaters and are never supplied by the client.
+         * The `appId`, `appKey` & `channelName` parameters are actually route parameters and are never supplied by the client.
          */
         $params = Arr::except($request->query(), ['auth_signature', 'body_md5', 'appId', 'appKey', 'channelName']);
 

src/HttpApi/Controllers/FetchChannelController.php

@@ -15,6 +15,6 @@ public function __invoke(Request $request)
             throw new HttpException(404, "Unknown channel `{$request->channelName}`.");
         }
 
-        return $channel->toArray();
+        return $channel->toArray($request->appId);
     }
 }

src/HttpApi/Controllers/FetchChannelsController.php

@@ -6,9 +6,22 @@
 use Illuminate\Support\Collection;
 use Illuminate\Support\Str;
 use Symfony\Component\HttpKernel\Exception\HttpException;
+use BeyondCode\LaravelWebSockets\PubSub\ReplicationInterface;
+use BeyondCode\LaravelWebSockets\WebSockets\Channels\ChannelManager;
+use BeyondCode\LaravelWebSockets\WebSockets\Channels\PresenceChannel;
 
 class FetchChannelsController extends Controller
 {
+    /** @var ReplicationInterface */
+    protected $replication;
+
+    public function __construct(ChannelManager $channelManager, ReplicationInterface $replication)
+    {
+        parent::__construct($channelManager);
+
+        $this->replication = $replication;
+    }
+
     public function __invoke(Request $request)
     {
         $attributes = [];
@@ -29,15 +42,28 @@ public function __invoke(Request $request)
             });
         }
 
-        return [
-            'channels' => $channels->map(function ($channel) use ($attributes) {
-                $info = new \stdClass;
-                if (in_array('user_count', $attributes)) {
-                    $info->user_count = count($channel->getUsers());
-                }
+        // 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();
+
+        // We ask the replication backend to get us the member count per channel.
+        // We get $counts back as a key-value array of channel names and their member count.
+        return $this->replication
+            ->channelMemberCounts($request->appId, $channelNames)
+            ->then(function (array $counts) use ($channels, $attributes) {
+                return [
+                    'channels' => $channels->map(function (PresenceChannel $channel) use ($counts, $attributes) {
+                        $info = new \stdClass;
+                        if (in_array('user_count', $attributes)) {
+                            $info->user_count = $counts[$channel->getChannelName()];
+                        }
 
-                return $info;
-            })->toArray() ?: new \stdClass,
-        ];
+                        return $info;
+                    })->toArray() ?: new \stdClass,
+                ];
+            });
     }
 }