Add tests for readEvents.

Author: goloroden

package-lock.json

@@ -14,8 +14,7 @@
 	},
 	"types": "./dist/index.d.ts",
 	"dependencies": {
-		"axios": "1.8.4",
-		"stream-to-async-iterator": "1.0.0"
+		"axios": "1.8.4"
 	},
 	"devDependencies": {
 		"@biomejs/biome": "1.9.4",

package.json

@@ -125,22 +125,29 @@ class Client {
 		const apiToken = this.#apiToken;
 
 		return (async function* () {
-			const response = await axios({
-				url,
+			const response = await fetch(url, {
 				method: 'post',
 				headers: {
 					authorization: `Bearer ${apiToken}`,
 					'content-type': 'application/json',
 				},
-				data: {
+				body: JSON.stringify({
 					subject,
 					options,
-				},
-				responseType: 'stream',
+				}),
 				signal: abortController.signal,
 			});
 
-			for await (const line of readNdJsonStream(response.data)) {
+			if (response.status !== 200) {
+				throw new Error(
+					`Failed to read events, got HTTP status code '${response.status}', expected '200'.`,
+				);
+			}
+			if (!response.body) {
+				throw new Error('Failed to read events.');
+			}
+
+			for await (const line of readNdJsonStream(response.body)) {
 				if (isStreamHeartbeat(line)) {
 					continue;
 				}

src/Client.ts

@@ -1,9 +1,10 @@
 import { Client } from './Client.js';
+import type { Event } from './Event.js';
 import type { EventCandidate } from './EventCandidate.js';
 import type { ObserveEventsOptions } from './ObserveEventsOptions.js';
 import type { ReadEventsOptions } from './ReadEventsOptions.js';
 import { isSubjectOnEventId } from './isSubjectOnEventId.js';
 import { isSubjectPristine } from './isSubjectPristine.js';
 
 export { Client, isSubjectPristine, isSubjectOnEventId };
-export type { EventCandidate, ReadEventsOptions, ObserveEventsOptions };
+export type { Event, EventCandidate, ReadEventsOptions, ObserveEventsOptions };

src/index.ts

@@ -3,10 +3,26 @@ import { Readable } from 'node:stream';
 import { suite, test } from 'node:test';
 import { readNdJsonStream } from './readNdJsonStream.js';
 
+function convertStreamToWebStream(stream: Readable): ReadableStream<Uint8Array> {
+	return new ReadableStream({
+		start(controller) {
+			stream.on('data', chunk => {
+				controller.enqueue(new Uint8Array(chunk));
+			});
+			stream.on('end', () => {
+				controller.close();
+			});
+			stream.on('error', err => {
+				controller.error(err);
+			});
+		},
+	});
+}
+
 suite('readNdJsonStream', (): void => {
 	test('returns an async generator that yields parsed json objects.', async (): Promise<void> => {
-		const stream = Readable.from(
-			Buffer.from('{"foo":"bar"}\n{"bar":"baz"}\n{"incomplete', 'utf-8'),
+		const stream = convertStreamToWebStream(
+			Readable.from(Buffer.from('{"foo":"bar"}\n{"bar":"baz"}\n{"incomplete', 'utf-8')),
 		);
 
 		const values: Record<string, unknown>[] = [];

src/ndjson/LinesDecoder.test.ts

@@ -1,19 +1,31 @@
-import type { Readable } from 'node:stream';
-import StreamToAsyncIterator from 'stream-to-async-iterator';
-import { LinesDecoder } from './LinesDecoder.js';
-
 const readNdJsonStream = async function* (
-	stream: Readable,
+	stream: ReadableStream<Uint8Array>,
 ): AsyncGenerator<Record<string, unknown>, void, void> {
-	const decoder = new LinesDecoder('utf-8');
+	const reader = stream.getReader();
+	const decoder = new TextDecoder('utf-8');
+	let buffer = '';
+
+	while (true) {
+		const { done, value } = await reader.read();
+		if (done) {
+			break;
+		}
+
+		buffer += decoder.decode(value, { stream: true });
 
-	for await (const chunk of new StreamToAsyncIterator<Buffer>(stream)) {
-		const lines = decoder.write(chunk);
+		let index: number;
+		while (true) {
+			index = buffer.indexOf('\n');
+			if (index < 0) {
+				break;
+			}
 
-		for (const line of lines) {
-			const parsedLine = JSON.parse(line);
+			const line = buffer.slice(0, index).trim();
+			buffer = buffer.slice(index + 1);
 
-			yield parsedLine;
+			if (line) {
+				yield JSON.parse(line);
+			}
 		}
 	}
 };

src/ndjson/LinesDecoder.ts

@@ -18,7 +18,14 @@ const blueprint: StreamCloudEvent = {
 };
 
 const isStreamCloudEvent = (line: unknown): line is StreamCloudEvent => {
-	return hasShapeOf(line, blueprint);
+	if (!hasShapeOf(line, blueprint)) {
+		return false;
+	}
+	if (line.type !== 'event') {
+		return false;
+	}
+
+	return true;
 };
 
 export { isStreamCloudEvent };

src/ndjson/readNdJsonStream.test.ts

@@ -9,7 +9,14 @@ const blueprint: StreamError = {
 };
 
 const isStreamError = (line: unknown): line is StreamError => {
-	return hasShapeOf(line, blueprint);
+	if (!hasShapeOf(line, blueprint)) {
+		return false;
+	}
+	if (line.type !== 'error') {
+		return false;
+	}
+
+	return true;
 };
 
 export { isStreamError };