From 2604059961bdbe22558166962b2c67432a87e157 Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Thu, 22 Oct 2026 08:35:20 +0800 Subject: [PATCH] refactor(process): share decoded NDJSON framing --- src/ghidra/GhidraClient.ts | 16 ++--- src/ghidra/GhidraResponseBuffer.ts | 37 ----------- src/ghidra/GhidraResponseRouter.test.ts | 49 +++++++++++++++ src/hopper/HopperResponseStream.test.ts | 70 +++++++++++++++++++++ src/hopper/HopperResponseStream.ts | 20 ++---- src/process/DecodedNdjsonLineBuffer.test.ts | 33 ++++++++++ src/process/DecodedNdjsonLineBuffer.ts | 31 +++++++++ 7 files changed, 196 insertions(+), 60 deletions(-) delete mode 100644 src/ghidra/GhidraResponseBuffer.ts create mode 100644 src/ghidra/GhidraResponseRouter.test.ts create mode 100644 src/hopper/HopperResponseStream.test.ts create mode 100644 src/process/DecodedNdjsonLineBuffer.test.ts create mode 100644 src/process/DecodedNdjsonLineBuffer.ts diff --git a/src/ghidra/GhidraClient.ts b/src/ghidra/GhidraClient.ts index b8255f30b2..9761b9e6ca 100644 --- a/src/ghidra/GhidraClient.ts +++ b/src/ghidra/GhidraClient.ts @@ -20,6 +20,7 @@ import { err, ok, type Result } from "../domain/result.js"; import { silentLogger } from "../logger.js"; import type { Logger } from "pino"; import { PendingOperations } from "../process/PendingOperations.js"; +import { DecodedNdjsonLineBuffer } from "../process/DecodedNdjsonLineBuffer.js"; import { ABORTED, waitForAbortable } from "../process/AbortablePromise.js"; import { SharedStartupAttempt, @@ -50,7 +51,6 @@ import { GHIDRA_PROCESS_DIAGNOSTIC_BYTES, } from "./GhidraDiagnostics.js"; import { GhidraLaunchError, type GhidraLaunch } from "./GhidraLauncher.js"; -import { GhidraResponseBuffer } from "./GhidraResponseBuffer.js"; import { GhidraResponseRouter } from "./GhidraResponseRouter.js"; import { GhidraRequestQueue } from "./GhidraRequestQueue.js"; import { @@ -166,11 +166,11 @@ export class GhidraClient { }), protocolFailure: (message, cause) => this.#abortProtocol(message, cause), }); - readonly #responseBuffer = new GhidraResponseBuffer({ - onLine: (line) => this.#responseRouter.route(line), - }); - readonly #onSocketData = (chunk: string): void => - this.#responseBuffer.push(chunk); + readonly #responseLines = new DecodedNdjsonLineBuffer(); + readonly #onSocketData = (chunk: string): void => { + for (const line of this.#responseLines.push(chunk)) + this.#responseRouter.route(line); + }; readonly #onSocketError = (): void => { this.#failAll(this.#failure("process", "Ghidra bridge socket failed")); }; @@ -974,7 +974,7 @@ export class GhidraClient { this.#detachSocket = undefined; this.#socket?.destroy(); this.#socket = undefined; - this.#responseBuffer.reset(); + this.#responseLines.reset(); } } @@ -1119,7 +1119,7 @@ export class GhidraClient { this.#endpointPath = undefined; this.#token = undefined; this.#runId = undefined; - this.#responseBuffer.reset(); + this.#responseLines.reset(); if (forcedStopFailure !== undefined) throw forcedStopFailure.cause; for (const failure of this.#cleanupFailures) failure.resolveCleanup(); this.#cleanupFailures.clear(); diff --git a/src/ghidra/GhidraResponseBuffer.ts b/src/ghidra/GhidraResponseBuffer.ts deleted file mode 100644 index 1d48c76db0..0000000000 --- a/src/ghidra/GhidraResponseBuffer.ts +++ /dev/null @@ -1,37 +0,0 @@ -/** Callbacks for one newline-delimited Ghidra response stream. */ -export interface GhidraResponseBufferOptions { - readonly onLine: (line: string) => void; -} - -/** Splits fragmented UTF-8 socket data into response lines. */ -export class GhidraResponseBuffer { - readonly #options: GhidraResponseBufferOptions; - #buffer = ""; - #fragments: string[] = []; - - constructor(options: GhidraResponseBufferOptions) { - this.#options = options; - } - - /** Consume one decoded socket chunk. */ - push(chunk: string): void { - this.#fragments.push(chunk); - if (!chunk.includes("\n")) return; - this.#buffer += this.#fragments.join(""); - this.#fragments = []; - let newline = this.#buffer.indexOf("\n"); - while (newline >= 0) { - const encodedLine = this.#buffer.slice(0, newline); - this.#buffer = this.#buffer.slice(newline + 1); - const line = encodedLine.trim(); - if (line.length > 0) this.#options.onLine(line); - newline = this.#buffer.indexOf("\n"); - } - } - - /** Drop any incomplete line when a session closes. */ - reset(): void { - this.#buffer = ""; - this.#fragments = []; - } -} diff --git a/src/ghidra/GhidraResponseRouter.test.ts b/src/ghidra/GhidraResponseRouter.test.ts new file mode 100644 index 0000000000..e4ecda4f02 --- /dev/null +++ b/src/ghidra/GhidraResponseRouter.test.ts @@ -0,0 +1,49 @@ +import { expect, it, vi } from "vitest"; + +import type { JsonValue } from "../domain/jsonValue.js"; +import { err, ok, type Result } from "../domain/result.js"; +import { DecodedNdjsonLineBuffer } from "../process/DecodedNdjsonLineBuffer.js"; +import { PendingOperations } from "../process/PendingOperations.js"; +import { GhidraResponseRouter } from "./GhidraResponseRouter.js"; +import { GhidraSessionError } from "./GhidraSessionError.js"; + +it("routes later Ghidra response lines after an unknown response id", async () => { + const pending = new PendingOperations< + number, + Result + >(); + const protocolFailure = vi.fn(); + const router = new GhidraResponseRouter({ + pending, + nextId: () => 2, + remoteFailure: (failure) => + new GhidraSessionError( + "remote", + failure.message, + {}, + { + remoteCode: failure.code, + }, + ), + protocolFailure: (message, cause) => + protocolFailure( + new GhidraSessionError("protocol", message, {}, { cause }), + ), + }); + const response = pending.wait(1, { + cancelledValue: () => + err(new GhidraSessionError("cancelled", "request cancelled")), + }); + const lines = new DecodedNdjsonLineBuffer(); + + for (const line of lines.push( + '{"id":2,"ok":true,"result":null}\n{"id":1,"ok":true,"result":"settled"}\n', + )) + router.route(line); + + expect(protocolFailure).toHaveBeenCalledOnce(); + expect(protocolFailure).toHaveBeenCalledWith( + expect.objectContaining({ kind: "protocol" }), + ); + await expect(response).resolves.toEqual(ok("settled")); +}); diff --git a/src/hopper/HopperResponseStream.test.ts b/src/hopper/HopperResponseStream.test.ts new file mode 100644 index 0000000000..0a9de6f041 --- /dev/null +++ b/src/hopper/HopperResponseStream.test.ts @@ -0,0 +1,70 @@ +import { describe, expect, it, vi } from "vitest"; + +import { HopperResponseStream } from "./HopperResponseStream.js"; + +describe("Hopper response stream", () => { + it("stops on malformed input and resumes with buffered response lines", () => { + const accepted: number[] = []; + const abort = vi.fn(); + const stream = new HopperResponseStream({ + accept: (message) => { + accepted.push(message.id); + return true; + }, + hasQueued: () => false, + nextRequestId: () => 4, + abort, + }); + + stream.push('not json\n{"id":2,"result":null}\n'); + expect(accepted).toEqual([]); + expect(abort).toHaveBeenCalledOnce(); + + stream.push('{"id":3,"result":null}\n'); + expect(accepted).toEqual([2, 3]); + expect(abort).toHaveBeenCalledOnce(); + }); + + it("stops on a future response id and retains later complete lines", () => { + const accepted: number[] = []; + const abort = vi.fn(); + const stream = new HopperResponseStream({ + accept: (message) => { + if (message.id >= 3) return false; + accepted.push(message.id); + return true; + }, + hasQueued: () => false, + nextRequestId: () => 3, + abort, + }); + + stream.push('{"id":3,"result":null}\n{"id":1,"result":null}\n'); + expect(accepted).toEqual([]); + expect(abort).toHaveBeenCalledWith( + "Hopper returned an unknown response id", + ); + + stream.push('{"id":2,"result":null}\n'); + expect(accepted).toEqual([1, 2]); + expect(abort).toHaveBeenCalledOnce(); + }); + + it("discards incomplete lines on reset", () => { + const accepted: number[] = []; + const stream = new HopperResponseStream({ + accept: (message) => { + accepted.push(message.id); + return true; + }, + hasQueued: () => false, + nextRequestId: () => 2, + abort: vi.fn(), + }); + + stream.push('{"id":1,"result":null}'); + stream.reset(); + stream.push('{"id":1,"result":null}\n'); + expect(accepted).toEqual([1]); + }); +}); diff --git a/src/hopper/HopperResponseStream.ts b/src/hopper/HopperResponseStream.ts index 0f3c294eab..fd6b184d6f 100644 --- a/src/hopper/HopperResponseStream.ts +++ b/src/hopper/HopperResponseStream.ts @@ -2,6 +2,7 @@ import { parseBridgeMessageLine, type HopperBridgeMessage, } from "./protocol.js"; +import { DecodedNdjsonLineBuffer } from "../process/DecodedNdjsonLineBuffer.js"; export interface HopperResponseStreamOptions { readonly accept: (message: HopperBridgeMessage) => boolean; @@ -12,28 +13,17 @@ export interface HopperResponseStreamOptions { /** Incrementally validates Hopper NDJSON response frames. */ export class HopperResponseStream { - #buffer = ""; - #fragments: string[] = []; + readonly #lines = new DecodedNdjsonLineBuffer(); constructor(readonly options: HopperResponseStreamOptions) {} push(chunk: string): void { - this.#fragments.push(chunk); - if (!chunk.includes("\n")) return; - this.#buffer += this.#fragments.join(""); - this.#fragments = []; - let newline = this.#buffer.indexOf("\n"); - while (newline >= 0) { - const line = this.#buffer.slice(0, newline).trim(); - this.#buffer = this.#buffer.slice(newline + 1); - if (line.length > 0 && !this.#acceptLine(line)) return; - newline = this.#buffer.indexOf("\n"); - } + for (const line of this.#lines.push(chunk)) + if (!this.#acceptLine(line)) break; } reset(): void { - this.#buffer = ""; - this.#fragments = []; + this.#lines.reset(); } #acceptLine(line: string): boolean { diff --git a/src/process/DecodedNdjsonLineBuffer.test.ts b/src/process/DecodedNdjsonLineBuffer.test.ts new file mode 100644 index 0000000000..088b00bd55 --- /dev/null +++ b/src/process/DecodedNdjsonLineBuffer.test.ts @@ -0,0 +1,33 @@ +import { describe, expect, it } from "vitest"; + +import { DecodedNdjsonLineBuffer } from "./DecodedNdjsonLineBuffer.js"; + +describe("decoded NDJSON line buffer", () => { + it("joins fragments and yields trimmed nonblank complete lines", () => { + const lines = new DecodedNdjsonLineBuffer(); + + expect([...lines.push(" first ")]).toEqual([]); + expect([...lines.push("\n \t\n second \n unfinished")]).toEqual([ + "first", + "second", + ]); + expect([...lines.push(" line\n")]).toEqual(["unfinished line"]); + }); + + it("retains unconsumed complete lines and reset discards them", () => { + const lines = new DecodedNdjsonLineBuffer(); + const iterator = lines.push('invalid\n{"id":2}\n'); + + expect(iterator.next()).toEqual({ value: "invalid", done: false }); + iterator.return(undefined); + expect([...lines.push('{"id":3}\n')]).toEqual(['{"id":2}', '{"id":3}']); + + const nextIterator = lines.push( + "discard me\nwill also be discarded\nkeep partial", + ); + expect(nextIterator.next()).toEqual({ value: "discard me", done: false }); + nextIterator.return(undefined); + lines.reset(); + expect([...lines.push("after reset\n")]).toEqual(["after reset"]); + }); +}); diff --git a/src/process/DecodedNdjsonLineBuffer.ts b/src/process/DecodedNdjsonLineBuffer.ts new file mode 100644 index 0000000000..4e601deebd --- /dev/null +++ b/src/process/DecodedNdjsonLineBuffer.ts @@ -0,0 +1,31 @@ +/** Incrementally splits decoded newline-delimited JSON socket data into lines. */ +export class DecodedNdjsonLineBuffer { + #buffer = ""; + #fragments: string[] = []; + + /** + * Consume a chunk and yield each complete, trimmed, nonblank line. + * Unconsumed complete lines remain buffered if iteration stops early. + */ + *push(chunk: string): Generator { + this.#fragments.push(chunk); + if (!chunk.includes("\n")) return; + this.#buffer += this.#fragments.join(""); + this.#fragments = []; + + let newline = this.#buffer.indexOf("\n"); + while (newline >= 0) { + const encodedLine = this.#buffer.slice(0, newline); + this.#buffer = this.#buffer.slice(newline + 1); + const line = encodedLine.trim(); + if (line.length > 0) yield line; + newline = this.#buffer.indexOf("\n"); + } + } + + /** Drop incomplete and unconsumed lines when a session closes. */ + reset(): void { + this.#buffer = ""; + this.#fragments = []; + } +}