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

Fixes #22729

3 files changed, 127 insertions, 1 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;
diff --git a/ext/node/polyfills/events.ts b/ext/node/polyfills/events.ts
index 6e7b52791..3b7381410 100644
--- a/ext/node/polyfills/events.ts
+++ b/ext/node/polyfills/events.ts
@@ -6,6 +6,7 @@ export {
   defaultMaxListeners,
   errorMonitor,
   EventEmitter,
+  EventEmitterAsyncResource,
   getEventListeners,
   listenerCount,
   on,
diff --git a/tests/unit_node/events_test.ts b/tests/unit_node/events_test.ts
index 13abf5f79..1fc7ad1e3 100644
--- a/tests/unit_node/events_test.ts
+++ b/tests/unit_node/events_test.ts
@@ -1,6 +1,6 @@
 // Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
 
-import { EventEmitter } from "node:events";
+import events, { EventEmitter } from "node:events";
 
 EventEmitter.captureRejections = true;
 
@@ -25,3 +25,12 @@ Deno.test("regression #20441", async () => {
   ee.emit("foo");
   await promise;
 });
+
+Deno.test("eventemitter async resource", () => {
+  // @ts-ignore: @types/node is outdated
+  class Foo extends events.EventEmitterAsyncResource {}
+
+  const foo = new Foo();
+  // @ts-ignore: @types/node is outdated
+  foo.emit("bar");
+});