Repository: denoland/deno
Pull request: fix(ext/node): fix `EventEmitter.on`

This fixes `EventEmitter.on` behaviors and enables `parallel/test-events-on-async-iterator.js` test case.

Details:
- Removing of AbortListener in `EventEmitter.on` wasn't working. This PR fixes it.
- patches `signal[kEvents]` field if AbortSignal is passed to `EventEmitter.on` to let abort signal simulate Node.js EventTarget behavior. This field is used in the test case.
- Added validation of `options` object in `EventEmitter.on`

towards #29595 

Changed files:
filename: ext/node/polyfills/_events.mjs
patch: @@ -25,15 +25,24 @@
 
 "use strict";
 
+import { primordials } from "ext:core/mod.js";
+const {
+  ArrayPrototypeMap,
+  ObjectDefineProperty,
+  ObjectEntries,
+  SafeMap,
+  SafeSet,
+} = primordials;
+
 const kRejection = Symbol.for("nodejs.rejection");
+export const kEvents = Symbol("kEvents");
 
 import { inspect } from "ext:deno_node/internal/util/inspect.mjs";
+import { kEmptyObject } from "ext:deno_node/internal/util.mjs";
 import {
   AbortError,
-  // kEnhanceStackBeforeInspector,
   ERR_INVALID_ARG_TYPE,
   ERR_INVALID_THIS,
-  ERR_OUT_OF_RANGE,
   ERR_UNHANDLED_ERROR,
 } from "ext:deno_node/internal/errors.ts";
 
@@ -43,6 +52,7 @@ import {
   validateBoolean,
   validateFunction,
   validateNumber,
+  validateObject,
   validateString,
 } from "ext:deno_node/internal/validators.mjs";
 import { spliceOne } from "ext:deno_node/_utils.ts";
@@ -935,28 +945,49 @@ function eventTargetAgnosticAddListener(emitter, name, listener, flags) {
   } else if (typeof emitter.addEventListener === "function") {
     // EventTarget does not have `error` event semantics like Node
     // EventEmitters, we do not listen to `error` events here.
-    emitter.addEventListener(name, (arg) => {
-      listener(arg);
-    }, flags);
+    emitter.addEventListener(name, listener, flags);
   } else {
     throw new ERR_INVALID_ARG_TYPE("emitter", "EventEmitter", emitter);
   }
 }
 
+const kEventsGetter = {
+  get() {
+    const data = this[eventTargetData];
+    if (!data) {
+      return new SafeMap();
+    }
+    return new SafeMap(
+      ArrayPrototypeMap(ObjectEntries(data.listeners), (
+        [key, listeners],
+      ) => [key, new SafeSet(listeners)]),
+    );
+  },
+};
+
 /**
  * Returns an `AsyncIterator` that iterates `event` events.
  * @param {EventEmitter} emitter
  * @param {string | symbol} event
  * @param {{ signal: AbortSignal; }} [options]
  * @returns {AsyncIterator}
  */
-export function on(emitter, event, options) {
+export function on(emitter, event, options = kEmptyObject) {
+  validateObject(options, "options");
   const signal = options?.signal;
   validateAbortSignal(signal, "options.signal");
   if (signal?.aborted) {
     throw new AbortError();
   }
 
+  if (signal) {
+    // Patches [kEvents] field of AbortSignal to simulate Node.js EventTarget
+    // This is necessary to pass `paralle/test-events-on-async-iterator.js` test
+    // TODO(kt3k): This can be removed if events.getEventListeners() is used
+    // instead of `signal[kEvents]` in upstream.
+    ObjectDefineProperty(signal, kEvents, kEventsGetter);
+  }
+
   const unconsumedEvents = [];
   const unconsumedPromises = [];
   let error = null;
@@ -1000,7 +1031,6 @@ export function on(emitter, event, options) {
           signal,
           "abort",
           abortListener,
-          { once: true },
         );
       }
 
@@ -1024,6 +1054,14 @@ export function on(emitter, event, options) {
       error = err;
       eventTargetAgnosticRemoveListener(emitter, event, eventHandler);
       eventTargetAgnosticRemoveListener(emitter, "error", errorHandler);
+
+      if (signal) {
+        eventTargetAgnosticRemoveListener(
+          signal,
+          "abort",
+          abortListener,
+        );
+      }
     },
 
     [Symbol.asyncIterator]() {

filename: ext/node/polyfills/internal/event_target.mjs
patch: @@ -38,13 +38,13 @@ import { inspect } from "node:util";
 const kIsEventTarget = Symbol.for("nodejs.event_target");
 const kIsNodeEventTarget = Symbol("kIsNodeEventTarget");
 
+import { kEvents } from "ext:deno_node/_events.mjs";
 import { EventEmitter } from "node:events";
 const {
   kMaxEventTargetListeners,
   kMaxEventTargetListenersWarned,
 } = EventEmitter;
 
-const kEvents = Symbol("kEvents");
 const kIsBeingDispatched = Symbol("kIsBeingDispatched");
 const kStop = Symbol("kStop");
 const kTarget = Symbol("kTarget");

filename: tests/node_compat/config.jsonc
patch: @@ -72,7 +72,6 @@
       "test-error-aggregateTwoErrors.js",
       "test-error-prepare-stack-trace.js",
       "test-event-emitter-prepend.js",
-      "test-events-on-async-iterator.js",
       "test-events-once.js",
       "test-fs-access.js",
       "test-fs-append-file.js",

filename: tests/node_compat/test/parallel/test-events-on-async-iterator.js
patch: @@ -2,7 +2,7 @@
 // deno-lint-ignore-file
 
 // Copyright Joyent and Node contributors. All rights reserved. MIT license.
-// Taken from Node 20.11.1
+// Taken from Node 23.9.0
 // This file is automatically generated by `tests/node_compat/runner/setup.ts`. Do not modify this file manually.
 
 // Flags: --expose-internals --no-warnings
@@ -13,6 +13,7 @@ const assert = require('assert');
 const { on, EventEmitter } = require('events');
 const {
   NodeEventTarget,
+  kEvents
 } = require('internal/event_target');
 
 async function basic() {
@@ -47,6 +48,15 @@ async function invalidArgType() {
     code: 'ERR_INVALID_ARG_TYPE',
     name: 'TypeError',
   }));
+
+  const ee = new EventEmitter();
+
+  [1, 'hi', null, false, () => {}, Symbol(), 1n].map((options) => {
+    return assert.throws(() => on(ee, 'foo', options), common.expectsError({
+      code: 'ERR_INVALID_ARG_TYPE',
+      name: 'TypeError',
+    }));
+  });
 }
 
 async function error() {
@@ -370,6 +380,36 @@ async function abortableOnAfterDone() {
   });
 }
 
+async function abortListenerRemovedAfterComplete() {
+  const ee = new EventEmitter();
+  const ac = new AbortController();
+
+  const i = setInterval(() => ee.emit('foo', 'foo'), 1);
+  try {
+    // Below: either the kEvents map is empty or the 'abort' listener list is empty
+
+    // Return case
+    const endedIterator = on(ee, 'foo', { signal: ac.signal });
+    assert.ok(ac.signal[kEvents].get('abort').size > 0);
+    endedIterator.return();
+    assert.strictEqual(ac.signal[kEvents].get('abort')?.size ?? ac.signal[kEvents].size, 0);
+
+    // Throw case
+    const throwIterator = on(ee, 'foo', { signal: ac.signal });
+    assert.ok(ac.signal[kEvents].get('abort').size > 0);
+    throwIterator.throw(new Error());
+    assert.strictEqual(ac.signal[kEvents].get('abort')?.size ?? ac.signal[kEvents].size, 0);
+
+    // Abort case
+    on(ee, 'foo', { signal: ac.signal });
+    assert.ok(ac.signal[kEvents].get('abort').size > 0);
+    ac.abort(new Error());
+    assert.strictEqual(ac.signal[kEvents].get('abort')?.size ?? ac.signal[kEvents].size, 0);
+  } finally {
+    clearInterval(i);
+  }
+}
+
 async function run() {
   const funcs = [
     basic,
@@ -389,6 +429,7 @@ async function run() {
     eventTargetAbortableOnAfter,
     eventTargetAbortableOnAfter2,
     abortableOnAfterDone,
+    abortListenerRemovedAfterComplete,
   ];
 
   for (const fn of funcs) {