decouple message handling

Author: taylorotwell

src/Protocols/Pusher/PusherPubSubIncomingMessageHandler.php

@@ -0,0 +1,21 @@
+<?php
+
+namespace Laravel\Reverb\Protocols\Pusher;
+
+use Laravel\Reverb\Servers\Reverb\Contracts\PubSubIncomingMessageHandler;
+
+class PusherPubSubIncomingMessageHandler implements PubSubIncomingMessageHandler
+{
+    /**
+     * Handle an incoming message from the PubSub provider.
+     */
+    public function handle(string $payload): void
+    {
+        $event = json_decode($payload, true);
+
+        EventDispatcher::dispatchSynchronously(
+            unserialize($event['application']),
+            $event['payload']
+        );
+    }
+}

src/Servers/Reverb/Contracts/PubSubIncomingMessageHandler.php

@@ -0,0 +1,11 @@
+<?php
+
+namespace Laravel\Reverb\Servers\Reverb\Contracts;
+
+interface PubSubIncomingMessageHandler
+{
+    /**
+     * Handle an incoming message from the PubSub provider.
+     */
+    public function handle(string $payload): void;
+}

src/Servers/Reverb/Factory.php

@@ -7,16 +7,18 @@
 use Laravel\Reverb\Protocols\Pusher\Contracts\ChannelConnectionManager;
 use Laravel\Reverb\Protocols\Pusher\Contracts\ChannelManager;
 use Laravel\Reverb\Protocols\Pusher\Http\Controllers\ChannelController;
-use Laravel\Reverb\Protocols\Pusher\Http\Controllers\ChannelsController;
 use Laravel\Reverb\Protocols\Pusher\Http\Controllers\ChannelUsersController;
+use Laravel\Reverb\Protocols\Pusher\Http\Controllers\ChannelsController;
 use Laravel\Reverb\Protocols\Pusher\Http\Controllers\ConnectionsController;
 use Laravel\Reverb\Protocols\Pusher\Http\Controllers\EventsBatchController;
 use Laravel\Reverb\Protocols\Pusher\Http\Controllers\EventsController;
 use Laravel\Reverb\Protocols\Pusher\Http\Controllers\PusherController;
 use Laravel\Reverb\Protocols\Pusher\Http\Controllers\UsersTerminateController;
 use Laravel\Reverb\Protocols\Pusher\Managers\ArrayChannelConnectionManager;
 use Laravel\Reverb\Protocols\Pusher\Managers\ArrayChannelManager;
+use Laravel\Reverb\Protocols\Pusher\PusherPubSubIncomingMessageHandler;
 use Laravel\Reverb\Protocols\Pusher\Server as PusherServer;
+use Laravel\Reverb\Servers\Reverb\Contracts\PubSubIncomingMessageHandler;
 use Laravel\Reverb\Servers\Reverb\Http\Route;
 use Laravel\Reverb\Servers\Reverb\Http\Router;
 use Laravel\Reverb\Servers\Reverb\Http\Server as HttpServer;
@@ -63,6 +65,11 @@ public static function makePusherServer(): Router
             fn () => new ArrayChannelConnectionManager
         );
 
+        app()->singleton(
+            PubSubIncomingMessageHandler::class,
+            fn () => new PusherPubSubIncomingMessageHandler,
+        );
+
         return new Router(new UrlMatcher(static::pusherRoutes(), new RequestContext));
     }
 

src/Servers/Reverb/Publishing/RedisPubSubProvider.php

@@ -4,6 +4,7 @@
 
 use Illuminate\Support\Facades\Config;
 use Laravel\Reverb\Protocols\Pusher\EventDispatcher;
+use Laravel\Reverb\Servers\Reverb\Contracts\PubSubIncomingMessageHandler;
 use Laravel\Reverb\Servers\Reverb\Contracts\PubSubProvider;
 use React\EventLoop\LoopInterface;
 use RuntimeException;
@@ -15,6 +16,7 @@ class RedisPubSubProvider implements PubSubProvider
     protected $subscribingClient;
 
     public function __construct(protected RedisClientFactory $clientFactory,
+        protected PubSubIncomingMessageHandler $messageHandler,
         protected string $channel)
     {
     }
@@ -38,12 +40,7 @@ public function subscribe(): void
         $this->subscribingClient->subscribe($this->channel);
 
         $this->subscribingClient->on('message', function (string $channel, string $payload) {
-            $event = json_decode($payload, true);
-
-            EventDispatcher::dispatchSynchronously(
-                unserialize($event['application']),
-                $event['payload']
-            );
+            $this->messageHandler->handle($payload);
         });
     }
 

src/Servers/Reverb/ReverbServerProvider.php

@@ -7,6 +7,7 @@
 use Laravel\Reverb\Contracts\ServerProvider;
 use Laravel\Reverb\Servers\Reverb\Console\Commands\RestartServer;
 use Laravel\Reverb\Servers\Reverb\Console\Commands\StartServer;
+use Laravel\Reverb\Servers\Reverb\Contracts\PubSubIncomingMessageHandler;
 use Laravel\Reverb\Servers\Reverb\Contracts\PubSubProvider;
 use Laravel\Reverb\Servers\Reverb\Publishing\RedisClientFactory;
 use Laravel\Reverb\Servers\Reverb\Publishing\RedisPubSubProvider;
@@ -35,6 +36,7 @@ public function register(): void
     {
         $this->app->singleton(PubSubProvider::class, fn ($app) => new RedisPubSubProvider(
             $app->make(RedisClientFactory::class),
+            $app->make(PubSubIncomingMessageHandler::class),
             $this->config['scaling']['channel'] ?? 'reverb'
         ));
     }