52 lines
1.6 KiB
TypeScript
52 lines
1.6 KiB
TypeScript
|
import type { IRustBridgeLogger } from './interfaces/index.js';
|
||
|
|
|
||
|
|
/**
|
||
|
|
* Buffer-based newline scanner for streaming binary data.
|
||
|
|
* Accumulates chunks and emits complete lines via callback.
|
||
|
|
* Used by both StdioTransport and SocketTransport.
|
||
|
|
*/
|
||
|
|
export class LineScanner {
|
||
|
|
private buffer: Buffer = Buffer.alloc(0);
|
||
|
|
private maxPayloadSize: number;
|
||
|
|
private logger: IRustBridgeLogger;
|
||
|
|
|
||
|
|
constructor(maxPayloadSize: number, logger: IRustBridgeLogger) {
|
||
|
|
this.maxPayloadSize = maxPayloadSize;
|
||
|
|
this.logger = logger;
|
||
|
|
}
|
||
|
|
|
||
|
|
/**
|
||
|
|
* Feed a chunk of data. Calls onLine for each complete newline-terminated line found.
|
||
|
|
*/
|
||
|
|
public push(chunk: Buffer, onLine: (line: string) => void): void {
|
||
|
|
this.buffer = Buffer.concat([this.buffer, chunk]);
|
||
|
|
|
||
|
|
let newlineIndex: number;
|
||
|
|
while ((newlineIndex = this.buffer.indexOf(0x0A)) !== -1) {
|
||
|
|
const lineBuffer = this.buffer.subarray(0, newlineIndex);
|
||
|
|
this.buffer = this.buffer.subarray(newlineIndex + 1);
|
||
|
|
|
||
|
|
if (lineBuffer.length > this.maxPayloadSize) {
|
||
|
|
this.logger.log('error', `Inbound message exceeds maxPayloadSize (${lineBuffer.length} bytes), dropping`);
|
||
|
|
continue;
|
||
|
|
}
|
||
|
|
|
||
|
|
const line = lineBuffer.toString('utf8').trim();
|
||
|
|
if (line) {
|
||
|
|
onLine(line);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// Prevent OOM if sender never sends newline
|
||
|
|
if (this.buffer.length > this.maxPayloadSize) {
|
||
|
|
this.logger.log('error', `Buffer exceeded maxPayloadSize (${this.buffer.length} bytes) without newline, clearing`);
|
||
|
|
this.buffer = Buffer.alloc(0);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
/** Reset the internal buffer. */
|
||
|
|
public clear(): void {
|
||
|
|
this.buffer = Buffer.alloc(0);
|
||
|
|
}
|
||
|
|
}
|