feat: allow ensureDelivery to be able to ensure delivery on emit

Signed-off-by: Remi Cattiau <[email protected]>

Author: loopingz

README.md

@@ -107,7 +107,7 @@ function sendWithAxios(message) {
 
 const emit = emitterFor(sendWithAxios, { mode: Mode.BINARY });
 // Set the emit
-Emitter.getSingleton().on("event", emit);
+Emitter.on("cloudevent", emit);
 
 ...
 // In any part of the code will send the event

src/event/cloudevent.ts

@@ -171,10 +171,11 @@ export class CloudEvent implements CloudEventV1, CloudEventV03 {
   /**
    * Emit this CloudEvent through the application
    *
-   * @return {CloudEvent} current CloudEvent object
+   * @param {boolean} ensureDelivery fail the promise if one listener fail
+   * @return {Promise<CloudEvent>} this
    */
-  public emit(): this {
-    Emitter.emitEvent(this);
+  public async emit(ensureDelivery = false): Promise<this> {
+    await Emitter.emitEvent(this, ensureDelivery);
     return this;
   }
 

src/transport/emitter.ts

@@ -98,7 +98,7 @@ export class Emitter extends EventEmitter {
   /**
    * Singleton store
    */
-  static singleton: Emitter | undefined = undefined;
+  static instance: Emitter | undefined = undefined;
   url?: string;
   protocol: Protocol;
   binaryEmitter: EmitterFunction;
@@ -113,7 +113,9 @@ export class Emitter extends EventEmitter {
    * @param {Object} [options] The configuration options for this event. Options
    * provided will be passed along to Node.js `http.request()`.
    * https://nodejs.org/api/http.html#http_http_request_options_callback
-   * @deprecated Will be removed in 4.0.0. Consider using the emitterFactory
+   * @deprecated in 4.0.0 this class functionality will be limited to events propagated
+   * throughout the Node.js process. To emit events over a transport protocol, consider using the
+   * emitterFactory
    */
   constructor(options: TransportOptions = { protocol: Protocol.HTTPBinary }) {
     super();
@@ -153,20 +155,42 @@ export class Emitter extends EventEmitter {
    *
    * @return {Emitter} return Emitter singleton
    */
-  static getSingleton(): Emitter {
-    if (!Emitter.singleton) {
-      Emitter.singleton = new Emitter();
+  static getInstance(): Emitter {
+    if (!Emitter.instance) {
+      Emitter.instance = new Emitter();
     }
-    return Emitter.singleton;
+    return Emitter.instance;
+  }
+
+  /**
+   * Add a listener for eventing
+   *
+   * @param {string} event type to listen to
+   * @param {Function} listener to call on event
+   * @return {void}
+   */
+  static on(event: "cloudevent" | "newListener" | "removeListener", listener: (...args: any[]) => void): void {
+    this.getInstance().on(event, listener);
   }
 
   /**
    * Emit an event inside this application
    *
    * @param {CloudEvent} event to emit
+   * @param {boolean} ensureDelivery fail the promise if one listener fail
    * @return {void}
    */
-  static emitEvent(event: CloudEvent): void {
-    this.getSingleton().emit("event", event);
+  static async emitEvent(event: CloudEvent, ensureDelivery = false): Promise<void> {
+    if (!ensureDelivery) {
+      // Ensure delivery is disabled so we don't wait for Promise
+      Emitter.getInstance().emit("cloudevent", event);
+    } else {
+      // Execute all listeners and wrap them in a Promise
+      await Promise.all(
+        Emitter.getInstance()
+          .listeners("cloudevent")
+          .map(async (l) => l(event)),
+      );
+    }
   }
 }

test/integration/emitter_singleton_test.ts

@@ -0,0 +1,53 @@
+import "mocha";
+
+import { emitterFor, HTTP, Mode, Message, Emitter } from "../../src";
+
+import { fixture, assertStructured } from "./emitter_factory_test";
+
+import { rejects, doesNotReject } from "assert";
+
+describe("Emitter Singleton", () => {
+  it("emit", async () => {
+    let msg: Message | unknown = undefined;
+    const fn = async (message: Message) => {
+      msg = message;
+    };
+    const emitter = emitterFor(fn, { binding: HTTP, mode: Mode.STRUCTURED });
+    Emitter.on("cloudevent", emitter);
+    fixture.emit();
+    while (msg === undefined) {
+      await new Promise((resolve) => setTimeout(resolve, 10));
+    }
+    let body: unknown = (<Message>(<unknown>msg)).body;
+    if (typeof body === "string") {
+      body = JSON.parse(body);
+    }
+    assertStructured({ ...(<any>body), ...(<Message>(<unknown>msg)).headers });
+  });
+
+  it("emitWithEnsureDelivery", async () => {
+    let msg: Message | unknown = undefined;
+    const fn = async (message: Message) => {
+      msg = message;
+    };
+    const emitter = emitterFor(fn, { binding: HTTP, mode: Mode.STRUCTURED });
+    Emitter.on("cloudevent", emitter);
+    await fixture.emit(true);
+    let body: any = (<Message>msg).body;
+    if (typeof body === "string") {
+      body = JSON.parse(body);
+    }
+    assertStructured({ ...(<any>body), ...(<Message>(<unknown>msg)).headers });
+  });
+
+  it("emitWithEnsureDeliveryError", async () => {
+    const emitter = async () => {
+      throw new Error("Not sent");
+    };
+    Emitter.on("cloudevent", emitter);
+    // Should fail with emitWithEnsureDelivery
+    await rejects(() => fixture.emit(true));
+    // Should not fail with emitWithEnsureDelivery
+    await doesNotReject(() => fixture.emit(false));
+  });
+});