improve message queue logic

src/VM.ts

@@ -12,7 +12,9 @@ export default class VM {
     #socket : Socket;
     #writeLock : Mutex = new Mutex();
     #messageQueue : Queue<protocol.ProtocolMessage> = new Queue<protocol.ProtocolMessage>();
+    #messageQueueRunning : boolean = false;
     #nopTimeout : NodeJS.Timeout | null = null;
+    #reconnectTimeout : NodeJS.Timeout | null = null;
     isConnectedToVM : boolean = false;
     #noNop : boolean = false;
     connected : boolean = false;
@@ -29,17 +31,26 @@ export default class VM {
         this.#socket.connect(socketpath);
         this.#socket.on('connect', () => {
             this.connected = true;
-            log("INFO", `Connected to VM at ${socketpath}`);
+            log("INFO", `Connected to ${id} at ${socketpath}`);
             this.#enqueueMessage({
                 Operation: protocol.ProtocolOperation.NOP
             });
         });
+        this.#socket.on('close', () => {
+            this.connected = false;
+            log("INFO", `Disconnected from ${id} at ${socketpath}`);
+            log("INFO", "Reconnecting in 5 seconds...");
+            this.#reconnectTimeout = setTimeout(() => {
+                this.#reconnectTimeout = null;
+                this.#socket.connect(socketpath);
+            }, 5000);
+        });
+        this.#socket.on('error', (err) => {
+            log("ERROR", `Error on socket ${id} at ${socketpath}: ${err.message}`);
+            
+        });
         this.#socket.on('data', (data) => this.#onData(data));
         this.#events = new EventEmitter();
-        this.#events.on('ack', () => {
-            if (this.#messageQueue.size > 0) 
-                this.messageQueueLoop();
-        })
     }
 
     UploadFile(file : File) {
@@ -52,15 +63,24 @@ export default class VM {
 
     messageQueueLoop() {
         return new Promise(async (res, rej) => {
+            this.#messageQueueRunning = true;
+            while (this.#messageQueue.size > 0) {
                 var msg = this.#messageQueue.dequeue();
                 var payload = msgpack.encode(msg);
                 await this.#sendMessage(payload);
+                await new Promise<void>((res, rej) => {
+                    this.#events.once('ack', () => {
+                        res();
+                    });
+                });
+            }
+            this.#messageQueueRunning = false;
         })
     }
 
     #enqueueMessage(msg : protocol.ProtocolMessage) {
         this.#messageQueue.enqueue(msg);
-        if (this.#messageQueue.size === 1) this.messageQueueLoop();
+        if (this.#messageQueueRunning) this.messageQueueLoop();
     }
 
     #sendMessage(data : Buffer) : Promise<void> {
@@ -104,7 +124,6 @@ export default class VM {
         this.#noNop = false;
         if (!this.isConnectedToVM) {
             log("INFO", `Got connection to VM at ${this.#id}`);
-            if (this.#messageQueue.size > 0) this.messageQueueLoop();
             this.isConnectedToVM = true;
         }
         switch (msg.Operation) {
@@ -124,6 +143,7 @@ export default class VM {
         } else {
             this.isConnectedToVM = false;
             clearTimeout(this.#nopTimeout!);
+            this.#messageQueue.clear();
             log("INFO", `Connection to VM at ${this.#id} timed out`);
         }
     }