Fixed backend streaming

Author: rennokki

docs/horizontal-scaling/getting-started.md

@@ -27,43 +27,8 @@ To enable the replication, simply change the `replication.driver` name in the `w
 ],
 ```
 
+Now, when your app broadcasts the message, it will make sure the connection reaches other servers which are under the same load balancer.
+
 The available drivers for replication are:
 
 - [Redis](redis)
-
-## Configure the Broadcasting driver
-
-Laravel WebSockets comes with an additional `websockets` broadcaster driver that accepts configurations like the Pusher driver, but will make sure the broadcasting will work across all websocket servers:
-
-```php
-'connections' => [
-    'pusher' => [
-        ...
-    ],
-
-    'websockets' => [
-        'driver' => 'websockets',
-        'key' => env('PUSHER_APP_KEY'),
-        'secret' => env('PUSHER_APP_SECRET'),
-        'app_id' => env('PUSHER_APP_ID'),
-        'options' => [
-            'cluster' => env('PUSHER_APP_CLUSTER'),
-            'encrypted' => true,
-            'host' => env('PUSHER_APP_HOST', '127.0.0.1'),
-            'port' => env('PUSHER_APP_PORT', 6001),
-            'scheme' => env('PUSHER_APP_SCHEME', 'http'),
-            'curl_options' => [
-                CURLOPT_SSL_VERIFYHOST => 0,
-                CURLOPT_SSL_VERIFYPEER => 0,
-            ],
-        ],
-    ],
-```
-
-Make sure to change the `BROADCAST_DRIVER`:
-
-```
-BROADCAST_DRIVER=websockets
-```
-
-Now, when your app broadcasts the message, it will make sure the connection reaches other servers which are under the same load balancer.

src/Contracts/PushesToPusher.php

@@ -16,16 +16,13 @@ trait PushesToPusher
      */
     public function getPusherBroadcaster(array $app)
     {
-        if (config('websockets.replication.driver') === 'redis') {
-            return new RedisPusherBroadcaster(
-                new Pusher($app['key'], $app['secret'], $app['id'], config('broadcasting.connections.websockets.options', [])),
-                $app['id'],
-                app('redis')
-            );
-        }
-
         return new PusherBroadcaster(
-            new Pusher($app['key'], $app['secret'], $app['id'], config('broadcasting.connections.pusher.options', []))
+            new Pusher(
+                $app['key'],
+                $app['secret'],
+                $app['id'],
+                config('broadcasting.connections.pusher.options', [])
+            )
         );
     }
 }

src/HttpApi/Controllers/Controller.php

@@ -19,6 +19,7 @@
 use React\Promise\PromiseInterface;
 use Symfony\Bridge\PsrHttpMessage\Factory\HttpFoundationFactory;
 use Symfony\Component\HttpKernel\Exception\HttpException;
+use BeyondCode\LaravelWebSockets\PubSub\ReplicationInterface;
 
 abstract class Controller implements HttpServerInterface
 {
@@ -51,15 +52,24 @@ abstract class Controller implements HttpServerInterface
      */
     protected $channelManager;
 
+    /**
+     * The replicator driver.
+     *
+     * @var \BeyondCode\LaravelWebSockets\PubSub\ReplicationInterface
+     */
+    protected $replicator;
+
     /**
      * Initialize the request.
      *
      * @param  \BeyondCode\LaravelWebSockets\WebSockets\Channels\ChannelManager  $channelManager
+     * @param  \BeyondCode\LaravelWebSockets\PubSub\ReplicationInterface  $replicator
      * @return void
      */
-    public function __construct(ChannelManager $channelManager)
+    public function __construct(ChannelManager $channelManager, ReplicationInterface $replicator)
     {
         $this->channelManager = $channelManager;
+        $this->replicator = $replicator;
     }
 
     /**

src/HttpApi/Controllers/FetchChannelsController.php

@@ -13,27 +13,6 @@
 
 class FetchChannelsController extends Controller
 {
-    /**
-     * The replicator driver.
-     *
-     * @var \BeyondCode\LaravelWebSockets\PubSub\ReplicationInterface
-     */
-    protected $replicator;
-
-    /**
-     * Initialize the class.
-     *
-     * @param  \BeyondCode\LaravelWebSockets\WebSockets\Channels\ChannelManager  $channelManager
-     * @param  \BeyondCode\LaravelWebSockets\PubSub\ReplicationInterface  $replicator
-     * @return void
-     */
-    public function __construct(ChannelManager $channelManager, ReplicationInterface $replicator)
-    {
-        parent::__construct($channelManager);
-
-        $this->replicator = $replicator;
-    }
-
     /**
      * Handle the incoming request.
      *

src/HttpApi/Controllers/TriggerEventController.php

@@ -18,14 +18,26 @@ public function __invoke(Request $request)
     {
         $this->ensureValidSignature($request);
 
-        foreach ($request->json()->get('channels', []) as $channelName) {
+        $channels = $request->channels ?: [];
+
+        foreach ($channels as $channelName) {
             $channel = $this->channelManager->find($request->appId, $channelName);
 
-            optional($channel)->broadcastToEveryoneExcept((object) [
+            $payload = (object) [
                 'channel' => $channelName,
-                'event' => $request->json()->get('name'),
-                'data' => $request->json()->get('data'),
-            ], $request->json()->get('socket_id'), $request->appId);
+                'event' => $request->name,
+                'data' => $request->data,
+            ];
+
+            optional($channel)->broadcastToEveryoneExcept($payload, $request->socket_id, $request->appId);
+
+            // If the setup is horizontally-scaled using the Redis Pub/Sub,
+            // then we're going to make sure it gets streamed to the other
+            // servers as well that are subscribed to the Pub/Sub topics
+            // attached to the current iterated app & channel.
+            // For local setups, the local driver will ignore the publishes.
+
+            $this->replicator->publish($request->appId, $channelName, $payload);
 
             DashboardLogger::log($request->appId, DashboardLogger::TYPE_API_MESSAGE, [
                 'channel' => $channelName,

src/PubSub/Broadcasters/RedisPusherBroadcaster.php

@@ -293,23 +293,23 @@ public function onMessage(string $redisChannel, string $payload)
             return;
         }
 
-        $socket = $payload->socket ?? null;
+        $socketId = $payload->socketId ?? null;
         $serverId = $payload->serverId ?? null;
 
         // Remove fields intended for internal use from the payload.
-        unset($payload->socket);
+        unset($payload->socketId);
         unset($payload->serverId);
         unset($payload->appId);
 
         // Push the message out to connected websocket clients.
-        $channel->broadcastToEveryoneExcept($payload, $socket, $appId, false);
+        $channel->broadcastToEveryoneExcept($payload, $socketId, $appId, false);
 
         DashboardLogger::log($appId, DashboardLogger::TYPE_REPLICATOR_MESSAGE_RECEIVED, [
             'channel' => $channel->getChannelName(),
             'redisChannel' => $redisChannel,
             'serverId' => $this->getServerId(),
             'incomingServerId' => $serverId,
-            'incomingSocketId' => $socket,
+            'incomingSocketId' => $socketId,
             'payload' => $payload,
         ]);
     }

src/PubSub/Drivers/RedisClient.php

@@ -8,7 +8,6 @@
 use BeyondCode\LaravelWebSockets\Dashboard\Http\Controllers\ShowDashboard;
 use BeyondCode\LaravelWebSockets\Dashboard\Http\Controllers\ShowStatistics;
 use BeyondCode\LaravelWebSockets\Dashboard\Http\Middleware\Authorize as AuthorizeDashboard;
-use BeyondCode\LaravelWebSockets\PubSub\Broadcasters\RedisPusherBroadcaster;
 use BeyondCode\LaravelWebSockets\Server\Router;
 use BeyondCode\LaravelWebSockets\Statistics\Drivers\StatisticsDriver;
 use BeyondCode\LaravelWebSockets\WebSockets\Channels\ChannelManager;
@@ -47,8 +46,6 @@ public function boot()
             Console\CleanStatistics::class,
             Console\RestartWebSocketServer::class,
         ]);
-
-        $this->configurePubSub();
     }
 
     /**
@@ -85,31 +82,6 @@ public function register()
         });
     }
 
-    /**
-     * Configure the PubSub replication.
-     *
-     * @return void
-     */
-    protected function configurePubSub()
-    {
-        $this->app->make(BroadcastManager::class)->extend('websockets', function ($app, array $config) {
-            $pusher = new Pusher(
-                $config['key'], $config['secret'],
-                $config['app_id'], $config['options'] ?? []
-            );
-
-            if ($config['log'] ?? false) {
-                $pusher->setLogger($this->app->make(LoggerInterface::class));
-            }
-
-            return new RedisPusherBroadcaster(
-                $pusher,
-                $config['app_id'],
-                $this->app->make('redis')
-            );
-        });
-    }
-
     /**
      * Register the dashboard routes.
      *

src/WebSocketsServiceProvider.php

@@ -50,7 +50,8 @@ public function can_send_message_on_redis_replication()
                 'data' => json_encode(['data' => 'yes']),
             ])
             ->seeJson([
-                'ok' => true,
+                'exception' => 'Failed to connect to Pusher.',
+                'ok' => false,
             ]);
     }