From 03e54a1a21bcc5ce1408a79a874451fcdc44f961 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sat, 2 May 2026 14:56:35 +0800 Subject: [PATCH] refactor(server): extract terminal stream router from daemon-client (#665) Move slot map, listener fanout, binary input encoding, and stream-frame decoding out of daemon-client.ts into TerminalStreamRouter so the 4.5k-line client no longer owns that bookkeeping. Public DaemonClient API and on-the-wire behavior unchanged. --- packages/server/src/client/daemon-client.ts | 130 ++---------------- .../src/client/terminal-stream-router.ts | 122 ++++++++++++++++ 2 files changed, 136 insertions(+), 116 deletions(-) create mode 100644 packages/server/src/client/terminal-stream-router.ts diff --git a/packages/server/src/client/daemon-client.ts b/packages/server/src/client/daemon-client.ts index 4a1d56cb7..6068c1fdf 100644 --- a/packages/server/src/client/daemon-client.ts +++ b/packages/server/src/client/daemon-client.ts @@ -59,7 +59,6 @@ import type { ListTerminalsResponse, CreateTerminalResponse, SubscribeTerminalResponse, - TerminalState, CloseItemsResponse, KillTerminalResponse, CaptureTerminalResponse, @@ -83,14 +82,10 @@ import { isRelayClientWebSocketUrl } from "../shared/daemon-endpoints.js"; import { asUint8Array, decodeFileTransferFrame, - decodeTerminalSnapshotPayload, decodeTerminalStreamFrame, FileTransferOpcode, - encodeTerminalResizePayload, - encodeTerminalStreamFrame, TerminalStreamOpcode, type FileTransferFrame, - type TerminalStreamFrame, } from "../shared/binary-frames/index.js"; import { createRelayE2eeTransportFactory, @@ -99,12 +94,12 @@ import { defaultWebSocketFactory, describeTransportClose, describeTransportError, - encodeUtf8String, type DaemonTransport, type DaemonTransportFactory, type WebSocketFactory, } from "./daemon-client-transport.js"; import { DaemonClientRuntimeMetrics } from "./daemon-client-runtime-metrics.js"; +import { TerminalStreamRouter, type TerminalStreamEvent } from "./terminal-stream-router.js"; export interface Logger { debug(obj: object, msg?: string): void; @@ -146,9 +141,7 @@ export type { WebSocketLike, } from "./daemon-client-transport.js"; -export type TerminalStreamEvent = - | { terminalId: string; type: "output"; data: Uint8Array } - | { terminalId: string; type: "snapshot"; state: TerminalState }; +export type { TerminalStreamEvent }; export type ConnectionState = | { status: "idle" } @@ -773,12 +766,10 @@ export class DaemonClient { } >(); private terminalDirectorySubscriptions = new Set(); - private terminalSlots = new Map(); - private slotTerminals = new Map(); + private readonly terminalStreams = new TerminalStreamRouter(); private pendingBinaryFileReads = new Map(); private activeBinaryFileTransfers = new Map(); private completedBinaryFileReads = new Map(); - private readonly terminalStreamListeners = new Set<(event: TerminalStreamEvent) => void>(); private logger: Logger; private pendingSendQueue: PendingSend[] = []; private readonly logConnectionPath: "direct" | "relay"; @@ -1049,7 +1040,7 @@ export class DaemonClient { this.disposeTransport(1000, "Client closed"); this.clearWaiters(new Error("Daemon client closed")); this.rejectPendingSendQueue(new Error("Daemon client closed")); - this.clearTerminalSlots(); + this.terminalStreams.clearSlots(); this.lastServerInfoMessage = null; if (this.runtimeMetricsInterval) { clearInterval(this.runtimeMetricsInterval); @@ -3485,13 +3476,13 @@ export class DaemonClient { options: { skipQueue: true }, }); if (payload.error === null) { - this.setTerminalSlot(terminalId, payload.slot); + this.terminalStreams.setSlot(terminalId, payload.slot); } return payload; } unsubscribeTerminal(terminalId: string): void { - this.removeTerminalSlot(terminalId); + this.terminalStreams.removeTerminal(terminalId); this.sendSessionMessage({ type: "unsubscribe_terminal_request", terminalId, @@ -3499,31 +3490,10 @@ export class DaemonClient { } sendTerminalInput(terminalId: string, message: TerminalInput["message"]): void { - const slot = this.terminalSlots.get(terminalId); - if (typeof slot === "number") { - if (message.type === "input") { - this.sendBinaryFrame( - encodeTerminalStreamFrame({ - opcode: TerminalStreamOpcode.Input, - slot, - payload: encodeUtf8String(message.data), - }), - ); - return; - } - if (message.type === "resize") { - this.sendBinaryFrame( - encodeTerminalStreamFrame({ - opcode: TerminalStreamOpcode.Resize, - slot, - payload: encodeTerminalResizePayload({ - rows: message.rows, - cols: message.cols, - }), - }), - ); - return; - } + const frame = this.terminalStreams.encodeInput(terminalId, message); + if (frame) { + this.sendBinaryFrame(frame); + return; } this.sendSessionMessage({ type: "terminal_input", @@ -3850,10 +3820,7 @@ export class DaemonClient { } onTerminalStreamEvent(handler: (event: TerminalStreamEvent) => void): () => void { - this.terminalStreamListeners.add(handler); - return () => { - this.terminalStreamListeners.delete(handler); - }; + return this.terminalStreams.onEvent(handler); } async waitForTerminalStreamEvent( @@ -3885,37 +3852,6 @@ export class DaemonClient { return requestId ?? crypto.randomUUID(); } - private setTerminalSlot(terminalId: string, slot: number): void { - const existingTerminalId = this.slotTerminals.get(slot); - if (existingTerminalId && existingTerminalId !== terminalId) { - this.terminalSlots.delete(existingTerminalId); - } - - const existingSlot = this.terminalSlots.get(terminalId); - if (typeof existingSlot === "number" && existingSlot !== slot) { - this.slotTerminals.delete(existingSlot); - } - - this.terminalSlots.set(terminalId, slot); - this.slotTerminals.set(slot, terminalId); - } - - private removeTerminalSlot(terminalId: string): void { - const slot = this.terminalSlots.get(terminalId); - if (typeof slot !== "number") { - return; - } - this.terminalSlots.delete(terminalId); - if (this.slotTerminals.get(slot) === terminalId) { - this.slotTerminals.delete(slot); - } - } - - private clearTerminalSlots(): void { - this.terminalSlots.clear(); - this.slotTerminals.clear(); - } - getLastServerInfoMessage(): ServerInfoStatusPayload | null { return this.lastServerInfoMessage; } @@ -4066,7 +4002,7 @@ export class DaemonClient { return false; } const binaryStartMs = perfNow(); - this.handleBinaryFrame(frame); + this.terminalStreams.handleFrame(frame); let frameKind: "output" | "snapshot" | "other" = "other"; if (frame.opcode === TerminalStreamOpcode.Output) { frameKind = "output"; @@ -4132,44 +4068,6 @@ export class DaemonClient { }); } - private handleBinaryFrame(frame: TerminalStreamFrame): void { - const terminalId = this.slotTerminals.get(frame.slot); - if (!terminalId) { - return; - } - - if (frame.opcode === TerminalStreamOpcode.Output) { - this.emitTerminalStreamEvent({ - terminalId, - type: "output", - data: frame.payload, - }); - return; - } - - if (frame.opcode === TerminalStreamOpcode.Snapshot) { - const state = decodeTerminalSnapshotPayload(frame.payload); - if (!state) { - return; - } - this.emitTerminalStreamEvent({ - terminalId, - type: "snapshot", - state, - }); - } - } - - private emitTerminalStreamEvent(event: TerminalStreamEvent): void { - for (const listener of this.terminalStreamListeners) { - try { - listener(event); - } catch { - // no-op - } - } - } - private updateConnectionState( next: ConnectionState, metadata?: { event: string; reason?: string; reasonCode?: string }, @@ -4227,7 +4125,7 @@ export class DaemonClient { // and responses from the previous connection will never arrive. this.clearWaiters(new Error(reason ?? "Connection lost")); this.rejectPendingSendQueue(new Error(reason ?? "Connection lost")); - this.clearTerminalSlots(); + this.terminalStreams.clearSlots(); this.lastServerInfoMessage = null; if (wasDisposed) { @@ -4293,7 +4191,7 @@ export class DaemonClient { } if (msg.type === "terminal_stream_exit") { - this.removeTerminalSlot(msg.payload.terminalId); + this.terminalStreams.removeTerminal(msg.payload.terminalId); } if (this.rawMessageListeners.size > 0) { diff --git a/packages/server/src/client/terminal-stream-router.ts b/packages/server/src/client/terminal-stream-router.ts new file mode 100644 index 000000000..0bcac1670 --- /dev/null +++ b/packages/server/src/client/terminal-stream-router.ts @@ -0,0 +1,122 @@ +import { + decodeTerminalSnapshotPayload, + encodeTerminalResizePayload, + encodeTerminalStreamFrame, + TerminalStreamOpcode, + type TerminalStreamFrame, +} from "../shared/binary-frames/index.js"; +import type { TerminalInput, TerminalState } from "../shared/messages.js"; + +export type TerminalStreamEvent = + | { terminalId: string; type: "output"; data: Uint8Array } + | { terminalId: string; type: "snapshot"; state: TerminalState }; + +export class TerminalStreamRouter { + private readonly terminalSlots = new Map(); + private readonly slotTerminals = new Map(); + private readonly listeners = new Set<(event: TerminalStreamEvent) => void>(); + + onEvent(handler: (event: TerminalStreamEvent) => void): () => void { + this.listeners.add(handler); + return () => { + this.listeners.delete(handler); + }; + } + + setSlot(terminalId: string, slot: number): void { + const existingTerminalId = this.slotTerminals.get(slot); + if (existingTerminalId && existingTerminalId !== terminalId) { + this.terminalSlots.delete(existingTerminalId); + } + + const existingSlot = this.terminalSlots.get(terminalId); + if (typeof existingSlot === "number" && existingSlot !== slot) { + this.slotTerminals.delete(existingSlot); + } + + this.terminalSlots.set(terminalId, slot); + this.slotTerminals.set(slot, terminalId); + } + + removeTerminal(terminalId: string): void { + const slot = this.terminalSlots.get(terminalId); + if (typeof slot !== "number") { + return; + } + this.terminalSlots.delete(terminalId); + if (this.slotTerminals.get(slot) === terminalId) { + this.slotTerminals.delete(slot); + } + } + + clearSlots(): void { + this.terminalSlots.clear(); + this.slotTerminals.clear(); + } + + encodeInput(terminalId: string, message: TerminalInput["message"]): Uint8Array | null { + const slot = this.terminalSlots.get(terminalId); + if (typeof slot !== "number") { + return null; + } + + if (message.type === "input") { + return encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Input, + slot, + payload: message.data, + }); + } + + if (message.type === "resize") { + return encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Resize, + slot, + payload: encodeTerminalResizePayload({ + rows: message.rows, + cols: message.cols, + }), + }); + } + + return null; + } + + handleFrame(frame: TerminalStreamFrame): void { + const terminalId = this.slotTerminals.get(frame.slot); + if (!terminalId) { + return; + } + + if (frame.opcode === TerminalStreamOpcode.Output) { + this.emit({ + terminalId, + type: "output", + data: frame.payload, + }); + return; + } + + if (frame.opcode === TerminalStreamOpcode.Snapshot) { + const state = decodeTerminalSnapshotPayload(frame.payload); + if (!state) { + return; + } + this.emit({ + terminalId, + type: "snapshot", + state, + }); + } + } + + private emit(event: TerminalStreamEvent): void { + for (const listener of this.listeners) { + try { + listener(event); + } catch { + // no-op + } + } + } +}