[refactor/tests] Added tests for Redis PubSub replication

.github/workflows/run-tests.yml

@@ -11,19 +11,25 @@ jobs:
                 os: [ubuntu-latest, windows-latest]
                 php: [7.4, 7.3, 7.2]
                 laravel: [6.*, 7.*]
+                redis: [5, 6]
                 dependency-version: [prefer-lowest, prefer-stable]
                 include:
                     -   laravel: 7.*
                         testbench: 5.*
                     -   laravel: 6.*
                         testbench: 4.*
 
-        name: P${{ matrix.php }} - L${{ matrix.laravel }} - ${{ matrix.dependency-version }} - ${{ matrix.os }}
+        name: P${{ matrix.php }} - L${{ matrix.laravel }} - R${{ matrix.redis }} - ${{ matrix.dependency-version }} - ${{ matrix.os }}
 
         steps:
             -   name: Checkout code
                 uses: actions/checkout@v1
 
+            -   name: Setup Redis ${{ matrix.redis }}
+                uses: supercharge/redis-github-action@1.1.0
+                with:
+                    redis-version: ${{ matrix.redis }}
+
             -   name: Cache dependencies
                 uses: actions/cache@v1
                 with:
@@ -42,9 +48,14 @@ jobs:
                     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: REPLICATION_DRIVER=local phpunit --coverage-text --coverage-clover=coverage_local.xml
+
+            -   name: Execute tests with Redis driver
+                run: REPLICATION_DRIVER=redis phpunit --coverage-text --coverage-clover=coverage_redis.xml
+                if: ${{ matrix.os == 'ubuntu-latest' }}
 
             -   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' => [
 

src/Console/StartWebSocketServer.php

@@ -166,7 +166,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/HttpApi/Controllers/FetchChannelsController.php

@@ -13,13 +13,13 @@
 class FetchChannelsController extends Controller
 {
     /** @var ReplicationInterface */
-    protected $replication;
+    protected $pubsub;
 
-    public function __construct(ChannelManager $channelManager, ReplicationInterface $replication)
+    public function __construct(ChannelManager $channelManager, ReplicationInterface $pubsub)
     {
         parent::__construct($channelManager);
 
-        $this->replication = $replication;
+        $this->pubsub = $pubsub;
     }
 
     public function __invoke(Request $request)
@@ -51,7 +51,7 @@ 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->pubsub
             ->channelMemberCounts($request->appId, $channelNames)
             ->then(function (array $counts) use ($channels, $attributes) {
                 return [

src/PubSub/Drivers/LocalClient.php

@@ -38,7 +38,6 @@ public function boot(LoopInterface $loop): ReplicationInterface
      */
     public function publish(string $appId, string $channel, stdClass $payload): bool
     {
-        // Nothing to do, nobody to publish to
         return true;
     }
 

src/PubSub/Drivers/RedisClient.php

@@ -257,20 +257,24 @@ public function channelMemberCounts(string $appId, array $channelNames): Promise
      */
     protected function getConnectionUri()
     {
-        $name = config('websockets.replication.connection') ?? 'default';
-        $config = config("database.redis.$name");
+        $name = config('websockets.replication.redis.connection') ?? 'default';
+        $config = config('database.redis')[$name];
+
         $host = $config['host'];
-        $port = $config['port'] ? (':'.$config['port']) : ':6379';
+        $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 : '');
+        return "redis://{$host}:{$port}".($query ? "?{$query}" : '');
     }
 }

src/WebSockets/Channels/Channel.php

@@ -15,15 +15,15 @@ class Channel
     protected $channelName;
 
     /** @var ReplicationInterface */
-    protected $replication;
+    protected $pubsub;
 
     /** @var \Ratchet\ConnectionInterface[] */
     protected $subscribedConnections = [];
 
     public function __construct(string $channelName)
     {
         $this->channelName = $channelName;
-        $this->replication = app(ReplicationInterface::class);
+        $this->pubsub = app(ReplicationInterface::class);
     }
 
     public function getChannelName(): string
@@ -68,7 +68,7 @@ public function subscribe(ConnectionInterface $connection, stdClass $payload)
         $this->saveConnection($connection);
 
         // Subscribe to broadcasted messages from the pub/sub backend
-        $this->replication->subscribe($connection->app->id, $this->channelName);
+        $this->pubsub->subscribe($connection->app->id, $this->channelName);
 
         $connection->send(json_encode([
             'event' => 'pusher_internal:subscription_succeeded',
@@ -81,7 +81,7 @@ public function unsubscribe(ConnectionInterface $connection)
         unset($this->subscribedConnections[$connection->socketId]);
 
         // Unsubscribe from the pub/sub backend
-        $this->replication->unsubscribe($connection->app->id, $this->channelName);
+        $this->pubsub->unsubscribe($connection->app->id, $this->channelName);
 
         if (! $this->hasConnections()) {
             DashboardLogger::vacated($connection, $this->channelName);
@@ -120,7 +120,7 @@ public function broadcastToEveryoneExcept($payload, ?string $socketId, string $a
         // in this case. If this came from TriggerEventController, then we still want
         // to publish to get the message out to other server instances.
         if ($publish) {
-            $this->replication->publish($appId, $this->channelName, $payload);
+            $this->pubsub->publish($appId, $this->channelName, $payload);
         }
 
         // Performance optimization, if we don't have a socket ID,

src/WebSockets/Channels/PresenceChannel.php

@@ -28,7 +28,7 @@ class PresenceChannel extends Channel
     public function getUsers(string $appId)
     {
         // Get the members list from the replication backend
-        return $this->replication
+        return $this->pubsub
             ->channelMembers($appId, $this->channelName);
     }
 
@@ -49,7 +49,7 @@ public function subscribe(ConnectionInterface $connection, stdClass $payload)
         $this->users[$connection->socketId] = $channelData;
 
         // Add the connection as a member of the channel
-        $this->replication
+        $this->pubsub
             ->joinChannel(
                 $connection->app->id,
                 $this->channelName,
@@ -59,7 +59,7 @@ public function subscribe(ConnectionInterface $connection, stdClass $payload)
 
         // We need to pull the channel data from the replication backend,
         // otherwise we won't be sending the full details of the channel
-        $this->replication
+        $this->pubsub
             ->channelMembers($connection->app->id, $this->channelName)
             ->then(function ($users) use ($connection) {
                 // Send the success event
@@ -86,7 +86,7 @@ public function unsubscribe(ConnectionInterface $connection)
         }
 
         // Remove the connection as a member of the channel
-        $this->replication
+        $this->pubsub
             ->leaveChannel(
                 $connection->app->id,
                 $this->channelName,
@@ -110,7 +110,7 @@ public function unsubscribe(ConnectionInterface $connection)
      */
     public function toArray(string $appId = null)
     {
-        return $this->replication
+        return $this->pubsub
             ->channelMembers($appId, $this->channelName)
             ->then(function ($users) {
                 return array_merge(parent::toArray(), [

src/WebSocketsServiceProvider.php

@@ -24,6 +24,7 @@
 use Illuminate\Support\ServiceProvider;
 use Psr\Log\LoggerInterface;
 use Pusher\Pusher;
+use React\EventLoop\Factory as LoopFactory;
 
 class WebSocketsServiceProvider extends ServiceProvider
 {
@@ -56,19 +57,19 @@ public function boot()
 
     protected function configurePubSub()
     {
-        if (config('websockets.replication.enabled') !== true || config('websockets.replication.driver') !== 'redis') {
+        if (config('websockets.replication.driver') === 'local') {
             $this->app->singleton(ReplicationInterface::class, function () {
-                return new LocalClient();
+                return new LocalClient;
             });
-
-            return;
         }
 
-        $this->app->singleton(ReplicationInterface::class, function () {
-            return (new RedisClient())->boot($this->loop);
-        });
+        if (config('websockets.replication.driver') === 'redis') {
+            $this->app->singleton(ReplicationInterface::class, function () {
+                return (new RedisClient)->boot($this->loop ?? LoopFactory::create());
+            });
+        }
 
-        $this->app->get(BroadcastManager::class)->extend('redis-pusher', function ($app, array $config) {
+        $this->app->get(BroadcastManager::class)->extend('websockets', function ($app, array $config) {
             $pusher = new Pusher(
                 $config['key'], $config['secret'],
                 $config['app_id'], $config['options'] ?? []

tests/Channels/ChannelReplicationTest.php

@@ -2,16 +2,7 @@
 
 namespace BeyondCode\LaravelWebSockets\Tests\Channels;
 
-use BeyondCode\LaravelWebSockets\Tests\TestsReplication;
-
 class ChannelReplicationTest extends ChannelTest
 {
-    use TestsReplication;
-
-    public function setUp(): void
-    {
-        parent::setUp();
-
-        $this->setupReplication();
-    }
+    //
 }

tests/Channels/PresenceChannelReplicationTest.php

@@ -2,16 +2,7 @@
 
 namespace BeyondCode\LaravelWebSockets\Tests\Channels;
 
-use BeyondCode\LaravelWebSockets\Tests\TestsReplication;
-
 class PresenceChannelReplicationTest extends PresenceChannelTest
 {
-    use TestsReplication;
-
-    public function setUp(): void
-    {
-        parent::setUp();
-
-        $this->setupReplication();
-    }
+    //
 }

tests/HttpApi/FetchChannelReplicationTest.php

@@ -2,16 +2,7 @@
 
 namespace BeyondCode\LaravelWebSockets\Tests\HttpApi;
 
-use BeyondCode\LaravelWebSockets\Tests\TestsReplication;
-
 class FetchChannelReplicationTest extends FetchChannelTest
 {
-    use TestsReplication;
-
-    public function setUp(): void
-    {
-        parent::setUp();
-
-        $this->setupReplication();
-    }
+    //
 }

tests/HttpApi/FetchChannelsReplicationTest.php

@@ -2,16 +2,7 @@
 
 namespace BeyondCode\LaravelWebSockets\Tests\HttpApi;
 
-use BeyondCode\LaravelWebSockets\Tests\TestsReplication;
-
 class FetchChannelsReplicationTest extends FetchChannelsTest
 {
-    use TestsReplication;
-
-    public function setUp(): void
-    {
-        parent::setUp();
-
-        $this->setupReplication();
-    }
+    //
 }

tests/HttpApi/FetchUsersReplicationTest.php

@@ -2,16 +2,7 @@
 
 namespace BeyondCode\LaravelWebSockets\Tests\HttpApi;
 
-use BeyondCode\LaravelWebSockets\Tests\TestsReplication;
-
 class FetchUsersReplicationTest extends FetchUsersTest
 {
-    use TestsReplication;
-
-    public function setUp(): void
-    {
-        parent::setUp();
-
-        $this->setupReplication();
-    }
+    //
 }

tests/TestCase.php

@@ -56,6 +56,18 @@ protected function getEnvironmentSetUp($app)
             ],
         ]);
 
+        $app['config']->set('database.redis.default', [
+            'host' => env('REDIS_HOST', '127.0.0.1'),
+            'password' => env('REDIS_PASSWORD', null),
+            'port' => env('REDIS_PORT', '6379'),
+            'database' => env('REDIS_DB', '0'),
+        ]);
+
+        $app['config']->set(
+            'websockets.replication.driver',
+            getenv('REPLICATION_DRIVER') ?: 'local'
+        );
+
         include_once __DIR__.'/../database/migrations/create_websockets_statistics_entries_table.php.stub';
 
         (new \CreateWebSocketsStatisticsEntriesTable())->up();