Getting rid of silly double queue

Author: georgeboot

src/Commands/VaporHandle.php

@@ -67,7 +67,10 @@ protected function handleWebsocketEvent(array $event, Context $context): void
      */
     protected function message()
     {
-        return tap(json_decode(base64_decode($this->argument('message')), true), function ($message) {
+        /** @var string $message */
+        $message = $this->argument('message');
+
+        return tap(json_decode(base64_decode($message), true), function ($message) {
             if ($message === false) {
                 throw new InvalidArgumentException('Unable to unserialize message.');
             }

src/Driver.php

@@ -2,7 +2,7 @@
 
 namespace Georgeboot\LaravelEchoApiGateway;
 
-use Georgeboot\LaravelEchoApiGateway\Jobs\QueueMessageToChannels;
+use Aws\ApiGatewayManagementApi\Exception\ApiGatewayManagementApiException;
 use Illuminate\Broadcasting\Broadcasters\Broadcaster;
 use Illuminate\Broadcasting\Broadcasters\UsePusherChannelConventions;
 use Illuminate\Support\Arr;
@@ -13,6 +13,15 @@ class Driver extends Broadcaster
 {
     use UsePusherChannelConventions;
 
+    protected SubscriptionRepository $subscriptionRepository;
+    protected ConnectionRepository $connectionRepository;
+
+    public function __construct(SubscriptionRepository $subscriptionRepository, ConnectionRepository $connectionRepository)
+    {
+        $this->subscriptionRepository = $subscriptionRepository;
+        $this->connectionRepository = $connectionRepository;
+    }
+
     /**
      * Authenticate the incoming request for a given channel.
      *
@@ -108,11 +117,11 @@ public function broadcast(array $channels, $event, array $payload = [])
             'data' => $payload,
         ], JSON_THROW_ON_ERROR);
 
-        dispatch(new QueueMessageToChannels(
-            $channels,
-            $data,
-            Arr::pull($payload, 'socket')
-        ));
+        $skipConnectionId = Arr::pull($payload, 'socket');
+
+        $this->subscriptionRepository->getConnectionIdsForChannels(...$channels)
+            ->reject(fn($connectionId) => $connectionId === $skipConnectionId)
+            ->each(fn(string $connectionId) => $this->sendMessage($connectionId, $data));
 
         return;
 
@@ -127,4 +136,26 @@ public function broadcast(array $channels, $event, array $payload = [])
         //         : 'Failed to connect to Pusher.'
         // );
     }
+
+    protected function sendMessage(string $connectionId, string $data): void
+    {
+        try {
+            $this->connectionRepository->sendMessage($connectionId, $data);
+        } catch (ApiGatewayManagementApiException $exception) {
+            if ($exception->getAwsErrorCode() === 'GoneException') {
+                $this->subscriptionRepository->clearConnection($connectionId);
+                return;
+            }
+
+            throw $exception;
+
+            // $exception->getErrorCode() is one of:
+            // GoneException
+            // LimitExceededException
+            // PayloadTooLargeException
+            // ForbiddenException
+
+            // otherwise: call $exception->getPrevious() which is a guzzle exception
+        }
+    }
 }

src/Handler.php

@@ -6,19 +6,16 @@
 use Bref\Event\ApiGateway\WebsocketEvent;
 use Bref\Event\ApiGateway\WebsocketHandler;
 use Bref\Event\Http\HttpResponse;
-use Illuminate\Contracts\Debug\ExceptionHandler;
 use Illuminate\Support\Str;
 use Throwable;
 
 class Handler extends WebsocketHandler
 {
-    protected ExceptionHandler $exceptionHandler;
     protected SubscriptionRepository $subscriptionRepository;
     protected ConnectionRepository $connectionRepository;
 
-    public function __construct(ExceptionHandler $exceptionHandler, SubscriptionRepository $subscriptionRepository, ConnectionRepository $connectionRepository)
+    public function __construct(SubscriptionRepository $subscriptionRepository, ConnectionRepository $connectionRepository)
     {
-        $this->exceptionHandler = $exceptionHandler;
         $this->subscriptionRepository = $subscriptionRepository;
         $this->connectionRepository = $connectionRepository;
     }
@@ -41,8 +38,7 @@ public function handleWebsocket(WebsocketEvent $event, Context $context): HttpRe
                 app('sentry')->captureException($throwable);
             }
 
-
-            $this->exceptionHandler->report($throwable);
+            report($throwable);
 
             throw $throwable;
         }

src/Jobs/QueueMessageToChannels.php

@@ -46,7 +46,7 @@ public function register()
     public function boot(BroadcastManager $broadcastManager): void
     {
         $broadcastManager->extend('laravel-echo-api-gateway', function (): Broadcaster {
-            return new Driver();
+            return $this->app->make(Driver::class);
         });
     }
 }