Update with pub sub replication and redis driver

Author: snellingio

PubSub/PubSubInterface.php

@@ -0,0 +1,12 @@
+<?php
+
+namespace BeyondCode\LaravelWebSockets\PubSub;
+
+use React\EventLoop\LoopInterface;
+
+interface PubSubInterface
+{
+    public function publish(string $appId, array $payload): bool;
+
+    public function subscribe(LoopInterface $loop): PubSubInterface;
+}

PubSub/Redis/RedisClient.php

@@ -0,0 +1,118 @@
+<?php
+
+namespace BeyondCode\LaravelWebSockets\PubSub\Redis;
+
+use BeyondCode\LaravelWebSockets\PubSub\PubSubInterface;
+use BeyondCode\LaravelWebSockets\WebSockets\Channels\ChannelManager;
+use Clue\React\Block;
+use Clue\React\Redis\Client;
+use Clue\React\Redis\Factory;
+use Illuminate\Support\Str;
+use React\EventLoop\LoopInterface;
+use React\Promise\PromiseInterface;
+
+class RedisClient implements PubSubInterface
+{
+
+    const REDIS_KEY = ':websockets:replication:';
+    protected $apps;
+    protected $loop;
+    protected $serverId;
+    protected $publishClient;
+    protected $subscribeClient;
+
+    public function __construct()
+    {
+        $this->apps     = collect(config('websockets.apps'));
+        $this->serverId = Str::uuid()->toString();
+    }
+
+    public function publish(string $appId, array $payload): bool
+    {
+        $payload['appId']    = $appId;
+        $payload['serverId'] = $this->serverId;
+        $this->publishClient->publish(self::REDIS_KEY, json_encode($payload));
+        return true;
+    }
+
+    public function subscribe(LoopInterface $loop): PubSubInterface
+    {
+        $this->loop = $loop;
+        [$this->publishClient, $this->subscribeClient] = Block\awaitAll([$this->publishConnection(), $this->subscribeConnection()], $this->loop);
+        return $this->publishClient;
+    }
+
+    protected function publishConnection(): PromiseInterface
+    {
+        $connectionUri = $this->getConnectionUri();
+        $factory       = new Factory($this->loop);
+        return $factory->createClient($connectionUri)->then(
+            function (Client $client) {
+                $this->publishClient = $client;
+                return $this;
+            }
+        );
+    }
+
+
+    protected function subscribeConnection(): PromiseInterface
+    {
+        $connectionUri = $this->getConnectionUri();
+        $factory       = new Factory($this->loop);
+        return $factory->createClient($connectionUri)->then(
+            function (Client $client) {
+                $this->subscribeClient = $client;
+                $this->onConnected();
+                return $this;
+            }
+        );
+    }
+
+    protected function getConnectionUri()
+    {
+        $name   = config('websockets.replication.connection') ?? 'default';
+        $config = config('database.redis.' . $name);
+        $host   = $config['host'];
+        $port   = $config['port'] ? (':' . $config['port']) : ':6379';
+
+        $query = [];
+        if ($config['password']) {
+            $query['password'] = $config['password'];
+        }
+        if ($config['database']) {
+            $query['database'] = $config['database'];
+        }
+        $query = http_build_query($query);
+
+        return "redis://$host$port" . ($query ? '?' . $query : '');
+    }
+
+    protected function onConnected()
+    {
+        $this->subscribeClient->subscribe(self::REDIS_KEY);
+        $this->subscribeClient->on('message', function ($channel, $payload) {
+            $this->onMessage($channel, $payload);
+        });
+    }
+
+    protected function onMessage($channel, $payload)
+    {
+        $payload = json_decode($payload);
+
+        if ($this->serverId === $payload->serverId) {
+            return false;
+        }
+
+        /* @var $channelManager ChannelManager */
+        $channelManager = app(ChannelManager::class);
+        $channelSearch        = $channelManager->find($payload->appId, $payload->channel);
+
+        if ($channelSearch === null) {
+            return false;
+        }
+
+        $channel->broadcast($payload);
+        return true;
+    }
+
+}

composer.json

@@ -25,7 +25,9 @@
         "php": "^7.1",
         "ext-json": "*",
         "cboden/ratchet": "^0.4.1",
+        "clue/block-react": "^1.3",
         "clue/buzz-react": "^2.5",
+        "clue/redis-react": "^2.2",
         "facade/ignition-contracts": "^1.0",
         "guzzlehttp/psr7": "^1.5",
         "illuminate/broadcasting": "5.7.* || 5.8.* || ^6.0",

config/websockets.php

@@ -124,6 +124,20 @@
         'passphrase' => env('LARAVEL_WEBSOCKETS_SSL_PASSPHRASE', null),
     ],
 
+    /*
+     * You can enable replication to publish and subscribe to messages across the driver
+     */
+    'replication' => [
+        'enabled' => false,
+
+        'driver' => 'redis',
+
+        'redis' => [
+            'connection' => 'default',
+        ],
+    ],
+
+
     /*
      * Channel Manager
      * This class handles how channel persistence is handled.

src/Console/StartWebSocketServer.php

@@ -10,6 +10,8 @@
 use React\EventLoop\Factory as LoopFactory;
 use React\Dns\Resolver\Factory as DnsFactory;
 use BeyondCode\LaravelWebSockets\Statistics\DnsResolver;
+use BeyondCode\LaravelWebSockets\PubSub\PubSubInterface;
+use BeyondCode\LaravelWebSockets\PubSub\Redis\RedisClient;
 use BeyondCode\LaravelWebSockets\Facades\StatisticsLogger;
 use BeyondCode\LaravelWebSockets\Facades\WebSocketsRouter;
 use BeyondCode\LaravelWebSockets\Server\Logger\HttpLogger;
@@ -45,6 +47,7 @@ public function handle()
             ->configureConnectionLogger()
             ->registerEchoRoutes()
             ->registerCustomRoutes()
+            ->configurePubSubReplication()
             ->startWebSocketServer();
     }
 
@@ -135,6 +138,23 @@ protected function startWebSocketServer()
             ->run();
     }
 
+    protected function configurePubSubReplication()
+    {
+        if (config('websockets.replication.enabled') !== true) {
+            return $this;
+        }
+
+        if (config('websockets.replication.driver') === 'redis') {
+            $connection = (new RedisClient())->subscribe($this->loop);
+        }
+
+        app()->singleton(PubSubInterface::class, function () use ($connection) {
+            return $connection;
+        });
+
+        return $this;
+    }
+
     protected function getDnsResolver(): ResolverInterface
     {
         if (! config('websockets.statistics.perform_dns_lookup')) {

src/HttpApi/Controllers/TriggerEventController.php

@@ -19,7 +19,7 @@ public function __invoke(Request $request)
                 'channel' => $channelName,
                 'event' => $request->json()->get('name'),
                 'data' => $request->json()->get('data'),
-            ], $request->json()->get('socket_id'));
+            ], $request->json()->get('socket_id'), $request->appId);
 
             DashboardLogger::apiMessage(
                 $request->appId,

src/WebSockets/Channels/Channel.php

@@ -5,6 +5,7 @@
 use stdClass;
 use Illuminate\Support\Str;
 use Ratchet\ConnectionInterface;
+use BeyondCode\LaravelWebSockets\PubSub\PubSubInterface;
 use BeyondCode\LaravelWebSockets\Dashboard\DashboardLogger;
 use BeyondCode\LaravelWebSockets\WebSockets\Exceptions\InvalidSignature;
 
@@ -88,11 +89,15 @@ public function broadcast($payload)
 
     public function broadcastToOthers(ConnectionInterface $connection, $payload)
     {
-        $this->broadcastToEveryoneExcept($payload, $connection->socketId);
+        $this->broadcastToEveryoneExcept($payload, $connection->socketId, $connection->app->id);
     }
 
-    public function broadcastToEveryoneExcept($payload, ?string $socketId = null)
+    public function broadcastToEveryoneExcept($payload, ?string $socketId = null, ?string $appId = null)
     {
+        if (config('websockets.replication.enabled') === true) {
+            app()->get(PubSubInterface::class)->publish($appId, $payload);
+        }
+
         if (is_null($socketId)) {
             return $this->broadcast($payload);
         }