cleanup & refactor of pubsub code

Author: anthonyvancauwenberghe

composer.json

@@ -42,7 +42,8 @@
     "require-dev": {
         "mockery/mockery": "^1.2",
         "orchestra/testbench": "3.7.* || 3.8.* || ^4.0",
-        "phpunit/phpunit": "^7.0 || ^8.0"
+        "phpunit/phpunit": "^7.0 || ^8.0",
+        "predis/predis": "^1.1"
     },
     "autoload": {
         "psr-4": {

phpunit.xml.dist

@@ -27,5 +27,6 @@
     </logging>
     <php>
         <env name="DB_CONNECTION" value="testing"/>
+        <env name="REDIS_HOST" value="redis"/>
     </php>
 </phpunit>

src/Console/StartWebSocketServer.php

@@ -12,7 +12,6 @@
 use BeyondCode\LaravelWebSockets\Statistics\DnsResolver;
 use BeyondCode\LaravelWebSockets\Facades\StatisticsLogger;
 use BeyondCode\LaravelWebSockets\Facades\WebSocketsRouter;
-use BeyondCode\LaravelWebSockets\PubSub\Redis\RedisClient;
 use BeyondCode\LaravelWebSockets\Server\Logger\HttpLogger;
 use BeyondCode\LaravelWebSockets\PubSub\ReplicationInterface;
 use BeyondCode\LaravelWebSockets\Server\WebSocketServerFactory;
@@ -117,7 +116,6 @@ protected function registerEchoRoutes()
     protected function registerCustomRoutes()
     {
         WebSocketsRouter::customRoutes();
-
         return $this;
     }
 
@@ -140,15 +138,7 @@ protected function startWebSocketServer()
 
     protected function configurePubSubReplication()
     {
-        if (config('websockets.replication.enabled') !== true) {
-            return $this;
-        }
-
-        if (config('websockets.replication.driver') === 'redis') {
-            $this->laravel->singleton(ReplicationInterface::class, function () {
-                return (new RedisClient())->boot($this->loop);
-            });
-        }
+        app(ReplicationInterface::class)->boot($this->loop);
 
         return $this;
     }

src/PubSub/Redis/RedisPusherBroadcaster.php renamed to src/PubSub/Broadcasters/RedisPusherBroadcaster.php

@@ -1,6 +1,6 @@
 <?php
 
-namespace BeyondCode\LaravelWebSockets\PubSub\Redis;
+namespace BeyondCode\LaravelWebSockets\PubSub\Broadcasters;
 
 use Pusher\Pusher;
 use Illuminate\Support\Arr;

src/PubSub/Drivers/EmptyClient.php

@@ -0,0 +1,112 @@
+<?php
+
+namespace BeyondCode\LaravelWebSockets\PubSub\Drivers;
+
+use stdClass;
+use React\EventLoop\LoopInterface;
+use React\Promise\FulfilledPromise;
+use React\Promise\PromiseInterface;
+use BeyondCode\LaravelWebSockets\PubSub\ReplicationInterface;
+
+class EmptyClient implements ReplicationInterface
+{
+
+    /**
+     * Boot the pub/sub provider (open connections, initial subscriptions, etc).
+     *
+     * @param LoopInterface $loop
+     * @return self
+     */
+    public function boot(LoopInterface $loop) : ReplicationInterface
+    {
+        return $this;
+    }
+
+    /**
+     * Publish a payload on a specific channel, for a specific app.
+     *
+     * @param string $appId
+     * @param string $channel
+     * @param stdClass $payload
+     * @return bool
+     */
+    public function publish(string $appId, string $channel, stdClass $payload) : bool
+    {
+        return true;
+    }
+
+    /**
+     * Subscribe to receive messages for a channel.
+     *
+     * @param string $appId
+     * @param string $channel
+     * @return bool
+     */
+    public function subscribe(string $appId, string $channel) : bool
+    {
+        return true;
+    }
+
+    /**
+     * Unsubscribe from a channel.
+     *
+     * @param string $appId
+     * @param string $channel
+     * @return bool
+     */
+    public function unsubscribe(string $appId, string $channel) : 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)
+    {
+    
+    }
+
+    /**
+     * 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
+    {
+        return new FulfilledPromise(null);
+    }
+
+    /**
+     * 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
+    {
+        return new FulfilledPromise(null);
+    }
+}

src/PubSub/Redis/RedisClient.php renamed to src/PubSub/Drivers/RedisClient.php

@@ -1,6 +1,6 @@
 <?php
 
-namespace BeyondCode\LaravelWebSockets\PubSub\Redis;
+namespace BeyondCode\LaravelWebSockets\PubSub\Drivers;
 
 use stdClass;
 use Illuminate\Support\Str;
@@ -129,6 +129,7 @@ protected function onMessage(string $redisChannel, string $payload)
      */
     public function subscribe(string $appId, string $channel): bool
     {
+        $this->publishClient->__call('hset', ["$appId:$channel", 541561516, "qsgdqgsd"]);
         if (! isset($this->subscribedChannels["$appId:$channel"])) {
             // We're not subscribed to the channel yet, subscribe and set the count to 1
             $this->subscribeClient->__call('subscribe', ["$appId:$channel"]);

src/WebSockets/Channels/Channel.php

@@ -14,12 +14,18 @@ class Channel
     /** @var string */
     protected $channelName;
 
+    /**
+     * @var ReplicationInterface
+     */
+    protected $pubSub;
+
     /** @var \Ratchet\ConnectionInterface[] */
     protected $subscribedConnections = [];
 
     public function __construct(string $channelName)
     {
         $this->channelName = $channelName;
+        $this->pubSub = app(ReplicationInterface::class);
     }
 
     public function getChannelName(): string
@@ -48,7 +54,7 @@ protected function verifySignature(ConnectionInterface $connection, stdClass $pa
             $signature .= ":{$payload->channel_data}";
         }
 
-        if (! hash_equals(
+        if (!hash_equals(
             hash_hmac('sha256', $signature, $connection->app->secret),
             Str::after($payload->auth, ':'))
         ) {
@@ -63,11 +69,8 @@ public function subscribe(ConnectionInterface $connection, stdClass $payload)
     {
         $this->saveConnection($connection);
 
-        if (config('websockets.replication.enabled') === true) {
-            // Subscribe for broadcasted messages from the pub/sub backend
-            app(ReplicationInterface::class)
-                ->subscribe($connection->app->id, $this->channelName);
-        }
+        // Subscribe to broadcasted messages from the pub/sub backend
+        $this->pubSub->subscribe($connection->app->id, $this->channelName);
 
         $connection->send(json_encode([
             'event' => 'pusher_internal:subscription_succeeded',
@@ -79,13 +82,10 @@ public function unsubscribe(ConnectionInterface $connection)
     {
         unset($this->subscribedConnections[$connection->socketId]);
 
-        if (config('websockets.replication.enabled') === true) {
-            // Unsubscribe from the pub/sub backend
-            app(ReplicationInterface::class)
-                ->unsubscribe($connection->app->id, $this->channelName);
-        }
+        // Unsubscribe from the pub/sub backend
+        $this->pubSub->unsubscribe($connection->app->id, $this->channelName);
 
-        if (! $this->hasConnections()) {
+        if (!$this->hasConnections()) {
             DashboardLogger::vacated($connection, $this->channelName);
         }
     }
@@ -96,7 +96,7 @@ protected function saveConnection(ConnectionInterface $connection)
 
         $this->subscribedConnections[$connection->socketId] = $connection;
 
-        if (! $hadConnectionsPreviously) {
+        if (!$hadConnectionsPreviously) {
             DashboardLogger::occupied($connection, $this->channelName);
         }
 
@@ -112,11 +112,8 @@ public function broadcast($payload)
 
     public function broadcastToOthers(ConnectionInterface $connection, $payload)
     {
-        if (config('websockets.replication.enabled') === true) {
-            // Also broadcast via the other websocket servers
-            app(ReplicationInterface::class)
-                ->publish($connection->app->id, $this->channelName, $payload);
-        }
+        // Also broadcast via the other websocket servers
+        $this->pubSub->publish($connection->app->id, $this->channelName, $payload);
 
         $this->broadcastToEveryoneExcept($payload, $connection->socketId);
     }

src/WebSocketsServiceProvider.php

@@ -2,6 +2,10 @@
 
 namespace BeyondCode\LaravelWebSockets;
 
+use BeyondCode\LaravelWebSockets\PubSub\Broadcasters\RedisPusherBroadcaster;
+use BeyondCode\LaravelWebSockets\PubSub\Drivers\EmptyClient;
+use BeyondCode\LaravelWebSockets\PubSub\Drivers\RedisClient;
+use BeyondCode\LaravelWebSockets\PubSub\ReplicationInterface;
 use Pusher\Pusher;
 use Psr\Log\LoggerInterface;
 use Illuminate\Support\Facades\Gate;
@@ -11,7 +15,6 @@
 use BeyondCode\LaravelWebSockets\Server\Router;
 use BeyondCode\LaravelWebSockets\Apps\AppProvider;
 use BeyondCode\LaravelWebSockets\WebSockets\Channels\ChannelManager;
-use BeyondCode\LaravelWebSockets\PubSub\Redis\RedisPusherBroadcaster;
 use BeyondCode\LaravelWebSockets\Dashboard\Http\Controllers\SendMessage;
 use BeyondCode\LaravelWebSockets\Dashboard\Http\Controllers\ShowDashboard;
 use BeyondCode\LaravelWebSockets\Dashboard\Http\Controllers\AuthenticateDashboard;
@@ -23,30 +26,47 @@
 
 class WebSocketsServiceProvider extends ServiceProvider
 {
-    public function boot(BroadcastManager $broadcastManager)
+    public function boot()
     {
         $this->publishes([
-            __DIR__.'/../config/websockets.php' => base_path('config/websockets.php'),
+            __DIR__ . '/../config/websockets.php' => base_path('config/websockets.php'),
         ], 'config');
 
-        if (! class_exists('CreateWebSocketsStatisticsEntries')) {
+        if (!class_exists('CreateWebSocketsStatisticsEntries')) {
             $this->publishes([
-                __DIR__.'/../database/migrations/create_websockets_statistics_entries_table.php.stub' => database_path('migrations/'.date('Y_m_d_His', time()).'_create_websockets_statistics_entries_table.php'),
+                __DIR__ . '/../database/migrations/create_websockets_statistics_entries_table.php.stub' => database_path('migrations/' . date('Y_m_d_His', time()) . '_create_websockets_statistics_entries_table.php'),
             ], 'migrations');
         }
 
         $this
             ->registerRoutes()
             ->registerDashboardGate();
 
-        $this->loadViewsFrom(__DIR__.'/../resources/views/', 'websockets');
+        $this->loadViewsFrom(__DIR__ . '/../resources/views/', 'websockets');
 
         $this->commands([
             Console\StartWebSocketServer::class,
             Console\CleanStatistics::class,
         ]);
 
-        $broadcastManager->extend('redis-pusher', function ($app, array $config) {
+        $this->configurePubSub();
+
+    }
+
+    protected function configurePubSub()
+    {
+        if (config('websockets.replication.enabled') !== true || config('websockets.replication.driver') !== 'redis') {
+            $this->app->singleton(ReplicationInterface::class, function () {
+                return (new EmptyClient());
+            });
+            return;
+        }
+
+        $this->app->singleton(ReplicationInterface::class, function () {
+            return (new RedisClient())->boot($this->loop);
+        });
+
+        app(BroadcastManager::class)->extend('redis-pusher', function ($app, array $config) {
             $pusher = new Pusher(
                 $config['key'], $config['secret'],
                 $config['app_id'], $config['options'] ?? []
@@ -67,7 +87,7 @@ public function boot(BroadcastManager $broadcastManager)
 
     public function register()
     {
-        $this->mergeConfigFrom(__DIR__.'/../config/websockets.php', 'websockets');
+        $this->mergeConfigFrom(__DIR__ . '/../config/websockets.php', 'websockets');
 
         $this->app->singleton('websockets.router', function () {
             return new Router();
@@ -88,7 +108,7 @@ protected function registerRoutes()
         Route::prefix(config('websockets.path'))->group(function () {
             Route::middleware(config('websockets.middleware', [AuthorizeDashboard::class]))->group(function () {
                 Route::get('/', ShowDashboard::class);
-                Route::get('/api/{appId}/statistics', [DashboardApiController::class,  'getStatistics']);
+                Route::get('/api/{appId}/statistics', [DashboardApiController::class, 'getStatistics']);
                 Route::post('auth', AuthenticateDashboard::class);
                 Route::post('event', SendMessage::class);
             });

src/PubSub/Fake/FakeReplication.php renamed to tests/Mocks/FakeReplicationClient.php

@@ -1,14 +1,14 @@
 <?php
 
-namespace BeyondCode\LaravelWebSockets\PubSub\Fake;
+namespace BeyondCode\LaravelWebSockets\Tests\Mocks;
 
 use stdClass;
 use React\EventLoop\LoopInterface;
 use React\Promise\FulfilledPromise;
 use React\Promise\PromiseInterface;
 use BeyondCode\LaravelWebSockets\PubSub\ReplicationInterface;
 
-class FakeReplication implements ReplicationInterface
+class FakeReplicationClient implements ReplicationInterface
 {
     protected $channels = [];