lib: unifying on and eventsToAsyncIteratorFactory functions

lib/events.js

@@ -23,7 +23,6 @@
 
 const {
   ArrayPrototypeJoin,
-  ArrayPrototypeShift,
   ArrayPrototypeSlice,
   ArrayPrototypeSplice,
   ArrayPrototypeUnshift,
@@ -33,6 +32,7 @@ const {
   FunctionPrototypeBind,
   FunctionPrototypeCall,
   NumberIsNaN,
+  NumberPOSITIVE_INFINITY,
   ObjectCreate,
   ObjectDefineProperty,
   ObjectDefineProperties,
@@ -84,7 +84,7 @@ const kErrorMonitor = Symbol('events.errorMonitor');
 const kMaxEventTargetListeners = Symbol('events.maxEventTargetListeners');
 const kMaxEventTargetListenersWarned =
   Symbol('events.maxEventTargetListenersWarned');
-
+let FixedQueue;
 let EventEmitterAsyncResource;
 // The EventEmitterAsyncResource has to be initialized lazily because event.js
 // is loaded so early in the bootstrap process, before async_hooks is available.
@@ -999,7 +999,13 @@ function eventTargetAgnosticAddListener(emitter, name, listener, flags) {
  * Returns an `AsyncIterator` that iterates `event` events.
  * @param {EventEmitter} emitter
  * @param {string | symbol} event
- * @param {{ signal: AbortSignal; }} [options]
+ * @param {{
+ *    signal: AbortSignal;
+ *    close?: string[];
+ *    pauseThreshold?: number,
+ *    resumeThreshold?: number,
+ *    firstEventParam: boolean
+ *   }} [options]
  * @returns {AsyncIterator}
  */
 function on(emitter, event, options) {
@@ -1008,16 +1014,44 @@ function on(emitter, event, options) {
   if (signal?.aborted)
     throw new AbortError(undefined, { cause: signal?.reason });
 
-  const unconsumedEvents = [];
-  const unconsumedPromises = [];
+  if (!FixedQueue) FixedQueue = require('internal/fixed_queue');
+
+  const unconsumedEvents = new FixedQueue();
+  const unconsumedPromises = new FixedQueue();
   let error = null;
   let finished = false;
+  const close = options?.close;
+  const firstEventParam = options?.firstEventParam;
+  let pauseThreshold = options?.pauseThreshold;
+  let resumeThreshold = options?.resumeThreshold;
+  let paused = false;
+  let size = 0;
+  if (pauseThreshold === undefined) {
+    pauseThreshold = NumberPOSITIVE_INFINITY;
+  }
+  if (resumeThreshold === undefined) {
+    resumeThreshold = 1;
+  }
 
   const iterator = ObjectSetPrototypeOf({
+    get highWaterMark() {
+      return pauseThreshold;
+    },
+    get isPaused() {
+      return paused;
+    },
+    get queueSize() {
+      return size;
+    },
     next() {
       // First, we consume all unread events
-      const value = unconsumedEvents.shift();
-      if (value) {
+      if (size) {
+        const value = unconsumedEvents.shift();
+        size--;
+        if (paused && size < resumeThreshold) {
+          emitter.resume();
+          paused = false;
+        }
         return PromiseResolve(createIterResult(value, false));
       }
 
@@ -1033,6 +1067,7 @@ function on(emitter, event, options) {
 
       // If the iterator is finished, resolve to done
       if (finished) {
+        removeListeners();
         return PromiseResolve(createIterResult(undefined, true));
       }
 
@@ -1043,24 +1078,7 @@ function on(emitter, event, options) {
     },
 
     return() {
-      eventTargetAgnosticRemoveListener(emitter, event, eventHandler);
-      eventTargetAgnosticRemoveListener(emitter, 'error', errorHandler);
-
-      if (signal) {
-        eventTargetAgnosticRemoveListener(
-          signal,
-          'abort',
-          abortListener,
-          { once: true });
-      }
-
-      finished = true;
-
-      for (const promise of unconsumedPromises) {
-        promise.resolve(createIterResult(undefined, true));
-      }
-
-      return PromiseResolve(createIterResult(undefined, true));
+      return closeHandler();
     },
 
     throw(err) {
@@ -1069,19 +1087,23 @@ function on(emitter, event, options) {
                                        'Error', err);
       }
       error = err;
-      eventTargetAgnosticRemoveListener(emitter, event, eventHandler);
-      eventTargetAgnosticRemoveListener(emitter, 'error', errorHandler);
+      removeListeners();
     },
 
     [SymbolAsyncIterator]() {
       return this;
     }
   }, AsyncIteratorPrototype);
 
-  eventTargetAgnosticAddListener(emitter, event, eventHandler);
+  eventTargetAgnosticAddListener(emitter, event, firstEventParam ? eventHandlerFirstParam : eventHandler);
   if (event !== 'error' && typeof emitter.on === 'function') {
     emitter.on('error', errorHandler);
   }
+  if (close && close.length) {
+    for (let i = 0; i <= close.length; i++) {
+      eventTargetAgnosticAddListener(emitter, close[i], closeHandler);
+    }
+  }
 
   if (signal) {
     eventTargetAgnosticAddListener(
@@ -1093,23 +1115,59 @@ function on(emitter, event, options) {
 
   return iterator;
 
+  function closeHandler() {
+    removeListeners();
+    finished = true;
+
+    while (!unconsumedPromises.isEmpty()) {
+      unconsumedPromises.shift().resolve(createIterResult(undefined, true));
+    }
+
+    return PromiseResolve(createIterResult(undefined, true));
+  }
+
+  function removeListeners() {
+    eventTargetAgnosticRemoveListener(emitter, event, firstEventParam ? eventHandlerFirstParam : eventHandler);
+    eventTargetAgnosticRemoveListener(emitter, 'error', errorHandler);
+    if (close && close.length) {
+      for (let i = 0; i <= close.length; i++) {
+        eventTargetAgnosticRemoveListener(emitter, close[i], closeHandler);
+      }
+    }
+    if (signal) {
+      eventTargetAgnosticRemoveListener(
+        signal,
+        'abort',
+        abortListener,
+        { once: true });
+    }
+  }
+
   function abortListener() {
     errorHandler(new AbortError(undefined, { cause: signal?.reason }));
   }
 
   function eventHandler(...args) {
-    const promise = ArrayPrototypeShift(unconsumedPromises);
-    if (promise) {
-      promise.resolve(createIterResult(args, false));
+    return eventHandlerFirstParam(args);
+  }
+
+  function eventHandlerFirstParam(arg) {
+    if (unconsumedPromises.isEmpty()) {
+      size++;
+      if (size > pauseThreshold) {
+        paused = true;
+        emitter.pause();
+      }
+      unconsumedEvents.push(arg);
     } else {
-      unconsumedEvents.push(args);
+      unconsumedPromises.shift().resolve(createIterResult(arg, false));
     }
   }
 
   function errorHandler(err) {
     finished = true;
 
-    const toError = ArrayPrototypeShift(unconsumedPromises);
+    const toError = unconsumedPromises.shift();
 
     if (toError) {
       toError.reject(err);

lib/internal/readline/interface.js

@@ -1343,12 +1343,11 @@ class Interface extends InterfaceConstructor {
    */
   [SymbolAsyncIterator]() {
     if (!this[kLineObjectStream]) {
-      this[kLineObjectStream] = require(
-        'internal/readline/eventsToAsyncIteratorFactory'
-      )(
-        this, {
-          itemEvents: ['line'],
-          closeEvents: ['close']
+      this[kLineObjectStream] = EventEmitter.on(
+        this, 'line', {
+          close: ['close'],
+          pauseThreshold: 1024,
+          firstEventParam: true,
         });
     }
     return this[kLineObjectStream];