fix(ext/node): implement EventEmitterAsyncResource (#22994)

Fixes #22729

1 files changed, 116 insertions, 0 deletions

diff --git a/ext/node/polyfills/_events.mjs b/ext/node/polyfills/_events.mjs
index bd9e92d06..46afb37b1 100644
--- a/ext/node/polyfills/_events.mjs
+++ b/ext/node/polyfills/_events.mjs
@@ -32,14 +32,17 @@ import {
   AbortError,
   // kEnhanceStackBeforeInspector,
   ERR_INVALID_ARG_TYPE,
+  ERR_INVALID_THIS,
   ERR_OUT_OF_RANGE,
   ERR_UNHANDLED_ERROR,
 } from "ext:deno_node/internal/errors.ts";
 
+import { AsyncResource } from "node:async_hooks";
 import {
   validateAbortSignal,
   validateBoolean,
   validateFunction,
+  validateString,
 } from "ext:deno_node/internal/validators.mjs";
 import { spliceOne } from "ext:deno_node/_utils.ts";
 import { nextTick } from "ext:deno_node/_process/process.ts";
@@ -1035,3 +1038,116 @@ export function on(emitter, event, options) {
     iterator.return();
   }
 }
+
+const kAsyncResource = Symbol("kAsyncResource");
+const kEventEmitter = Symbol("kEventEmitter");
+
+class EventEmitterReferencingAsyncResource extends AsyncResource {
+  /**
+   * @param {EventEmitter} ee
+   * @param {string} [type]
+   * @param {{
+   *   triggerAsyncId?: number,
+   *   requireManualDestroy?: boolean,
+   * }} [options]
+   */
+  constructor(ee, type, options) {
+    super(type, options);
+    this[kEventEmitter] = ee;
+  }
+
+  /**
+   * @type {EventEmitter}
+   */
+  get eventEmitter() {
+    if (this[kEventEmitter] === undefined) {
+      throw new ERR_INVALID_THIS("EventEmitterReferencingAsyncResource");
+    }
+    return this[kEventEmitter];
+  }
+}
+
+export class EventEmitterAsyncResource extends EventEmitter {
+  /**
+   * @param {{
+   *   name?: string,
+   *   triggerAsyncId?: number,
+   *   requireManualDestroy?: boolean,
+   * }} [options]
+   */
+  constructor(options = undefined) {
+    let name;
+    if (typeof options === "string") {
+      name = options;
+      options = undefined;
+    } else {
+      if (new.target === EventEmitterAsyncResource) {
+        validateString(options?.name, "options.name");
+      }
+      name = options?.name || new.target.name;
+    }
+    super(options);
+
+    this[kAsyncResource] = new EventEmitterReferencingAsyncResource(
+      this,
+      name,
+      options,
+    );
+  }
+
+  /**
+   * @param {symbol,string} event
+   * @param  {...any} args
+   * @returns {boolean}
+   */
+  emit(event, ...args) {
+    if (this[kAsyncResource] === undefined) {
+      throw new ERR_INVALID_THIS("EventEmitterAsyncResource");
+    }
+    const { asyncResource } = this;
+    args.unshift(super.emit, this, event);
+    return asyncResource.runInAsyncScope.apply(asyncResource, args);
+  }
+
+  /**
+   * @returns {void}
+   */
+  emitDestroy() {
+    if (this[kAsyncResource] === undefined) {
+      throw new ERR_INVALID_THIS("EventEmitterAsyncResource");
+    }
+    this.asyncResource.emitDestroy();
+  }
+
+  /**
+   * @type {number}
+   */
+  get asyncId() {
+    if (this[kAsyncResource] === undefined) {
+      throw new ERR_INVALID_THIS("EventEmitterAsyncResource");
+    }
+    return this.asyncResource.asyncId();
+  }
+
+  /**
+   * @type {number}
+   */
+  get triggerAsyncId() {
+    if (this[kAsyncResource] === undefined) {
+      throw new ERR_INVALID_THIS("EventEmitterAsyncResource");
+    }
+    return this.asyncResource.triggerAsyncId();
+  }
+
+  /**
+   * @type {EventEmitterReferencingAsyncResource}
+   */
+  get asyncResource() {
+    if (this[kAsyncResource] === undefined) {
+      throw new ERR_INVALID_THIS("EventEmitterAsyncResource");
+    }
+    return this[kAsyncResource];
+  }
+}
+
+EventEmitter.EventEmitterAsyncResource = EventEmitterAsyncResource;