v0.4.0 Improve performance of listening (#5)

In testing I discovered a couple of things:

* Printers sometimes will provide much bigger transfer sizes if you
offer them.
* Being able to detect a printer is taking a while to respond is useful.

This adds features for both. 

getInput now has an optional parameter to increase the requested message
size.

There is now an event loop in the listener promise, every 500ms it can
run checks before resuming awaiting the same promise for input. This may
get additional features in the future.

Also more tests.

Author: Cellivar

package.json

@@ -1,6 +1,6 @@
 {
   "name": "web-device-mux",
-  "version": "0.3.0",
+  "version": "0.4.0",
   "description": "A small library for managing connections to WebUSB, WebHID, and Web Serial devices from multiple tabs at the same time.",
   "type": "module",
   "repository": {

src/InputListener.test.ts

@@ -1,18 +1,98 @@
-import { test, expect, describe } from 'vitest';
-import { InputMessageListener } from './InputListener.js';
+import { it, expect, describe } from 'vitest';
+import { InputMessageListener, type InputHandler } from './InputListener.js';
+import { DeviceCommunicationError } from './Error.js';
+
+function getValidListener(
+  callback: InputHandler<string> = (i) => { console.log(i); return Promise.resolve({}); }
+) {
+  return new InputMessageListener<string>(
+    async () => {
+      await new Promise(resolve => setTimeout(resolve, 50));
+      return ['hello'];
+    },
+    callback
+  );
+}
+
+function getErrorListener() {
+  return new InputMessageListener<string>(
+    async () => {
+      return new DeviceCommunicationError("This is a test error");
+    },
+    async (input) => { return {remainderData: input}}
+  );
+}
+
+function getSlowListener(
+  callback: InputHandler<string> = (i) => { console.log(i); return Promise.resolve({}); }
+) {
+  return new InputMessageListener<string>(
+    async () => {
+      await new Promise(resolve => setTimeout(resolve, 510));
+      return ['hello, but slowly.'];
+    },
+    callback
+  );
+}
 
 describe('InputListener Disposes Cleanly', () => {
-  test('Dispose does not throw', () => {
-    const listener = new InputMessageListener<string>(
-      async () => {
-        await new Promise(resolve => setTimeout(resolve, 250));
-        return ['hello '];
-      },
-      // eslint-disable-next-line @typescript-eslint/no-unused-vars
-      async (_input) => { return {}}
-    );
+  it('Dispose does not throw', () => {
+    const listener = getValidListener();
+    expect(listener["_disposed"]).toBe(false);
     listener.start();
+    expect(listener["_disposed"]).toBe(false);
     listener.dispose();
     expect(listener["_disposed"]).toEqual(true);
   });
+
+  it('Double dispose does not throw', () => {
+    const listener = getValidListener();
+    expect(listener["_disposed"]).toBe(false);
+    listener.start();
+    expect(listener["_disposed"]).toBe(false);
+    listener.dispose();
+    expect(listener["_disposed"]).toBe(true);
+    listener.dispose();
+    expect(listener["_disposed"]).toBe(true);
+  });
+
+  it('Dispose on error', async () => {
+    const listener = getErrorListener();
+    expect(listener["_disposed"]).toBe(false);
+    listener.start();
+    await new Promise(resolve => setTimeout(resolve, 50));
+    expect(listener["_disposed"]).toBe(true);
+  });
+
+  it('Slow listener returns eventually', async () => {
+    let awaitResolve: (value: boolean) => void;
+    const signal = new Promise<boolean>((resolve) => { awaitResolve = resolve});
+    const callback = (i: string[]) => {
+      awaitResolve(true);
+      expect(i).toStrictEqual(['hello, but slowly.']);
+      return Promise.resolve({});
+    }
+    const listener = getSlowListener(callback);
+    listener.start();
+    await signal;
+  });
+
+  it('Disposed slow listener is ignored', async () => {
+    let awaitResolve: (value: boolean) => void;
+    const signal = new Promise<boolean>((resolve) => { awaitResolve = resolve});
+    const callback = (i: string[]) => {
+      awaitResolve(true);
+      console.log(i);
+      return Promise.resolve({});
+    }
+    const listener = getSlowListener(callback);
+    listener.start();
+    // Disposing immediately after starting should prevent the signal from resolving.
+    listener.dispose();
+    const result = await Promise.race([
+      signal,
+      new Promise<string>(resolve => setTimeout(() => resolve('Correct!'), 510))
+    ]);
+    expect(result).toEqual('Correct!');
+  });
 });

src/InputListener.ts

@@ -8,6 +8,24 @@ export interface IHandlerResponse<TInput> {
   remainderData?: TInput[];
 }
 
+function promiseWithTimeoutDefault<T>(
+  promise: Promise<T>,
+  ms: number,
+  resolveDefault: T
+): Promise<T> {
+  // create a promise that rejects in milliseconds
+  const timeout = new Promise<T>((resolve) => {
+    setTimeout(() => {
+      resolve(resolveDefault);
+    }, ms);
+  });
+
+  // returns a race between timeout and the passed promise
+  return Promise.race<T>([promise, timeout]);
+}
+
+const listenSentinel = new DeviceCommunicationError('ListenSentinel');
+
 export class InputMessageListener<TInput> {
   public start() {
     // We intentionally want to kick off an async promise here without needing
@@ -16,8 +34,22 @@ export class InputMessageListener<TInput> {
     // eslint-disable-next-line no-async-promise-executor
     new Promise<void>(async (_, reject) => {
       let aggregate: TInput[] = [];
+      let inputPromise: undefined | Promise<TInput[] | DeviceCommunicationError> = undefined;
       do {
-        const data = await this._dataProvider();
+        // Allow resuming awaiting an already-in-progress promise.
+        inputPromise ??= this._dataProvider();
+        const data = await promiseWithTimeoutDefault(inputPromise, 500, listenSentinel);
+
+        // Printers can idle or otherwise not response for long periods of time.
+        // We have a little event loop here, where every 500 milliseconds we
+        // can do some housekeeping to see if we should continue waiting or give
+        // up entirely.
+        if (data === listenSentinel) {
+          continue;
+        } else {
+          // This indicates we've left our event loop and got a real result.
+          inputPromise = undefined;
+        }
 
         if (data instanceof DeviceCommunicationError) {
           console.error(`Error getting data from source: `, data);
@@ -39,14 +71,15 @@ export class InputMessageListener<TInput> {
         }
 
         // The handler is not required to be stateful, this is instead.
-        // The handler  may indicate more data is expected by returning incomplete
+        // The handler may indicate more data is expected by returning incomplete
         // data back. We prefix our buffer with that incomplete data to add more
         // to it and wait again for more.
         aggregate = handleResult.remainderData ?? [];
       } while (!this._disposed);
     })
     .catch((reason) => {
       // TODO: Better logging?
+      this._disposed = true;
       this.logIfDebug(`Device stream listener stopped listening unexpectedly.`, reason);
     });
   }