feat: non-blocking async in default (#25)

* WIP

* WIP

* WIP

* WIP

* WIP

* WIP

* WIP

* WIP

* WIP

* wip

Author: lisez

README.md

@@ -107,6 +107,8 @@ console.log(count); // 1
 
 ### Mixed async/sync handlers
 
+Non-blocking in default.
+
 ```typescript
 const emitter = new Xevt();
 const result: number[] = [];
@@ -128,27 +130,39 @@ for (let i = 0; i < 5; i++) {
   emitter.emit("event", i);
 }
 
-// [0, 0, 1, 1, 2, 2, 3, 3, 4, 4]
+// [0, 1, 2, 3, 4, 0, 1, 2, 3, 4]
 ```
 
+Blocking mode.
+
 ```typescript
 const emitter = new Xevt();
 const result: number[] = [];
-emitter.conjoin(["event1", "event2"], async () => {
-  result.push(1);
-});
-emitter.conjoin(["event1", "event2"], async () => {
-  result.push(2);
+emitter.on("event", (data) => {
+  result.push(data);
 });
+emitter.on(
+  "event",
+  // deno-lint-ignore require-await
+  async (data) =>
+    new Promise((res) => {
+      setTimeout(() => {
+        result.push(data);
+        res(true);
+      }, 1);
+    }),
+  { async: true },
+);
 
 for (let i = 0; i < 5; i++) {
-  emitter.emit("event1");
-  emitter.emit("event2");
+  emitter.emit("event", i);
 }
+await delay(15);
 
-// [1, 2, 1, 2, 1, 2, 1, 2, 1, 2]
+// [0, 0, 1, 1, 2, 2, 3, 3, 4, 4]
 ```
 
+
 ## Return unscriber after registered an event
 
 ```typescript

modules/conjoin_emitter.ts

@@ -16,6 +16,7 @@ import { CoreEmitter } from "modules/core_emitter.ts";
 import { Emitter } from "modules/emitter.ts";
 import { SeriesRunner } from "modules/runners/series.ts";
 import { ConjoinQueue } from "modules/conjoin_queue.ts";
+import { RelayRunner } from "modules/runners/relay.ts";
 import * as helpers from "modules/helpers.ts";
 
 export class ConjoinEmitter extends CoreEmitter<ConjoinEvents>
@@ -76,6 +77,7 @@ export class ConjoinEmitter extends CoreEmitter<ConjoinEvents>
       handler,
       options: {
         once: options?.once,
+        async: options?.async,
         dual: helpers.isDualHandler(handler),
       },
     } as DualEventHandlerSignature<any> | GeneralEventHandlerSignature<any>;
@@ -136,10 +138,7 @@ export class ConjoinEmitter extends CoreEmitter<ConjoinEvents>
       return this.prevEvents;
     };
 
-    if (this.prevEvents instanceof Promise) {
-      return Promise.resolve(this.prevEvents).then(next);
-    }
-    return next();
+    return new RelayRunner().exec(this.prevEvents, next, { async: true });
   }
 
   off(event: ConjoinEvents, handler?: EventHandler): void {

modules/core_emitter.ts

@@ -10,6 +10,7 @@ import type {
 } from "modules/types.ts";
 
 import { Logger } from "modules/logger.ts";
+import * as helpers from "modules/helpers.ts";
 
 export abstract class CoreEmitter<T> implements XCoreEmitter<T> {
   protected handlers: RegisteredHandlers;
@@ -35,15 +36,14 @@ export abstract class CoreEmitter<T> implements XCoreEmitter<T> {
     name: EventName,
     signature: EventHandlerSignature<any>,
   ): EventUnscriber {
-    // @ts-ignore TS7053
-    const async = signature.handler[Symbol.toStringTag] === "AsyncFunction" ||
-      ("then" in signature.handler);
-
-    signature.options ??= {};
-    signature.options.async = async;
-
     if (this.debug) this.logger.debug("on", name, signature);
 
+    if (
+      signature.options?.async && !helpers.isAsyncFunction(signature.handler)
+    ) {
+      delete signature.options.async;
+    }
+
     const handlers = this.handlers.get(name);
     if (handlers) {
       handlers.push(signature);

modules/emitter.ts

@@ -29,6 +29,7 @@ export class Emitter extends CoreEmitter<EventName> implements XevtEmitter {
       handler,
       options: {
         once: options?.once || event === EmitDone,
+        async: !!options?.async,
         dual: helpers.isDualHandler(handler),
       },
     } as DualEventHandlerSignature<any> | GeneralEventHandlerSignature<any>;

modules/helpers.ts

@@ -26,3 +26,14 @@ export function isDualHandler(
     "true" in handler || "false" in handler
   );
 }
+
+/**
+ * Check if a handler is an async function.
+ * @param handler The handler to check.
+ */
+export function isAsyncFunction(handler: any) {
+  // @ts-ignore TS7053
+  return (typeof handler === "function" &&
+    handler[Symbol.toStringTag] === "AsyncFunction") ||
+    ("then" in handler);
+}

modules/runners/dual.ts

@@ -4,6 +4,7 @@ import type {
 } from "modules/types.ts";
 
 import { SequenceRunner } from "modules/runners/sequence.ts";
+import { RelayRunner } from "modules/runners/relay.ts";
 
 /**
  * Run a dual event handler.
@@ -53,9 +54,6 @@ export class DualRunner<N = any> {
    * @param args The arguments to pass to the dual handler.
    */
   exec(result: any) {
-    if (result instanceof Promise) {
-      return Promise.resolve(result).then((res) => this.dualExec(res));
-    }
-    return this.dualExec(result);
+    return new RelayRunner().exec(result, (p) => this.dualExec(p));
   }
 }

modules/runners/relay.ts

@@ -0,0 +1,30 @@
+import type { EventHandler } from "modules/types.ts";
+
+/** Options to RelayRunner */
+export type RelayRunnerOptions = {
+  /**
+   * If the prev handler is a promise, it will await for it before executing the next handler.
+   * @default false
+   */
+  async: boolean;
+};
+
+/** Handle how to run before and after handler. */
+export class RelayRunner {
+  /**
+   * Execute the handler.
+   * @param prev The result of the previous handler.
+   * @param next The next handler.
+   * @param options The options to run the next handler.
+   */
+  exec<T extends EventHandler>(
+    prev: any,
+    next: T,
+    options?: Partial<RelayRunnerOptions>,
+  ): Promise<ReturnType<T>> | ReturnType<T> {
+    if (prev instanceof Promise && options?.async) {
+      return Promise.resolve(prev).then((res) => next(res));
+    }
+    return next(prev);
+  }
+}

modules/runners/sequence.ts

@@ -1,6 +1,7 @@
 import type { GeneralEventHandlerSignature } from "modules/types.ts";
 
 import { SingleRunner } from "modules/runners/single.ts";
+import { RelayRunner } from "modules/runners/relay.ts";
 
 /**
  * Run handlers in sequence.
@@ -26,18 +27,15 @@ export class SequenceRunner<
   exec(
     args: Parameters<N["handler"]>,
     index: number = 0,
-  ): void | Promise<void> {
+  ): ReturnType<N["handler"]> | Promise<ReturnType<N["handler"]>> | void {
     const profile = this.handlers[index];
     if (!profile) return;
 
     const result = new SingleRunner<N>(profile).exec(args) as any;
-
-    /**
-     * Wait for the handler to finish before moving to the next handler.
-     */
-    if (profile.options?.async || result instanceof Promise) {
-      return Promise.resolve(result).then(() => this.exec(args, index + 1));
-    }
-    return this.exec(args, index + 1);
+    return new RelayRunner().exec(
+      result,
+      () => this.exec(args, index + 1),
+      profile.options,
+    ) as any;
   }
 }

modules/runners/series.ts

@@ -1,6 +1,7 @@
 import type { EventName, RegisteredHandlers } from "modules/types.ts";
 
 import { StepRunner } from "modules/runners/step.ts";
+import { RelayRunner } from "modules/runners/relay.ts";
 
 /**
  * Run handlers each in series.
@@ -24,9 +25,9 @@ export class SeriesRunner {
     if (!key) return;
 
     const step = new StepRunner(this.handlers).exec(key);
-    if (step instanceof Promise) {
-      return Promise.resolve(step).then(() => this.exec(series, idx + 1));
-    }
-    return this.exec(series, idx + 1);
+    return new RelayRunner().exec(
+      step,
+      () => this.exec(series, idx + 1),
+    ) as any;
   }
 }

modules/runners/step.ts

@@ -9,6 +9,7 @@ import type {
 import { DualRunner } from "modules/runners/dual.ts";
 import { SingleRunner } from "modules/runners/single.ts";
 import { SequenceRunner } from "modules/runners/sequence.ts";
+import { RelayRunner } from "modules/runners/relay.ts";
 import * as helpers from "modules/helpers.ts";
 
 /**
@@ -51,17 +52,13 @@ export class StepRunner {
 
     const next = (result: any) => {
       const dualResult = new DualRunner(duals).exec(!!result);
-      if (dualResult instanceof Promise) {
-        return dualResult.then(() =>
-          this.execByIndex(handlers, duals, args, idx + 1)
-        );
-      }
+      return new RelayRunner().exec(
+        dualResult,
+        () => this.execByIndex(handlers, duals, args, idx + 1),
+      );
     };
 
-    if (handler.options?.async) {
-      return Promise.resolve(result).then(() => next(result));
-    }
-    return next(result);
+    return new RelayRunner().exec(result, next, handler.options) as any;
   }
 
   /**