Merge pull request #453 from beyondcode/refactor/pubsub-replication-t…

…ests

[refactor/tests] Added tests for Redis PubSub replication

.github/workflows/run-tests.yml

@@ -24,6 +24,12 @@ jobs:
             -   name: Checkout code
                 uses: actions/checkout@v1
 
+            -   name: Setup Redis
+                uses: supercharge/[email protected]
+                with:
+                    redis-version: 6
+                if: ${{ matrix.os == 'ubuntu-latest' }}
+
             -   name: Cache dependencies
                 uses: actions/cache@v1
                 with:
@@ -35,16 +41,25 @@ jobs:
                 with:
                     php-version: ${{ matrix.php }}
                     extensions: dom, curl, libxml, mbstring, zip, pcntl, pdo, sqlite, pdo_sqlite, bcmath, soap, intl, gd, exif, iconv, imagick
-                    coverage: pcov
+                    coverage: xdebug
 
             -   name: Install dependencies
                 run: |
                     composer require "laravel/framework:${{ matrix.laravel }}" "orchestra/testbench:${{ matrix.testbench }}" --no-interaction --no-update
                     composer update --${{ matrix.dependency-version }} --prefer-dist --no-interaction --no-suggest
 
-            -   name: Execute tests
-                run: vendor/bin/phpunit --coverage-text --coverage-clover=coverage.xml
+            -   name: Execute tests with Local driver
+                run: vendor/bin/phpunit --coverage-text --coverage-clover=coverage_local.xml
+                env:
+                    REPLICATION_DRIVER: local
+
+            -   name: Execute tests with Redis driver
+                run: vendor/bin/phpunit --coverage-text --coverage-clover=coverage_redis.xml
+                if: ${{ matrix.os == 'ubuntu-latest' }}
+                env:
+                    REPLICATION_DRIVER: redis
 
             -   uses: codecov/codecov-action@v1
                 with:
                     fail_ci_if_error: false
+                    file: '*.xml'

config/websockets.php

@@ -143,23 +143,21 @@
 
     /*
     |--------------------------------------------------------------------------
-    | Broadcasting Replication
+    | Broadcasting Replication PubSub
     |--------------------------------------------------------------------------
     |
     | 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
+
+    | By default, it is set to 'local', but you can configure it to use drivers
     | like Redis to ensure connection between multiple instances of
-    | WebSocket servers.
+    | WebSocket servers. Just set the driver to 'redis' to enable the PubSub using Redis.
     |
     */
 
     'replication' => [
 
-        'enabled' => false,
-
-        'driver' => 'redis',
+        'driver' => 'local',
 
         'redis' => [
 

resources/views/dashboard.blade.php

@@ -70,16 +70,14 @@
                 <thead>
                 <tr>
                     <th>Type</th>
-                    <th>Socket</th>
                     <th>Details</th>
                     <th>Time</th>
                 </tr>
                 </thead>
                 <tbody>
                 <tr v-for="log in logs.slice().reverse()">
                     <td><span class="badge" :class="getBadgeClass(log)">@{{ log.type }}</span></td>
-                    <td>@{{ log.socketId }}</td>
-                    <td>@{{ log.details }}</td>
+                    <td><pre>@{{ log.details }}</pre></td>
                     <td>@{{ log.time }}</td>
                 </tr>
                 </tbody>
@@ -207,6 +205,8 @@
                     'subscribed',
                     'client-message',
                     'api-message',
+                    'replicator-subscribed',
+                    'replicator-unsubscribed',
                 ].forEach(channelName => this.subscribeToChannel(channelName))
             },
 

src/Console/StartWebSocketServer.php

@@ -4,6 +4,8 @@
 
 use BeyondCode\LaravelWebSockets\Facades\StatisticsLogger;
 use BeyondCode\LaravelWebSockets\Facades\WebSocketsRouter;
+use BeyondCode\LaravelWebSockets\PubSub\Drivers\LocalClient;
+use BeyondCode\LaravelWebSockets\PubSub\Drivers\RedisClient;
 use BeyondCode\LaravelWebSockets\PubSub\ReplicationInterface;
 use BeyondCode\LaravelWebSockets\Server\Logger\ConnectionLogger;
 use BeyondCode\LaravelWebSockets\Server\Logger\HttpLogger;
@@ -53,6 +55,7 @@ public function handle()
             ->configureMessageLogger()
             ->configureConnectionLogger()
             ->configureRestartTimer()
+            ->configurePubSub()
             ->registerEchoRoutes()
             ->registerCustomRoutes()
             ->configurePubSubReplication()
@@ -130,6 +133,28 @@ public function configureRestartTimer()
         return $this;
     }
 
+    /**
+     * Configure the replicators.
+     *
+     * @return void
+     */
+    public function configurePubSub()
+    {
+        if (config('websockets.replication.driver', 'local') === 'local') {
+            $this->laravel->singleton(ReplicationInterface::class, function () {
+                return new LocalClient;
+            });
+        }
+
+        if (config('websockets.replication.driver', 'local') === 'redis') {
+            $this->laravel->singleton(ReplicationInterface::class, function () {
+                return (new RedisClient)->boot($this->loop);
+            });
+        }
+
+        return $this;
+    }
+
     protected function registerEchoRoutes()
     {
         WebSocketsRouter::echo();
@@ -166,7 +191,9 @@ protected function startWebSocketServer()
 
     protected function configurePubSubReplication()
     {
-        $this->laravel->get(ReplicationInterface::class)->boot($this->loop);
+        $this->laravel
+            ->get(ReplicationInterface::class)
+            ->boot($this->loop);
 
         return $this;
     }

src/Dashboard/DashboardLogger.php

@@ -9,68 +9,116 @@
 class DashboardLogger
 {
     const LOG_CHANNEL_PREFIX = 'private-websockets-dashboard-';
+
     const TYPE_DISCONNECTION = 'disconnection';
+
     const TYPE_CONNECTION = 'connection';
+
     const TYPE_VACATED = 'vacated';
+
     const TYPE_OCCUPIED = 'occupied';
+
     const TYPE_SUBSCRIBED = 'subscribed';
+
     const TYPE_CLIENT_MESSAGE = 'client-message';
+
     const TYPE_API_MESSAGE = 'api-message';
 
+    const TYPE_REPLICATOR_SUBSCRIBED = 'replicator-subscribed';
+
+    const TYPE_REPLICATOR_UNSUBSCRIBED = 'replicator-unsubscribed';
+
     public static function connection(ConnectionInterface $connection)
     {
         /** @var \GuzzleHttp\Psr7\Request $request */
         $request = $connection->httpRequest;
 
         static::log($connection->app->id, static::TYPE_CONNECTION, [
-            'details' => "Origin: {$request->getUri()->getScheme()}://{$request->getUri()->getHost()}",
-            'socketId' => $connection->socketId,
+            'details' => [
+                'origin' => "{$request->getUri()->getScheme()}://{$request->getUri()->getHost()}",
+                'socketId' => $connection->socketId,
+            ],
         ]);
     }
 
     public static function occupied(ConnectionInterface $connection, string $channelName)
     {
         static::log($connection->app->id, static::TYPE_OCCUPIED, [
-            'details' => "Channel: {$channelName}",
+            'details' => [
+                'channel' => $channelName,
+            ],
         ]);
     }
 
     public static function subscribed(ConnectionInterface $connection, string $channelName)
     {
         static::log($connection->app->id, static::TYPE_SUBSCRIBED, [
-            'socketId' => $connection->socketId,
-            'details' => "Channel: {$channelName}",
+            'details' => [
+                'socketId' => $connection->socketId,
+                'channel' => $channelName,
+            ],
         ]);
     }
 
     public static function clientMessage(ConnectionInterface $connection, stdClass $payload)
     {
         static::log($connection->app->id, static::TYPE_CLIENT_MESSAGE, [
-            'details' => "Channel: {$payload->channel}, Event: {$payload->event}",
-            'socketId' => $connection->socketId,
-            'data' => json_encode($payload),
+            'details' => [
+                'socketId' => $connection->socketId,
+                'channel' => $payload->channel,
+                'event' => $payload->event,
+                'data' => $payload,
+            ],
         ]);
     }
 
     public static function disconnection(ConnectionInterface $connection)
     {
         static::log($connection->app->id, static::TYPE_DISCONNECTION, [
-            'socketId' => $connection->socketId,
+            'details' => [
+                'socketId' => $connection->socketId,
+            ],
         ]);
     }
 
     public static function vacated(ConnectionInterface $connection, string $channelName)
     {
         static::log($connection->app->id, static::TYPE_VACATED, [
-            'details' => "Channel: {$channelName}",
+            'details' => [
+                'socketId' => $connection->socketId,
+                'channel' => $channelName,
+            ],
         ]);
     }
 
     public static function apiMessage($appId, string $channel, string $event, string $payload)
     {
         static::log($appId, static::TYPE_API_MESSAGE, [
-            'details' => "Channel: {$channel}, Event: {$event}",
-            'data' => $payload,
+            'details' => [
+                'channel' => $connection,
+                'event' => $event,
+                'payload' => $payload,
+            ],
+        ]);
+    }
+
+    public static function replicatorSubscribed(string $appId, string $channel, string $serverId)
+    {
+        static::log($appId, static::TYPE_REPLICATOR_SUBSCRIBED, [
+            'details' => [
+                'serverId' => $serverId,
+                'channel' => $channel,
+            ],
+        ]);
+    }
+
+    public static function replicatorUnsubscribed(string $appId, string $channel, string $serverId)
+    {
+        static::log($appId, static::TYPE_REPLICATOR_UNSUBSCRIBED, [
+            'details' => [
+                'serverId' => $serverId,
+                'channel' => $channel,
+            ],
         ]);
     }
 

src/HttpApi/Controllers/FetchChannelsController.php

@@ -8,18 +8,19 @@
 use Illuminate\Http\Request;
 use Illuminate\Support\Collection;
 use Illuminate\Support\Str;
+use stdClass;
 use Symfony\Component\HttpKernel\Exception\HttpException;
 
 class FetchChannelsController extends Controller
 {
     /** @var ReplicationInterface */
-    protected $replication;
+    protected $replicator;
 
-    public function __construct(ChannelManager $channelManager, ReplicationInterface $replication)
+    public function __construct(ChannelManager $channelManager, ReplicationInterface $replicator)
     {
         parent::__construct($channelManager);
 
-        $this->replication = $replication;
+        $this->replicator = $replicator;
     }
 
     public function __invoke(Request $request)
@@ -51,18 +52,21 @@ public function __invoke(Request $request)
 
         // 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
+        return $this->replicator
             ->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()];
-                        }
+                $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();
+
+                return [
+                    'channels' => $channels ?: new stdClass,
                 ];
             });
     }