wip

Author: rennokki

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();

src/PubSub/Drivers/LocalClient.php

@@ -21,9 +21,10 @@ class LocalClient implements ReplicationInterface
      * Boot the pub/sub provider (open connections, initial subscriptions, etc).
      *
      * @param  LoopInterface  $loop
+     * @param  string|null  $factoryClass
      * @return self
      */
-    public function boot(LoopInterface $loop): ReplicationInterface
+    public function boot(LoopInterface $loop, $factoryClass = null): ReplicationInterface
     {
         return $this;
     }

src/PubSub/Drivers/RedisClient.php

@@ -66,14 +66,17 @@ public function __construct()
      * Boot the RedisClient, initializing the connections.
      *
      * @param  LoopInterface  $loop
+     * @param  string|null  $factoryClass
      * @return ReplicationInterface
      */
-    public function boot(LoopInterface $loop): ReplicationInterface
+    public function boot(LoopInterface $loop, $factoryClass = null): ReplicationInterface
     {
+        $factoryClass = $factoryClass ?: Factory::class;
+
         $this->loop = $loop;
 
         $connectionUri = $this->getConnectionUri();
-        $factory = new Factory($this->loop);
+        $factory = new $factoryClass($this->loop);
 
         $this->publishClient = $factory->createLazyClient($connectionUri);
         $this->subscribeClient = $factory->createLazyClient($connectionUri);
@@ -108,7 +111,7 @@ protected function onMessage(string $redisChannel, string $payload)
         // We need to put the channel name in the payload.
         // We strip the app ID from the channel name, websocket clients
         // expect the channel name to not include the app ID.
-        $payload->channel = Str::after($redisChannel, "$appId:");
+        $payload->channel = Str::after($redisChannel, "{$appId}:");
 
         $channelManager = app(ChannelManager::class);
 
@@ -296,4 +299,34 @@ protected function getConnectionUri()
 
         return "redis://{$host}:{$port}".($query ? "?{$query}" : '');
     }
+
+    /**
+     * Get the Subscribe client instance.
+     *
+     * @return Client
+     */
+    public function getSubscribeClient()
+    {
+        return $this->subscribeClient;
+    }
+
+    /**
+     * Get the Publish client instance.
+     *
+     * @return Client
+     */
+    public function getPublishClient()
+    {
+        return $this->publishClient;
+    }
+
+    /**
+     * Get the unique identifier for the server.
+     *
+     * @return string
+     */
+    public function getServerId()
+    {
+        return $this->serverId;
+    }
 }

src/PubSub/ReplicationInterface.php

@@ -12,9 +12,10 @@ interface ReplicationInterface
      * Boot the pub/sub provider (open connections, initial subscriptions, etc).
      *
      * @param  LoopInterface  $loop
+     * @param  string|null  $factoryClass
      * @return self
      */
-    public function boot(LoopInterface $loop): self;
+    public function boot(LoopInterface $loop, $factoryClass = null): self;
 
     /**
      * Publish a payload on a specific channel, for a specific app.

src/WebSocketsServiceProvider.php

@@ -9,9 +9,6 @@
 use BeyondCode\LaravelWebSockets\Dashboard\Http\Controllers\ShowDashboard;
 use BeyondCode\LaravelWebSockets\Dashboard\Http\Middleware\Authorize as AuthorizeDashboard;
 use BeyondCode\LaravelWebSockets\PubSub\Broadcasters\RedisPusherBroadcaster;
-use BeyondCode\LaravelWebSockets\PubSub\Drivers\LocalClient;
-use BeyondCode\LaravelWebSockets\PubSub\Drivers\RedisClient;
-use BeyondCode\LaravelWebSockets\PubSub\ReplicationInterface;
 use BeyondCode\LaravelWebSockets\Server\Router;
 use BeyondCode\LaravelWebSockets\Statistics\Http\Controllers\WebSocketStatisticsEntriesController;
 use BeyondCode\LaravelWebSockets\Statistics\Http\Middleware\Authorize as AuthorizeStatistics;
@@ -54,18 +51,6 @@ public function boot()
 
     protected function configurePubSub()
     {
-        if (config('websockets.replication.driver', 'local') === 'local') {
-            $this->app->singleton(ReplicationInterface::class, function () {
-                return new LocalClient;
-            });
-        }
-
-        if (config('websockets.replication.driver', 'local') === 'redis') {
-            $this->app->singleton(ReplicationInterface::class, function () {
-                return (new RedisClient)->boot($this->loop ?? LoopFactory::create());
-            });
-        }
-
         $this->app->make(BroadcastManager::class)->extend('websockets', function ($app, array $config) {
             $pusher = new Pusher(
                 $config['key'], $config['secret'],

tests/Channels/ChannelReplicationTest.php

@@ -6,5 +6,13 @@
 
 class ChannelReplicationTest extends TestCase
 {
-    //
+    /**
+     * {@inheritdoc}
+     */
+    public function setUp(): void
+    {
+        parent::setUp();
+
+        $this->runOnlyOnRedisReplication();
+    }
 }

tests/Channels/PresenceChannelReplicationTest.php

@@ -6,5 +6,13 @@
 
 class PresenceChannelReplicationTest extends TestCase
 {
-    //
+    /**
+     * {@inheritdoc}
+     */
+    public function setUp(): void
+    {
+        parent::setUp();
+
+        $this->runOnlyOnRedisReplication();
+    }
 }

tests/Channels/PresenceChannelTest.php

@@ -2,6 +2,7 @@
 
 namespace BeyondCode\LaravelWebSockets\Tests\Channels;
 
+use BeyondCode\LaravelWebSockets\PubSub\ReplicationInterface;
 use BeyondCode\LaravelWebSockets\Tests\Mocks\Message;
 use BeyondCode\LaravelWebSockets\Tests\TestCase;
 use BeyondCode\LaravelWebSockets\WebSockets\Exceptions\InvalidSignature;
@@ -55,9 +56,27 @@ public function clients_with_valid_auth_signatures_can_join_presence_channels()
 
         $this->pusherServer->onMessage($connection, $message);
 
-        $connection->assertSentEvent('pusher_internal:subscription_succeeded', [
-            'channel' => 'presence-channel',
-        ]);
+        $this->getPublishClient()
+            ->assertCalledWithArgs('hset', [
+                '1234:presence-channel',
+                $connection->socketId,
+                json_encode($channelData),
+            ])
+            ->assertCalledWithArgs('hgetall', [
+                '1234:presence-channel'
+            ]);
+            // TODO: This fails somehow
+            // Debugging shows the exact same pattern as good.
+            /* ->assertCalledWithArgs('publish', [
+                '1234:presence-channel',
+                json_encode([
+                    'event' => 'pusher_internal:member_added',
+                    'channel' => 'presence-channel',
+                    'data' => $channelData,
+                    'appId' => '1234',
+                    'serverId' => $this->app->make(ReplicationInterface::class)->getServerId(),
+                ]),
+            ]) */
     }
 
     /** @test */

tests/Channels/PrivateChannelReplicationTest.php

@@ -0,0 +1,18 @@
+<?php
+
+namespace BeyondCode\LaravelWebSockets\Tests\Channels;
+
+use BeyondCode\LaravelWebSockets\Tests\TestCase;
+
+class PrivateChannelReplicationTest extends TestCase
+{
+    /**
+     * {@inheritdoc}
+     */
+    public function setUp(): void
+    {
+        parent::setUp();
+
+        $this->runOnlyOnRedisReplication();
+    }
+}