refactor into a testable JsonlStream interface

Author: skyrpex

src/test/jsonl-stream.test.ts

@@ -0,0 +1,118 @@
+import * as assert from "node:assert";
+
+import { LogLevel } from "vscode";
+import type { LogOutputChannel, Event } from "vscode";
+
+import { createJsonlStream } from "../utils/jsonl-stream.js";
+
+const setup = () => {
+	const mockOutputChannel: LogOutputChannel = {
+		name: "test",
+		logLevel: LogLevel.Info,
+		onDidChangeLogLevel: {} as Event<LogLevel>,
+		append: () => {},
+		appendLine: () => {},
+		replace: () => {},
+		clear: () => {},
+		show: () => {},
+		hide: () => {},
+		dispose: () => {},
+		trace: () => {},
+		debug: () => {},
+		info: () => {},
+		warn: () => {},
+		error: () => {},
+	};
+
+	const jsonlStream = createJsonlStream(mockOutputChannel);
+
+	const callbackCalls: unknown[] = [];
+	const callback = (data: unknown) => {
+		callbackCalls.push(data);
+	};
+	jsonlStream.on(callback);
+
+	return {
+		jsonlStream,
+		callbackCalls,
+	};
+};
+
+suite("JSONL Streams", () => {
+	test("should parse and emit complete JSONL messages", () => {
+		const { jsonlStream, callbackCalls } = setup();
+
+		// Test with multiple JSON objects in a single write
+		const testData = Buffer.from('{"key1":"value1"}\n{"key2":"value2"}\n');
+
+		jsonlStream.write(testData);
+
+		assert.strictEqual(callbackCalls.length, 2);
+		assert.deepStrictEqual(callbackCalls[0], { key1: "value1" });
+		assert.deepStrictEqual(callbackCalls[1], { key2: "value2" });
+	});
+
+	test("should handle incomplete JSONL messages across multiple writes", () => {
+		const { jsonlStream, callbackCalls } = setup();
+
+		// First write with partial message
+		const firstChunk = Buffer.from('{"key":"value');
+		jsonlStream.write(firstChunk);
+
+		// Shouldn't emit anything yet
+		assert.strictEqual(callbackCalls.length, 0);
+
+		// Complete the message in second write
+		const secondChunk = Buffer.from('1"}\n');
+		jsonlStream.write(secondChunk);
+
+		// Now it should emit the complete message
+		assert.strictEqual(callbackCalls.length, 1);
+		assert.deepStrictEqual(callbackCalls[0], { key: "value1" });
+	});
+
+	test("should handle multiple messages in chunks", () => {
+		const { jsonlStream, callbackCalls } = setup();
+
+		// Write first message and part of second
+		const firstChunk = Buffer.from('{"first":1}\n{"second":');
+		jsonlStream.write(firstChunk);
+
+		// First message should be emitted
+		assert.strictEqual(callbackCalls.length, 1);
+		assert.deepStrictEqual(callbackCalls[0], { first: 1 });
+
+		// Complete second message and add third
+		const secondChunk = Buffer.from('2}\n{"third":3}\n');
+		jsonlStream.write(secondChunk);
+
+		// Should have all three messages now
+		assert.strictEqual(callbackCalls.length, 3);
+		assert.deepStrictEqual(callbackCalls[1], { second: 2 });
+		assert.deepStrictEqual(callbackCalls[2], { third: 3 });
+	});
+
+	test("should ignore invalid JSON lines", () => {
+		const { jsonlStream, callbackCalls } = setup();
+
+		const testData = Buffer.from('not json\n{"valid":true}\n{invalid}\n');
+
+		jsonlStream.write(testData);
+
+		// Should only emit the valid JSON object
+		assert.strictEqual(callbackCalls.length, 1);
+		assert.deepStrictEqual(callbackCalls[0], { valid: true });
+	});
+
+	test("should handle empty lines", () => {
+		const { jsonlStream, callbackCalls } = setup();
+
+		const testData = Buffer.from('\n\n{"key":"value"}\n\n');
+
+		jsonlStream.write(testData);
+
+		// Should only emit the valid JSON object
+		assert.strictEqual(callbackCalls.length, 1);
+		assert.deepStrictEqual(callbackCalls[0], { key: "value" });
+	});
+});

src/utils/container-status.ts

@@ -4,6 +4,7 @@ import type { Disposable, LogOutputChannel } from "vscode";
 import * as z from "zod/v4-mini";
 
 import { createEmitter } from "./emitter.ts";
+import { createJsonlStream } from "./jsonl-stream.ts";
 
 export type ContainerStatus = "running" | "stopping" | "stopped";
 
@@ -63,14 +64,6 @@ const DockerEventsSchema = z.object({
 	}),
 });
 
-function safeJsonParse(text: string): unknown {
-	try {
-		return JSON.parse(text);
-	} catch {
-		return undefined;
-	}
-}
-
 function listenToContainerStatus(
 	containerName: string,
 	outputChannel: LogOutputChannel,
@@ -130,41 +123,33 @@ function listenToContainerStatus(
 				throw new Error("Failed to get stdout from docker events process");
 			}
 
-			let buffer = "";
-			dockerEvents.stdout.on("data", (data: Buffer) => {
-				buffer += data.toString();
-
-				// Process all complete lines
-				let newlineIndex = buffer.indexOf("\n");
-				while (newlineIndex !== -1) {
-					const line = buffer.substring(0, newlineIndex).trim();
-					buffer = buffer.substring(newlineIndex + 1);
-
-					const json = safeJsonParse(line);
-					const parsed = DockerEventsSchema.safeParse(json);
-					if (!parsed.success) {
-						continue;
-					}
-
-					if (parsed.data.Actor.Attributes.name !== containerName) {
-						continue;
-					}
+			const jsonlStream = createJsonlStream(outputChannel);
+			jsonlStream.on((json) => {
+				const parsed = DockerEventsSchema.safeParse(json);
+				if (!parsed.success) {
+					return;
+				}
 
-					switch (parsed.data.Action) {
-						case "start":
-							onStatusChange("running");
-							break;
-						case "kill":
-							onStatusChange("stopping");
-							break;
-						case "die":
-							onStatusChange("stopped");
-							break;
-					}
+				if (parsed.data.Actor.Attributes.name !== containerName) {
+					return;
+				}
 
-					newlineIndex = buffer.indexOf("\n");
+				switch (parsed.data.Action) {
+					case "start":
+						onStatusChange("running");
+						break;
+					case "kill":
+						onStatusChange("stopping");
+						break;
+					case "die":
+						onStatusChange("stopped");
+						break;
 				}
 			});
+
+			dockerEvents.stdout.on("data", (data: Buffer) => {
+				jsonlStream.write(data);
+			});
 		} catch (error) {
 			// If we can't spawn the process, try again after a delay
 			scheduleRestart();

src/utils/jsonl-stream.ts

@@ -0,0 +1,44 @@
+import type { LogOutputChannel } from "vscode";
+
+import type { Callback } from "./emitter.ts";
+import { createEmitter } from "./emitter.ts";
+
+interface JsonlStream {
+  write(data: Buffer): void;
+  on(callback: Callback<unknown>): void;
+}
+
+function safeJsonParse(text: string): unknown {
+	try {
+		return JSON.parse(text);
+	} catch {
+		return undefined;
+	}
+}
+
+export const createJsonlStream = (outputChannel: LogOutputChannel): JsonlStream => {
+  const emitter = createEmitter(outputChannel)
+  let buffer = "";
+  return {
+    write(data) {
+      buffer += data.toString();
+
+				// Process all complete lines
+				let newlineIndex = buffer.indexOf("\n");
+				while (newlineIndex !== -1) {
+					const line = buffer.substring(0, newlineIndex).trim();
+					buffer = buffer.substring(newlineIndex + 1);
+
+					const json = safeJsonParse(line);
+          if (json) {
+            void emitter.emit(json)
+          }
+
+					newlineIndex = buffer.indexOf("\n");
+				}
+    },
+    on(callback) {
+      emitter.on(callback)
+    },
+  }
+};