feat: allow HttpClient program to stop preemptively (#248)

This PR adds gracefull handling of the deletion/stopping of the
`HttpClient` program. When an unexpected packet is received, the host
responds with a RST segment. Also, when the connection is closed the
receiver queue is closed now.

---------

Co-authored-by: Pedro Gallino <[email protected]>

Author: MegaRedHand

src/programs/http_client.ts

@@ -35,6 +35,8 @@ export class HttpClient extends ProgramBase {
   private dstId: DeviceId;
   private resource: string;
 
+  private stopped = false;
+
   protected _parseInputs(inputs: string[]): void {
     if (inputs.length !== 2) {
       console.error(
@@ -50,13 +52,14 @@ export class HttpClient extends ProgramBase {
     // This starts the request from the background
     (async () => {
       await this.sendHttpRequest();
-      this.signalStop();
+      if (!this.stopped) {
+        this.signalStop();
+      }
     })();
   }
 
   protected _stop() {
-    // TODO: stop request preemptively?
-    // Nothing to do
+    this.stopped = true;
   }
 
   private async sendHttpRequest() {
@@ -86,6 +89,10 @@ export class HttpClient extends ProgramBase {
       console.warn("HttpClient failed to connect");
       return;
     }
+    if (this.stopped) {
+      socket.abort();
+      return;
+    }
     const wrote = await socket.write(httpRequest);
     if (wrote < 0) {
       console.error("HttpClient failed to write to socket");
@@ -100,6 +107,10 @@ export class HttpClient extends ProgramBase {
     const expectedLength = RESOURCE_MAP.get(this.resource)?.length || 0;
     let totalRead = 0;
     while (totalRead < expectedLength) {
+      if (this.stopped) {
+        socket.abort();
+        return;
+      }
       const readLength = await socket.read(buffer);
       if (readLength < 0) {
         console.error("HttpClient failed to read from socket");

src/types/network-modules/asyncQueue.ts

@@ -3,28 +3,58 @@ export class AsyncQueue<T> {
   private queue: T[] = [];
   // Queue to hold resolve functions for promises created by pop().
   private resolveQueue: ((value: T) => void)[] = [];
+  // Flag to indicate if the queue is closed.
+  private closed = false;
 
-  push(item: T) {
-    if (this.resolveQueue.length > 0) {
+  /**
+   * Pushes an item onto the queue.
+   * @param item The item to push.
+   * @returns true if the item was pushed, false if the queue is closed.
+   */
+  push(item: T): boolean {
+    if (this.closed) {
+      return false;
+    } else if (this.resolveQueue.length > 0) {
       const resolve = this.resolveQueue.shift();
       if (resolve) {
         resolve(item);
       }
     } else {
       this.queue.push(item);
     }
+    return true;
   }
 
-  // TODO: add timeouts
-  pop(): Promise<T> {
+  /**
+   * Pops an item from the queue.
+   * @returns A promise that resolves to the popped item, or undefined if the queue was closed.
+   */
+  pop(): Promise<T | undefined> {
     if (!this.isEmpty()) {
       return Promise.resolve(this.queue.shift());
     }
+    if (this.closed) {
+      return Promise.resolve(undefined);
+    }
     return new Promise((resolve) => {
       this.resolveQueue.push(resolve);
     });
   }
 
+  close() {
+    this.closed = true;
+    while (this.resolveQueue.length > 0) {
+      const resolve = this.resolveQueue.shift();
+      if (resolve) {
+        resolve(undefined);
+      }
+    }
+  }
+
+  isClosed(): boolean {
+    return this.closed;
+  }
+
   isEmpty(): boolean {
     return this.queue.length === 0;
   }