Merge pull request #490 from beyondcode/fix/backend-streaming

[2.x] Removed 'websockets' broadcaster driver

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.

resources/views/dashboard.blade.php

@@ -405,13 +405,7 @@ class="rounded-full px-3 py-1 inline-block text-sm"
 
           axios
             .post('/event', payload)
-            .then(() => {
-              this.form = {
-                channel: null,
-                event: null,
-                data: null,
-              };
-            })
+            .then(() => {})
             .catch(err => {
               alert('Error sending event.');
             })

src/Contracts/PushesToPusher.php

@@ -2,7 +2,6 @@
 
 namespace BeyondCode\LaravelWebSockets\Contracts;
 
-use BeyondCode\LaravelWebSockets\PubSub\Broadcasters\RedisPusherBroadcaster;
 use Illuminate\Broadcasting\Broadcasters\PusherBroadcaster;
 use Pusher\Pusher;
 
@@ -16,16 +15,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

@@ -3,6 +3,7 @@
 namespace BeyondCode\LaravelWebSockets\HttpApi\Controllers;
 
 use BeyondCode\LaravelWebSockets\Apps\App;
+use BeyondCode\LaravelWebSockets\PubSub\ReplicationInterface;
 use BeyondCode\LaravelWebSockets\QueryParameters;
 use BeyondCode\LaravelWebSockets\WebSockets\Channels\ChannelManager;
 use Exception;
@@ -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

@@ -2,8 +2,6 @@
 
 namespace BeyondCode\LaravelWebSockets\HttpApi\Controllers;
 
-use BeyondCode\LaravelWebSockets\PubSub\ReplicationInterface;
-use BeyondCode\LaravelWebSockets\WebSockets\Channels\ChannelManager;
 use BeyondCode\LaravelWebSockets\WebSockets\Channels\PresenceChannel;
 use Illuminate\Http\Request;
 use Illuminate\Support\Collection;
@@ -13,27 +11,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,
         ]);
     }