stream: use finished for async iteration

lib/internal/errors.js

@@ -142,7 +142,7 @@ const maybeOverridePrepareStackTrace = (globalThis, error, trace) => {
 };
 
 const aggregateTwoErrors = hideStackFrames((innerError, outerError) => {
-  if (innerError && outerError) {
+  if (innerError && outerError && innerError !== outerError) {
     if (ArrayIsArray(outerError.errors)) {
       // If `outerError` is already an `AggregateError`.
       ArrayPrototypePush(outerError.errors, innerError);

lib/internal/streams/readable.js

@@ -45,6 +45,7 @@ const { Buffer } = require('buffer');
 const {
   addAbortSignalNoValidate,
 } = require('internal/streams/add-abort-signal');
+const eos = require('internal/streams/end-of-stream');
 
 let debug = require('internal/util/debuglog').debuglog('stream', (fn) => {
   debug = fn;
@@ -57,12 +58,14 @@ const {
 } = require('internal/streams/state');
 
 const {
-  ERR_INVALID_ARG_TYPE,
-  ERR_METHOD_NOT_IMPLEMENTED,
-  ERR_STREAM_PREMATURE_CLOSE,
-  ERR_STREAM_PUSH_AFTER_EOF,
-  ERR_STREAM_UNSHIFT_AFTER_END_EVENT,
-} = require('internal/errors').codes;
+  aggregateTwoErrors,
+  codes: {
+    ERR_INVALID_ARG_TYPE,
+    ERR_METHOD_NOT_IMPLEMENTED,
+    ERR_STREAM_PUSH_AFTER_EOF,
+    ERR_STREAM_UNSHIFT_AFTER_END_EVENT,
+  }
+} = require('internal/errors');
 const { validateObject } = require('internal/validators');
 
 const kPaused = Symbol('kPaused');
@@ -1090,12 +1093,6 @@ function streamToAsyncIterator(stream, options) {
 async function* createAsyncIterator(stream, options) {
   let callback = nop;
 
-  const opts = {
-    destroyOnReturn: true,
-    destroyOnError: true,
-    ...options,
-  };
-
   function next(resolve) {
     if (this === stream) {
       callback();
@@ -1105,54 +1102,38 @@ async function* createAsyncIterator(stream, options) {
     }
   }
 
-  const state = stream._readableState;
+  stream.on('readable', next);
+
+  let error;
+  eos(stream, { writable: false }, (err) => {
+    error = err ? aggregateTwoErrors(error, err) : null;
+    callback();
+    callback = nop;
+  });
 
-  let error = state.errored;
-  let errorEmitted = state.errorEmitted;
-  let endEmitted = state.endEmitted;
-  let closeEmitted = state.closeEmitted;
-
-  stream
-    .on('readable', next)
-    .on('error', function(err) {
-      error = err;
-      errorEmitted = true;
-      next.call(this);
-    })
-    .on('end', function() {
-      endEmitted = true;
-      next.call(this);
-    })
-    .on('close', function() {
-      closeEmitted = true;
-      next.call(this);
-    });
-
-  let errorThrown = false;
   try {
     while (true) {
       const chunk = stream.destroyed ? null : stream.read();
       if (chunk !== null) {
         yield chunk;
-      } else if (errorEmitted) {
+      } else if (error) {
         throw error;
-      } else if (endEmitted) {
-        break;
-      } else if (closeEmitted) {
-        throw new ERR_STREAM_PREMATURE_CLOSE();
+      } else if (error === null) {
+        return;
       } else {
         await new Promise(next);
       }
     }
   } catch (err) {
-    if (opts.destroyOnError) {
-      destroyImpl.destroyer(stream, err);
-    }
-    errorThrown = true;
-    throw err;
+    error = aggregateTwoErrors(error, err);
+    throw error;
   } finally {
-    if (!errorThrown && opts.destroyOnReturn) {
-      if (state.autoDestroy || !endEmitted) {
+    if (error) {
+      if (options?.destroyOnError !== false) {
+        destroyImpl.destroyer(stream, error);
+      }
+    } else if (options?.destroyOnReturn !== false) {
+      if (error === undefined || stream._readableState.autoDestroy) {
         destroyImpl.destroyer(stream, null);
       }
     }

test/parallel/test-stream-readable-async-iterators.js

@@ -205,7 +205,7 @@ async function tests() {
     const iterator = readable[Symbol.asyncIterator]();
 
     const err = new Error('kaboom');
-    readable.destroy(new Error('kaboom'));
+    readable.destroy(err);
     await assert.rejects(iterator.next.bind(iterator), err);
   }