Adding whisper feature (#24)

Author: jdavidbakr

js-src/Channel.ts

@@ -1,7 +1,7 @@
-import {EventFormatter} from 'laravel-echo/src/util';
-import {Channel as BaseChannel} from 'laravel-echo/src/channel/channel';
-import {PresenceChannel} from "laravel-echo/src/channel";
-import {Websocket} from "./Websocket";
+import { EventFormatter } from 'laravel-echo/src/util';
+import { Channel as BaseChannel } from 'laravel-echo/src/channel/channel';
+import { PresenceChannel } from "laravel-echo/src/channel";
+import { Websocket } from "./Websocket";
 
 /**
  * This class represents a Pusher channel.
@@ -105,9 +105,12 @@ export class Channel extends BaseChannel implements PresenceChannel {
     }
 
     whisper(event: string, data: object): Channel {
+        let channel = this.name;
+        let formattedEvent = "client-" + event;
         this.socket.send({
-            event,
+            "event": formattedEvent,
             data,
+            channel,
         })
 
         return this;

js-tests/Connector.test.ts

@@ -1,6 +1,6 @@
 import WS from "jest-websocket-mock";
-import {Connector} from "../js-src/Connector";
-import {Channel} from "../js-src/Channel";
+import { Connector } from "../js-src/Connector";
+import { Channel } from "../js-src/Channel";
 
 describe('Connector', () => {
     let server: WS;
@@ -54,4 +54,33 @@ describe('Connector', () => {
         expect(handler1).toBeCalled();
         expect(handler2).not.toBeCalled();
     })
+
+    test('we can send a whisper event', async () => {
+        const connector = new Connector({
+            host: "ws://localhost:1234",
+        })
+
+        await server.connected;
+
+        await expect(server).toReceiveMessage('{"event":"whoami"}');
+        server.send('{"event":"whoami","data":{"socket_id":"test-socket-id"}}')
+
+        const channel = connector.channel('my-test-channel')
+
+        await expect(server).toReceiveMessage('{"event":"subscribe","data":{"channel":"my-test-channel"}}');
+
+        server.send('{"event":"subscription_succeeded","channel":"my-test-channel"}')
+
+        expect(channel).toBeInstanceOf(Channel)
+
+        const handler1 = jest.fn();
+        const handler2 = jest.fn();
+
+        channel.on('client-whisper', handler1)
+
+        server.send('{"event":"client-whisper","data":"whisper","channel":"my-test-channel","data":{}}')
+
+        expect(handler1).toBeCalled();
+        expect(handler2).not.toBeCalled();
+    })
 });

src/Handler.php

@@ -2,10 +2,12 @@
 
 namespace Georgeboot\LaravelEchoApiGateway;
 
+use Aws\ApiGatewayManagementApi\Exception\ApiGatewayManagementApiException;
 use Bref\Context\Context;
 use Bref\Event\ApiGateway\WebsocketEvent;
 use Bref\Event\ApiGateway\WebsocketHandler;
 use Bref\Event\Http\HttpResponse;
+use Illuminate\Support\Arr;
 use Illuminate\Support\Str;
 use Throwable;
 
@@ -69,9 +71,11 @@ protected function handleMessage(WebsocketEvent $event, Context $context): void
             $this->subscribe($event, $context);
         } elseif ($eventType === 'unsubscribe') {
             $this->unsubscribe($event, $context);
+        } elseif (Str::startsWith($eventType, 'client-')) {
+            $this->broadcastToChannel($event, $context);
         } else {
             $this->sendMessage($event, $context, [
-                'event' => 'error',
+                'event' => 'error'
             ]);
         }
     }
@@ -134,8 +138,42 @@ protected function unsubscribe(WebsocketEvent $event, Context $context): void
         ]);
     }
 
+    public function broadcastToChannel(WebsocketEvent $event, Context $context): void
+    {
+        $skipConnectionId = $event->getConnectionId();
+        $eventBody = json_decode($event->getBody(), true);
+        $channel = Arr::get($eventBody, 'channel');
+        $event = Arr::get($eventBody, 'event');
+        $payload = Arr::get($eventBody, 'data');
+        if (is_object($payload) || is_array($payload)) {
+            $payload = json_encode($payload);
+        }
+        $data = json_encode([
+            'event'=>$event,
+            'channel'=>$channel,
+            'data'=>$payload,
+        ]) ?: '';
+        $this->subscriptionRepository->getConnectionIdsForChannel($channel)
+            ->reject(fn ($connectionId) => $connectionId === $skipConnectionId)
+            ->each(fn (string $connectionId) => $this->sendMessageToConnection($connectionId, $data));
+    }
+
     public function sendMessage(WebsocketEvent $event, Context $context, array $data): void
     {
         $this->connectionRepository->sendMessage($event->getConnectionId(), json_encode($data, JSON_THROW_ON_ERROR));
     }
+
+    protected function sendMessageToConnection(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;
+        }
+    }
 }

tests/HandlerTest.php

@@ -79,6 +79,40 @@
     ], $context);
 });
 
+it('can broadcast a whisper', function () {
+    app()->instance(SubscriptionRepository::class, Mockery::mock(SubscriptionRepository::class, function ($mock) {
+        /** @var Mock $mock */
+        $mock->shouldReceive('getConnectionIdsForChannel')->withArgs(function (string $channel): bool {
+            return $channel === 'test-channel';
+        })->once()
+        ->andReturn(collect(['connection-id-1', 'connection-id-2']));
+    }));
+
+    app()->instance(ConnectionRepository::class, Mockery::mock(ConnectionRepository::class, function ($mock) {
+        /** @var Mock $mock */
+        $mock->shouldReceive('sendMessage')->withArgs(function (string $connectionId, string $data): bool {
+            return $connectionId === 'connection-id-2' and $data === '{"event":"client-test","channel":"test-channel","data":"whisper"}';
+        })->once();
+    }));
+
+    /** @var Handler $handler */
+    $handler = app(Handler::class);
+
+    $context = new Context('request-id-1', 50_000, 'function-arn', 'trace-id-1');
+
+    $handler->handle([
+        'requestContext' => [
+            'routeKey' => 'my-test-route-key',
+            'eventType' => 'MESSAGE',
+            'connectionId' => 'connection-id-1',
+            'domainName' => 'test-domain',
+            'apiId' => 'api-id-1',
+            'stage' => 'stage-test',
+        ],
+        'body' => json_encode(['event' => 'client-test', 'channel' => 'test-channel', 'data'=>'whisper']),
+    ], $context);
+});
+
 it('handles dropped connections', function () {
     $mock = new MockHandler();