cool

trpc · Oct 13, 2023 · d7c501a · d7c501a
1 parent 8202201
commit d7c501a
Show file tree

Hide file tree

Showing 6 changed files with 60 additions and 49 deletions.
diff --git a/examples/sse/src/index.html b/examples/sse/src/index.html
@@ -1,16 +1,18 @@
-<h1>SSE test</h1>
+<html>
+	<h1>SSE test</h1>
 
-See log output / inspector tools
+	See log output / inspector tools
 
-<script>
-	const eventSource = new EventSource("/sse");
-	eventSource.onopen = () => {
-		console.log("opened");
-	};
-	eventSource.onmessage = (msg) => {
-		console.log("message", msg);
-	};
-	eventSource.onerror = (source, err) => {
-		console.log({ source, err });
-	};
-</script>
+	<script>
+		const eventSource = new EventSource("/sse");
+		eventSource.onopen = () => {
+			console.log("opened");
+		};
+		eventSource.onmessage = (msg) => {
+			console.log("message", msg);
+		};
+		eventSource.onerror = (source, err) => {
+			console.error({ source, err });
+		};
+	</script>
+</html>
diff --git a/package.json b/package.json
@@ -61,6 +61,7 @@
 		"eslint-plugin-regexp": "^1.15.0",
 		"eslint-plugin-vitest": "^0.3.1",
 		"eslint-plugin-yml": "^1.9.0",
+		"event-source-polyfill": "^1.0.31",
 		"jsonc-eslint-parser": "^2.3.0",
 		"knip": "^2.31.0",
 		"markdownlint": "^0.31.1",

diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
diff --git a/src/async/serializeAsync.ts b/src/async/serializeAsync.ts
@@ -12,6 +12,7 @@ import {
 	TsonTypeTesterCustom,
 	TsonTypeTesterPrimitive,
 } from "../sync/syncTypes.js";
+import { TsonStreamInterruptedError } from "./asyncErrors.js";
 import {
 	BrandSerialized,
 	TsonAsyncIndex,
@@ -268,7 +269,9 @@ export function createTsonSSEResponse(opts: TsonAsyncOptions) {
 				controller.enqueue(`data: ${JSON.stringify(chunk)}\n\n`);
 			}
 
-			controller.close();
+			controller.error(
+				new TsonStreamInterruptedError(new Error("SSE stream ended")),
+			);
 		}
 
 		iterate().catch((err) => {

diff --git a/src/async/sse.test.ts b/src/async/sse.test.ts
@@ -1,34 +1,19 @@
-import { expect, test, vitest } from "vitest";
+/* eslint-disable @typescript-eslint/no-unnecessary-condition */
+import { EventSourcePolyfill, NativeEventSource } from "event-source-polyfill";
+import { expect, test } from "vitest";
+global.EventSource = NativeEventSource || EventSourcePolyfill;
 
-import {
-	TsonAsyncOptions,
-	TsonParseAsyncOptions,
-	TsonType,
-	createTsonParseAsync,
-	tsonAsyncIterable,
-	tsonBigint,
-	tsonPromise,
-} from "../index.js";
-import { assert } from "../internals/assert.js";
-import {
-	createDeferred,
-	createTestServer,
-	sleep,
-	waitError,
-	waitFor,
-} from "../internals/testUtils.js";
-import { TsonSerialized } from "../sync/syncTypes.js";
+import { TsonAsyncOptions, tsonAsyncIterable, tsonPromise } from "../index.js";
+import { createTestServer, sleep } from "../internals/testUtils.js";
 import { createTsonAsync } from "./createTsonAsync.js";
-import { mapIterable, readableStreamToAsyncIterable } from "./iterableUtils.js";
 
 test("SSE response test", async () => {
 	function createMockObj() {
 		async function* generator() {
-			for (let i = 0; i < 10; i++) {
-				yield i;
-				await sleep(1);
-
-				await sleep(1);
+			let i = 0;
+			while (true) {
+				yield i++;
+				await sleep(100);
 			}
 		}
 
@@ -44,11 +29,12 @@ test("SSE response test", async () => {
 
 	// ------------- server -------------------
 	const opts = {
+		nonce: () => "__tson",
 		types: [tsonPromise, tsonAsyncIterable],
 	} satisfies TsonAsyncOptions;
 
 	const server = await createTestServer({
-		handleRequest: async (_req, res) => {
+		handleRequest: async (req, res) => {
 			const tson = createTsonAsync(opts);
 
 			const obj = createMockObj();
@@ -72,14 +58,27 @@ test("SSE response test", async () => {
 	// do a streamed fetch request
 	const sse = new EventSource(server.url);
 
-	let messages: MessageEvent[];
-	await new Promise(() => {
+	const messages: MessageEvent["data"][] = [];
+	await new Promise<void>((resolve) => {
 		sse.onmessage = (msg) => {
-			messages.push(msg);
-		};
+			// console.log(sse.readyState);
+			// console.log({ msg });
+			messages.push(msg.data);
 
-		sse.addEventListener("error", () => {
-			console.error("error");
-		});
+			if (messages.length === 5) {
+				sse.close();
+				resolve();
+			}
+		};
 	});
+
+	expect(messages).toMatchInlineSnapshot(`
+		[
+		  "{\\"json\\":{\\"foo\\":\\"bar\\",\\"iterable\\":[\\"AsyncIterable\\",0,\\"__tson\\"],\\"promise\\":[\\"Promise\\",1,\\"__tson\\"],\\"rejectedPromise\\":[\\"Promise\\",2,\\"__tson\\"]},\\"nonce\\":\\"__tson\\"}",
+		  "[0,[0,0]]",
+		  "[1,[0,42]]",
+		  "[2,[1,{}]]",
+		  "[0,[0,1]]",
+		]
+	`);
 });
diff --git a/src/internals/testUtils.ts b/src/internals/testUtils.ts
@@ -51,7 +51,6 @@ export async function createTestServer(opts: {
 		const server = http.createServer((req, res) => {
 			Promise.resolve(opts.handleRequest(req, res)).catch((err) => {
 				console.error(err);
-				res.statusCode = 500;
 				res.end();
 			});
 		});