Merge pull request #5 from Rabrennie/reconnect

Update to reconnect when eventsource disconnects
Rabrennie · May 23, 2023 · 08bf153 · 08bf153
2 parents 758170e + 1b5781b
commit 08bf153
Show file tree

Hide file tree

Showing 4 changed files with 36 additions and 21 deletions.
diff --git a/src/lib/room/SseClient.server.ts b/src/lib/room/SseClient.server.ts
@@ -1,11 +1,14 @@
 import { EventEmitter } from 'node:events';
 import type { RoomEvent, RoomEvents } from './events';
+import crypto from 'crypto';
 
 export class SseClient {
     eventEmitter: EventEmitter;
+    id: string;
 
     constructor() {
         this.eventEmitter = new EventEmitter();
+        this.id = crypto.randomBytes(16).toString('base64url');
     }
 
     send<T extends RoomEvent>(event: T, data: RoomEvents[T]) {

diff --git a/src/routes/room/[room]/+page.svelte b/src/routes/room/[room]/+page.svelte
@@ -8,9 +8,10 @@
     import SearchModal from './SearchModal.svelte';
     import SpinnerModal from './SpinnerModal.svelte';
     import type { PageData } from './$types';
-    import { browser } from '$app/environment';
     import { roomStore } from '$lib/stores/RoomStore';
     import { enhance } from '$app/forms';
+    import { invalidateAll } from '$app/navigation';
+    import { onMount } from 'svelte';
 
     export let data: PageData;
     roomStore.set(data.room);
@@ -19,8 +20,7 @@
 
     async function subscribe() {
         const RoomEventSource = (await import('$lib/room/RoomEventSource')).default;
-        // TODO: handle reconnecting
-        const sse = new RoomEventSource(`${window.location.pathname}/events`);
+        let sse = new RoomEventSource(`${window.location.pathname}/events`);
 
         sse.addEventListener('room:users:update', (event) => {
             console.log('myevent', event);
@@ -34,22 +34,26 @@
             eliminating = $roomStore?.choices[data.userId];
         });
 
-        return () => sse.close();
-    }
 
-    if (browser) {
-        subscribe();
+        sse.onerror = async (ev) => {
+            console.log(ev);
+            sse.close();
+            await invalidateAll();
+            subscribe();
+        };
     }
 
+    onMount(subscribe);
+
     function onSpinnerComplete() {
-    	setTimeout(() => {
+        setTimeout(() => {
             if (!eliminating) {
                 return;
             }
 
             roomStore.onEliminated({ userId: eliminating.userId });
             eliminating = undefined;
-    	}, 1000);
+        }, 1000);
     }
 
     $: winner = Object.values($roomStore?.choices ?? {}).find((c) => !c.eliminated);

diff --git a/src/routes/room/[room]/events/+server.ts b/src/routes/room/[room]/events/+server.ts
@@ -6,26 +6,34 @@ import type { RoomEvent } from '$lib/room/events';
 export const GET = (async ({ params }) => {
     const sseClient = new SseClient();
 
-    rooms.set(params.room, [...(rooms.get(params.room) ?? []), sseClient]);
+    rooms.set(params.room, { ...(rooms.get(params.room) ?? {}), [sseClient.id]: sseClient });
 
     const stream = new ReadableStream({
         start(controller) {
             sseClient.eventEmitter.on('send', (event: RoomEvent, data) => {
-                controller.enqueue(`event: ${event}\ndata:${JSON.stringify(data)}\n\n`);
+                try {
+                    controller.enqueue(`event: ${event}\ndata:${JSON.stringify(data)}\n\n`);
+                } catch {
+                    this.cancel?.();
+                }
             });
 
             sseClient.eventEmitter.on('delaySend', (event: RoomEvent, delay: number, data) => {
-                setTimeout(
-                    () => controller.enqueue(`event: ${event}\ndata:${JSON.stringify(data)}\n\n`),
-                    delay
-                );
+                try {
+                    setTimeout(
+                        () =>
+                            controller.enqueue(`event: ${event}\ndata:${JSON.stringify(data)}\n\n`),
+                        delay
+                    );
+                } catch {
+                    this.cancel?.();
+                }
             });
         },
         cancel() {
-            const clients = rooms.get(params.room) ?? [];
-            const index = clients.indexOf(sseClient);
-            if (index >= 0) {
-                clients.splice(index, 1);
+            const clients = rooms.get(params.room) ?? {};
+            if (clients[sseClient.id]) {
+                delete clients[sseClient.id];
             }
         }
     });

diff --git a/src/routes/room/[room]/events/state.server.ts b/src/routes/room/[room]/events/state.server.ts
@@ -1,11 +1,11 @@
 import type { SseClient } from '$lib/room/SseClient.server';
 import type { RoomEvent, RoomEvents } from '$lib/room/events';
 
-export const rooms = new Map<string, SseClient[]>();
+export const rooms = new Map<string, { [key: string]: SseClient }>();
 
 export const roomsState = {
     rooms,
     broadcast<T extends RoomEvent>(roomId: string, event: T, data: RoomEvents[T]) {
-        rooms.get(roomId)?.forEach((c) => c.send(event, data));
+        Object.values(rooms.get(roomId) ?? {})?.forEach((c) => c.send(event, data));
     }
 };