wip

src/async/deserializeAsync.ts

@@ -11,11 +11,17 @@ import {
 } from "../sync/syncTypes.js";
 import { TsonStreamInterruptedError } from "./asyncErrors.js";
 import {
+	BrandSerialized,
 	TsonAsyncIndex,
 	TsonAsyncOptions,
 	TsonAsyncStringifierIterable,
 	TsonAsyncType,
 } from "./asyncTypes.js";
+import {
+	createReadableStream,
+	mapIterable,
+	readableStreamToAsyncIterable,
+} from "./iterableUtils.js";
 import { TsonAsyncValueTuple } from "./serializeAsync.js";
 
 type WalkFn = (value: unknown) => unknown;
@@ -259,3 +265,39 @@ export function createTsonParseAsync(opts: TsonAsyncOptions): TsonParseAsync {
 		return await instance(tsonIterable, opts ?? {});
 	}) as TsonParseAsync;
 }
+
+// export function createTsonParseEventSource(opts: TsonAsyncOptions) {
+// 	const instance = createTsonDeserializer(opts);
+
+// 	return async <TValue = unknown>(
+// 		url: string,
+// 		parseOpts?: TsonParseAsyncOptions & {
+// 			abortSignal: AbortSignal;
+// 		},
+// 	) => {
+// 		const [stream, controller] = createReadableStream<string>();
+// 		const eventSource = new EventSource(url);
+
+// 		const onAbort = () => {
+// 			eventSource.close();
+// 			controller.close();
+// 			parseOpts?.abortSignal.removeEventListener("abort", onAbort);
+// 		};
+
+// 		parseOpts?.abortSignal.addEventListener("abort", onAbort);
+
+// 		eventSource.onmessage = (msg) => {
+// 			// eslint-disable-next-line @typescript-eslint/no-unsafe-argument
+// 			controller.enqueue(msg.data);
+// 		};
+
+// 		const iterable = mapIterable(
+// 			readableStreamToAsyncIterable(stream),
+// 			(msg) => {
+// 				return JSON.parse(msg) as TsonAsyncValueTuple | TsonSerialized;
+// 			},
+// 		);
+
+// 		return (await instance(iterable, parseOpts ?? {})) as TValue;
+// 	};
+// }

src/async/iterableUtils.ts

@@ -1,3 +1,5 @@
+import { assert } from "../internals/assert";
+
 export async function* readableStreamToAsyncIterable<T>(
 	stream: ReadableStream<T>,
 ): AsyncIterable<T> {
@@ -31,3 +33,17 @@ export async function* mapIterable<T, TValue>(
 		yield fn(value);
 	}
 }
+
+export function createReadableStream<TValue>() {
+	let controller: ReadableStreamDefaultController<TValue> =
+		null as unknown as ReadableStreamDefaultController<TValue>;
+	const stream = new ReadableStream<TValue>({
+		start(c) {
+			controller = c;
+		},
+	});
+
+	assert(controller, `Could not find controller - this is a bug`);
+
+	return [stream, controller] as const;
+}

src/async/sse.test.ts

@@ -25,7 +25,7 @@ test("SSE response test", async () => {
 		};
 	}
 
-	type MockObj = ReturnType<typeof createMockObj>;
+	// type MockObj = ReturnType<typeof createMockObj>;
 
 	// ------------- server -------------------
 	const opts = {
@@ -53,7 +53,7 @@ test("SSE response test", async () => {
 	});
 
 	// ------------- client -------------------
-	const tson = createTsonAsync(opts);
+	// const tson = createTsonAsync(opts);
 
 	// do a streamed fetch request
 	const sse = new EventSource(server.url);
@@ -82,3 +82,5 @@ test("SSE response test", async () => {
 		]
 	`);
 });
+
+test.todo("parse SSE response");