stream: let Readable.toWeb reuse eos immediate path

Author: ikeyan

lib/internal/streams/end-of-stream.js

@@ -67,8 +67,6 @@ function bindAsyncResource(fn, type) {
   };
 }
 
-const kEosImmediateClose = Symbol('kEosImmediateClose');
-
 /**
  * Returns the current stream error tracked by eos(), if any.
  * @param {import('stream').Stream} stream
@@ -87,7 +85,7 @@ function getEosErrored(stream) {
  * @param {boolean | null} readableFinished
  * @param {boolean} writable
  * @param {boolean | null} writableFinished
- * @returns {Error | typeof kEosImmediateClose}
+ * @returns {Error | null}
  */
 function getEosOnCloseError(stream, readable, readableFinished, writable, writableFinished) {
   const errored = getEosErrored(stream);
@@ -106,56 +104,57 @@ function getEosOnCloseError(stream, readable, readableFinished, writable, writab
     }
   }
 
-  return kEosImmediateClose;
+  return null;
 }
 
-function getEosInitialState(stream, options = kEmptyObject) {
+const kEosNodeSyncronousCallback = Symbol('kEosNodeSynchronousCallback');
+
+function eos(stream, options, callback) {
+  if (arguments.length === 2) {
+    callback = options;
+    options = kEmptyObject;
+  } else if (options == null) {
+    options = kEmptyObject;
+  } else {
+    validateObject(options, 'options');
+  }
+  validateFunction(callback, 'callback');
+  validateAbortSignal(options.signal, 'options.signal');
+
+  if (isReadableStream(stream) || isWritableStream(stream)) {
+    return eosWeb(stream, options, callback);
+  }
+
+  if (!isNodeStream(stream)) {
+    throw new ERR_INVALID_ARG_TYPE('stream', ['ReadableStream', 'WritableStream', 'Stream'], stream);
+  }
+
   const readable = options.readable ?? isReadableNodeStream(stream);
   const writable = options.writable ?? isWritableNodeStream(stream);
 
   // TODO (ronag): Improve soft detection to include core modules and
   // common ecosystem modules that do properly emit 'close' but fail
   // this generic check.
-  const willEmitClose = (
+  let willEmitClose = (
     _willEmitClose(stream) &&
     isReadableNodeStream(stream) === readable &&
     isWritableNodeStream(stream) === writable
   );
+  let writableFinished = isWritableFinished(stream, false);
+  let readableFinished = isReadableFinished(stream, false);
 
-  return {
-    readable,
-    writable,
-    willEmitClose,
-    writableFinished: isWritableFinished(stream, false),
-    readableFinished: isReadableFinished(stream, false),
-    closed: isClosed(stream),
-  };
-}
-
-/**
- * Classifies whether eos() can synchronously determine the result for the
- * current stream snapshot, or if it must defer to future events.
- * @param {import('stream').Stream} stream
- * @param {object} [options]
- * @param {boolean} [options.readable]
- * @param {boolean} [options.writable]
- * @param {ReturnType<typeof getEosInitialState>} [state]
- * @returns {Error | typeof kEosImmediateClose | null}
- */
-function getEosImmediateResult(stream, options = kEmptyObject, state = getEosInitialState(stream, options)) {
-  const {
-    readable,
-    writable,
-    willEmitClose,
-    writableFinished,
-    readableFinished,
-    closed,
-  } = state;
   const wState = stream._writableState;
   const rState = stream._readableState;
 
-  if (closed) {
-    return getEosOnCloseError(
+  /**
+   * @type {Error | null | undefined}
+   * undefined: to be determined
+   * null: no error
+   * Error: an error occurred
+   */
+  let immediateResult;
+  if (isClosed(stream)) {
+    immediateResult = getEosOnCloseError(
       stream,
       readable,
       readableFinished,
@@ -164,39 +163,57 @@ function getEosImmediateResult(stream, options = kEmptyObject, state = getEosIni
     );
   } else if (wState?.errorEmitted || rState?.errorEmitted) {
     if (!willEmitClose) {
-      return getEosErrored(stream) ?? kEosImmediateClose;
+      immediateResult = getEosErrored(stream);
     }
   } else if (
     !readable &&
     (!willEmitClose || isReadable(stream)) &&
     (writableFinished || isWritable(stream) === false) &&
     (wState == null || wState.pendingcb === undefined || wState.pendingcb === 0)
   ) {
-    return getEosErrored(stream) ?? kEosImmediateClose;
+    immediateResult = getEosErrored(stream);
   } else if (
     !writable &&
     (!willEmitClose || isWritable(stream)) &&
     (readableFinished || isReadable(stream) === false)
   ) {
-    return getEosErrored(stream) ?? kEosImmediateClose;
+    immediateResult = getEosErrored(stream);
   } else if ((rState && stream.req && stream.aborted)) {
-    return getEosErrored(stream) ?? kEosImmediateClose;
+    immediateResult = getEosErrored(stream);
+  }
+  const returnImmediately = (result) => {
+    const args = result === null ? [] : [result];
+    if (options[kEosNodeSyncronousCallback]) {
+      callback.call(stream, ...args);
+    } else {
+      if (AsyncContextFrame.current() || enabledHooksExist()) {
+        // Avoid AsyncResource.bind() because it calls ObjectDefineProperties which
+        // is a bottleneck here.
+        callback = bindAsyncResource(callback, 'STREAM_END_OF_STREAM');
+      }
+      process.nextTick(() => callback.call(stream, ...args));
+    }
+  };
+  let cleanup = () => {
+    callback = nop;
+  };
+  if (immediateResult !== undefined) {
+    if (options.error !== false) {
+      const onerror = () => {};
+      stream.on('error', onerror);
+      cleanup = () => {
+        callback = nop;
+        stream.removeListener('error', onerror);
+      };
+    }
+    returnImmediately(immediateResult);
+    return cleanup;
   }
 
-  return null;
-}
-
-function eos(stream, options, callback) {
-  if (arguments.length === 2) {
-    callback = options;
-    options = kEmptyObject;
-  } else if (options == null) {
-    options = kEmptyObject;
-  } else {
-    validateObject(options, 'options');
+  if (options.signal?.aborted) {
+    returnImmediately(new AbortError(undefined, { cause: options.signal.reason }));
+    return cleanup;
   }
-  validateFunction(callback, 'callback');
-  validateAbortSignal(options.signal, 'options.signal');
 
   if (AsyncContextFrame.current() || enabledHooksExist()) {
     // Avoid AsyncResource.bind() because it calls ObjectDefineProperties which
@@ -206,30 +223,12 @@ function eos(stream, options, callback) {
     callback = once(callback);
   }
 
-  if (isReadableStream(stream) || isWritableStream(stream)) {
-    return eosWeb(stream, options, callback);
-  }
-
-  if (!isNodeStream(stream)) {
-    throw new ERR_INVALID_ARG_TYPE('stream', ['ReadableStream', 'WritableStream', 'Stream'], stream);
-  }
-
-  const eosState = getEosInitialState(stream, options);
-  const wState = stream._writableState;
-
   const onlegacyfinish = () => {
     if (!stream.writable) {
       onfinish();
     }
   };
 
-  const { readable, writable } = eosState;
-  let {
-    willEmitClose,
-    writableFinished,
-    readableFinished,
-    closed,
-  } = eosState;
   const onfinish = () => {
     writableFinished = true;
     // Stream should not be destroyed here. If it is that
@@ -271,10 +270,8 @@ function eos(stream, options, callback) {
   };
 
   const onclose = () => {
-    closed = true;
-
     const error = getEosOnCloseError(stream, readable, readableFinished, writable, writableFinished);
-    if (error === kEosImmediateClose) {
+    if (error === null) {
       callback.call(stream);
     } else {
       callback.call(stream, error);
@@ -312,16 +309,7 @@ function eos(stream, options, callback) {
   }
   stream.on('close', onclose);
 
-  const immediateResult = getEosImmediateResult(stream, options, eosState);
-  if (immediateResult !== null) {
-    if (immediateResult === kEosImmediateClose) {
-      process.nextTick(() => callback.call(stream));
-    } else {
-      process.nextTick(() => callback.call(stream, immediateResult));
-    }
-  }
-
-  const cleanup = () => {
+  cleanup = () => {
     callback = nop;
     stream.removeListener('aborted', onclose);
     stream.removeListener('complete', onfinish);
@@ -336,7 +324,7 @@ function eos(stream, options, callback) {
     stream.removeListener('close', onclose);
   };
 
-  if (options.signal && !closed) {
+  if (options.signal) {
     const abort = () => {
       // Keep it because cleanup removes it.
       const endCallback = callback;
@@ -345,23 +333,27 @@ function eos(stream, options, callback) {
         stream,
         new AbortError(undefined, { cause: options.signal.reason }));
     };
-    if (options.signal.aborted) {
-      process.nextTick(abort);
-    } else {
-      addAbortListener ??= require('internal/events/abort_listener').addAbortListener;
-      const disposable = addAbortListener(options.signal, abort);
-      const originalCallback = callback;
-      callback = once((...args) => {
-        disposable[SymbolDispose]();
-        ReflectApply(originalCallback, stream, args);
-      });
-    }
+    addAbortListener ??= require('internal/events/abort_listener').addAbortListener;
+    const disposable = addAbortListener(options.signal, abort);
+    const originalCallback = callback;
+    callback = once((...args) => {
+      disposable[SymbolDispose]();
+      ReflectApply(originalCallback, stream, args);
+    });
   }
 
   return cleanup;
 }
 
 function eosWeb(stream, options, callback) {
+  if (AsyncContextFrame.current() || enabledHooksExist()) {
+    // Avoid AsyncResource.bind() because it calls ObjectDefineProperties which
+    // is a bottleneck here.
+    callback = once(bindAsyncResource(callback, 'STREAM_END_OF_STREAM'));
+  } else {
+    callback = once(callback);
+  }
+
   let isAborted = false;
   let abort = nop;
   if (options.signal) {
@@ -420,6 +412,5 @@ function finished(stream, opts) {
 module.exports = {
   eos,
   finished,
-  getEosImmediateResult,
-  kEosImmediateClose,
+  kEosNodeSyncronousCallback,
 };