Imrove error handling test coverage

src/dispatch-queue.ts

@@ -1,13 +1,14 @@
 import { DispatchQueueEventMap } from "./events/events-map.ts";
 import { DispatchQueueEvents } from "./events/events.ts";
-import { DispatchQueueStartupErrorEvent } from "./events/startup-error-event.ts";
+import { DispatchQueueRuntimeErrorEvent } from "./events/runtime-error-event.ts";
 import { DispatchQueueWorkerErrorEvent } from "./events/worker-error-event.ts";
 import { Queue } from "./queue.ts";
 import { Worker } from "./worker.ts";
 
 interface DispatchQueueOptions<T> {
-  processor(value: T, workerId: string): Promise<void>;
+  autoStart?: boolean;
   concurrentWorkers?: number;
+  processor(value: T, workerId: string): Promise<void>;
 }
 
 /**
@@ -20,7 +21,11 @@ export class DispatchQueue<T> {
   private readonly _queue: Queue<T> = new Queue<T>();
   private readonly _readyWorkerQueue: Queue<Worker<T>> = new Queue<Worker<T>>();
 
-  constructor({ processor, concurrentWorkers = 2 }: DispatchQueueOptions<T>) {
+  constructor({
+    autoStart = true,
+    concurrentWorkers = 2,
+    processor,
+  }: DispatchQueueOptions<T>) {
     for (let i = 0; i < concurrentWorkers; i++) {
       this._readyWorkerQueue.enque(
         new Worker(`worker-${i}`, {
@@ -35,6 +40,10 @@ export class DispatchQueue<T> {
       );
     }
 
+    if (!autoStart) {
+      return;
+    }
+
     this.startProcessing();
   }
 
@@ -64,8 +73,8 @@ export class DispatchQueue<T> {
    * the DispatchQueue
    */
   startProcessing() {
-    this.start().catch((error) => {
-      this._events.dispatchEvent(new DispatchQueueStartupErrorEvent(error));
+    this.run().catch((error) => {
+      this._events.dispatchEvent(new DispatchQueueRuntimeErrorEvent(error));
     });
   }
 
@@ -76,7 +85,7 @@ export class DispatchQueue<T> {
     this._abortController?.abort();
   }
 
-  private async start(): Promise<void> {
+  private async run(): Promise<void> {
     this._abortController = new AbortController();
 
     while (!this._abortController.signal.aborted) {

src/events/events-map.ts

@@ -1,8 +1,8 @@
-import { DispatchQueueStartupErrorEvent } from "./startup-error-event.ts";
+import { DispatchQueueRuntimeErrorEvent } from "./runtime-error-event.ts";
 import { DispatchQueueWorkerErrorEvent } from "./worker-error-event.ts";
 import { DispatchQueueEvents } from "./events.ts";
 
 export interface DispatchQueueEventMap {
   [DispatchQueueEvents.WorkerError]: DispatchQueueWorkerErrorEvent;
-  [DispatchQueueEvents.StartupError]: DispatchQueueStartupErrorEvent;
+  [DispatchQueueEvents.RuntimeError]: DispatchQueueRuntimeErrorEvent;
 }

src/events/events.ts

@@ -1,4 +1,4 @@
 export enum DispatchQueueEvents {
   WorkerError = "worker-error",
-  StartupError = "startup-error",
+  RuntimeError = "runtime-error",
 }

src/events/startup-error-event.ts → src/events/runtime-error-event.ts

@@ -1,10 +1,10 @@
 import { DispatchQueueEvents } from "./events.ts";
 
-export class DispatchQueueStartupErrorEvent extends Event {
+export class DispatchQueueRuntimeErrorEvent extends Event {
   constructor(
     public readonly error: unknown,
     public readonly message: string = "Startup exception",
   ) {
-    super(DispatchQueueEvents.StartupError);
+    super(DispatchQueueEvents.RuntimeError);
   }
 }

tests/dispatch-queue.test.ts

@@ -15,6 +15,14 @@ import {
 import { DispatchQueue } from "../src/dispatch-queue.ts";
 import { DispatchQueueEvents } from "../src/events/events.ts";
 import { DispatchQueueWorkerErrorEvent } from "../src/events/worker-error-event.ts";
+import {
+  assertSpyCalls,
+  returnsNext,
+  Stub,
+  stub,
+} from "https://deno.land/std@0.158.0/testing/mock.ts";
+import { Queue } from "../src/queue.ts";
+import { DispatchQueueRuntimeErrorEvent } from "../src/events/runtime-error-event.ts";
 
 describe("Dispatch", () => {
   let dispatcher: DispatchQueue<string>;
@@ -156,4 +164,85 @@ describe("Dispatch", () => {
       assertSpyCall(mockedProcessor, 0);
     });
   });
+
+  describe("when removeEventListener()", () => {
+    let eventListener: Spy<
+      unknown,
+      [_evt: DispatchQueueWorkerErrorEvent],
+      void
+    >;
+
+    beforeEach(() => {
+      mockedProcessor = spy((_value, _workerId) =>
+        Promise.reject("this is an error")
+      );
+      dispatcher = new DispatchQueue<string>({
+        processor: mockedProcessor,
+      });
+
+      eventListener = spy((_evt: DispatchQueueWorkerErrorEvent) => {});
+
+      dispatcher.addEventListener(
+        DispatchQueueEvents.WorkerError,
+        eventListener,
+      );
+    });
+
+    it("does not handle event", async () => {
+      // Arrange
+      dispatcher.removeEventListener(
+        DispatchQueueEvents.WorkerError,
+        eventListener,
+      );
+
+      // Act
+      dispatcher.process("test1");
+      await delay(10);
+
+      // Assert
+      assertSpyCalls(eventListener, 0);
+    });
+  });
+
+  describe("when unhandle runtime error occurs", () => {
+    let stubbedDequeue: Stub<Queue<unknown>, [], Promise<unknown>>;
+
+    beforeEach(() => {
+      stubbedDequeue = stub(
+        Queue.prototype,
+        "deque",
+        returnsNext([Promise.reject()]),
+      );
+
+      mockedProcessor = spy((_value, _workerId) => Promise.resolve());
+
+      dispatcher = new DispatchQueue<string>({
+        processor: mockedProcessor,
+      });
+    });
+
+    afterEach(() => {
+      stubbedDequeue.restore();
+    });
+
+    it("it handles exception", async () => {
+      const deferredPromise = new Deferred<void>();
+
+      const eventListener = spy((_evt: DispatchQueueRuntimeErrorEvent) => {
+        deferredPromise.resolve();
+      });
+
+      dispatcher.addEventListener(
+        DispatchQueueEvents.RuntimeError,
+        eventListener,
+      );
+
+      dispatcher.process("test");
+      dispatcher.startProcessing();
+
+      await deferredPromise;
+
+      assertSpyCalls(eventListener, 2);
+    });
+  });
 });