QMP client now buffers lines properly

qemu/src/QmpClient.ts

@@ -1,4 +1,4 @@
-import { EventEmitter } from "node:events";
+import { EventEmitter } from 'node:events';
 
 enum QmpClientState {
 	Handshaking,
@@ -12,14 +12,14 @@ function qmpStringify(obj: any) {
 // this writer interface is used to poll back to a higher level
 // I/O layer that we want to write some data.
 export interface IQmpClientWriter {
-    writeSome(data: Buffer) : void;
+	writeSome(data: Buffer): void;
 }
 
 export type QmpClientCallback = (err: Error | null, res: any | null) => void;
 
 type QmpClientCallbackEntry = {
-    id: number,
+	id: number;
-    callback: QmpClientCallback | null
+	callback: QmpClientCallback | null;
 };
 
 export enum QmpEvent {
@@ -33,66 +33,93 @@ export enum QmpEvent {
 	VncDisconnected = 'VNC_DISCONNECTED',
 	VncInitalized = 'VNC_INITALIZED',
 	Watchdog = 'WATCHDOG'
-};
+}
 
+class LineStream extends EventEmitter {
+	// The given line seperator for the stream
+	lineSeperator = '\r\n';
+	buffer = '';
+
+	constructor() {
+		super();
+	}
+
+	push(data: Buffer) {
+		this.buffer += data.toString('utf-8');
+
+		let lines = this.buffer.split(this.lineSeperator);
+		if (lines.length > 1) {
+			this.buffer = lines.pop()!;
+			lines = lines.filter((l) => !!l);
+
+			//console.log(lines)
+			lines.forEach(l => this.emit('line', l));
+		}
+		return [];
+	}
+
+	reset() {
+		this.buffer = '';
+	}
+}
 
 // A QMP client
 export class QmpClient extends EventEmitter {
 	private state = QmpClientState.Handshaking;
-    private capabilities = "";
 	private writer: IQmpClientWriter | null = null;
 
 	private lastID = 0;
 	private callbacks = new Array<QmpClientCallbackEntry>();
 
+	private lineStream = new LineStream();
+
 	constructor() {
 		super();
+
+		let self = this;
+		this.lineStream.on('line', (line: string) => {
+			self.handleQmpLine(line);
+		});
 	}
 
 	setWriter(writer: IQmpClientWriter) {
 		this.writer = writer;
 	}
 
-    feed(data: Buffer) : void {
+	feed(data: Buffer): void {
-        let str = data.toString();
+		// Forward to the line stream. It will generate 'line' events
-
+		// as it is able to split out lines automatically.
-   /* I don't think this is needed but if it is i'm keeping this for now
+		this.lineStream.push(data);
-        if(!str.endsWith('\r\n')) {
-            console.log("incomplete message!");
-            return;
 	}
-    */
 
-        let obj = JSON.parse(str);
+	private handleQmpLine(line: string) {
+		let obj = JSON.parse(line);
 
-        switch(this.state) {
+		switch (this.state) {
 			case QmpClientState.Handshaking:
-                if(obj["return"] != undefined) {
+				if (obj['return'] != undefined) {
 					this.state = QmpClientState.Connected;
 					this.emit('connected');
 					return;
 				}
 
 				let capabilities = qmpStringify({
-                    execute: "qmp_capabilities"
+					execute: 'qmp_capabilities'
 				});
 
 				this.writer?.writeSome(Buffer.from(capabilities, 'utf8'));
 				break;
 
 			case QmpClientState.Connected:
-                if(obj["return"] != undefined || obj['error'] != undefined) {
+				if (obj['return'] != undefined || obj['error'] != undefined) {
-                    if(obj['id'] == null)
+					if (obj['id'] == null) return;
-                        return;
 
 					let cb = this.callbacks.find((v) => v.id == obj['id']);
-                    if(cb == undefined)
+					if (cb == undefined) return;
-                        return;
 
 					let error: Error | null = obj.error ? new Error(obj.error.desc) : null;
 
-                    if(cb.callback)
+					if (cb.callback) cb.callback(error, obj.return);
-                        cb.callback(error, obj.return);
 
 					this.callbacks.slice(this.callbacks.indexOf(cb));
 				} else if (obj['event']) {
@@ -116,24 +143,23 @@ export class QmpClient extends EventEmitter {
 			id: entry.id
 		};
 
-        if(args !== undefined)
+		if (args !== undefined) qmpOut['arguments'] = args;
-                qmpOut['arguments'] = args;
 
 		this.callbacks.push(entry);
 		this.writer?.writeSome(Buffer.from(qmpStringify(qmpOut), 'utf8'));
 	}
 
-    async execute(command: string, args: any | undefined = undefined) : Promise<any> {
+	async execute(command: string, args: any | undefined = undefined): Promise<any> {
 		return new Promise((res, rej) => {
 			this.executeSync(command, args, (err, result) => {
-                if(err)
+				if (err) rej(err);
-                    rej(err);
 				res(result);
 			});
 		});
 	}
 
 	reset() {
+		this.lineStream.reset();
 		this.state = QmpClientState.Handshaking;
 	}
 }