wip

Author: skyrpex

src/test/jsonl-stream.test.ts

@@ -3,34 +3,16 @@ import * as assert from "node:assert";
 import { LogLevel } from "vscode";
 import type { LogOutputChannel, Event } from "vscode";
 
-import { createJsonlStream } from "../utils/jsonl-stream.js";
+import { JsonlStream } 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 jsonlStream = new JsonlStream();
 
 	const callbackCalls: unknown[] = [];
 	const callback = (data: unknown) => {
 		callbackCalls.push(data);
 	};
-	jsonlStream.on(callback);
+	jsonlStream.onJson(callback);
 
 	return {
 		jsonlStream,

src/utils/container-status.ts

@@ -4,7 +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";
+import { JsonlStream } from "./jsonl-stream.ts";
 
 export type ContainerStatus = "running" | "stopping" | "stopped";
 
@@ -123,8 +123,8 @@ function listenToContainerStatus(
 				throw new Error("Failed to get stdout from docker events process");
 			}
 
-			const jsonlStream = createJsonlStream(outputChannel);
-			jsonlStream.on((json) => {
+			const jsonlStream = new JsonlStream();
+			jsonlStream.onJson((json) => {
 				const parsed = DockerEventsSchema.safeParse(json);
 				if (!parsed.success) {
 					return;
@@ -134,6 +134,8 @@ function listenToContainerStatus(
 					return;
 				}
 
+				outputChannel.debug(`[container.status]: ${parsed.data.Action}`);
+
 				switch (parsed.data.Action) {
 					case "start":
 						onStatusChange("running");
@@ -147,9 +149,7 @@ function listenToContainerStatus(
 				}
 			});
 
-			dockerEvents.stdout.on("data", (data: Buffer) => {
-				jsonlStream.write(data);
-			});
+			dockerEvents.stdout.pipe(jsonlStream);
 		} catch (error) {
 			// If we can't spawn the process, try again after a delay
 			scheduleRestart();

src/utils/jsonl-stream.ts

@@ -1,44 +1,52 @@
-import type { LogOutputChannel } from "vscode";
+import { Writable } from "node:stream";
 
-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 {
+/**
+ * Safely parses a JSON string, returning null if parsing fails.
+ * @param str - The JSON string to parse.
+ * @returns The parsed object or null if invalid.
+ */
+export function safeJsonParse(str: string): unknown {
 	try {
-		return JSON.parse(text);
+		return JSON.parse(str);
 	} catch {
 		return undefined;
 	}
 }
 
-export const createJsonlStream = (outputChannel: LogOutputChannel): JsonlStream => {
-  const emitter = createEmitter(outputChannel)
-  let buffer = "";
-  return {
-    write(data) {
-      buffer += data.toString();
+/**
+ * Writable stream that buffers data until a newline,
+ * parses each line as JSON, and emits the parsed object.
+ */
+export class JsonlStream extends Writable {
+	constructor() {
+		let buffer = "";
+		super({
+			write: (chunk, _encoding, callback) => {
+				buffer += String(chunk);
 
-				// 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)
-          }
+					if (json !== null) {
+						this.emit("json", json);
+					}
 
 					newlineIndex = buffer.indexOf("\n");
 				}
-    },
-    on(callback) {
-      emitter.on(callback)
-    },
-  }
-};
+
+				callback();
+			},
+		});
+	}
+
+	/**
+	 * Registers a listener for parsed JSON objects.
+	 * @param listener - Function called with each parsed object.
+	 */
+	onJson(callback: (json: unknown) => void) {
+		this.on("json", callback);
+	}
+}