stream: make finished call the callback if the stream is closed

Make stream.finished callback invoked if stream is already
closed/destroyed.

PR-URL: https://github.com/nodejs/node/pull/28748
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>

2 files changed, 23 insertions, 27 deletions

diff --git a/lib/internal/streams/async_iterator.js b/lib/internal/streams/async_iterator.js
index 07f2191e71..083befb89c 100644
--- a/lib/internal/streams/async_iterator.js
+++ b/lib/internal/streams/async_iterator.js
@@ -112,16 +112,6 @@ const ReadableStreamAsyncIteratorPrototype = Object.setPrototypeOf({
   return() {
     return new Promise((resolve, reject) => {
       const stream = this[kStream];
-
-      // TODO(ronag): Remove this check once finished() handles
-      // already ended and/or destroyed streams.
-      const ended = stream.destroyed || stream.readableEnded ||
-        (stream._readableState && stream._readableState.endEmitted);
-      if (ended) {
-        resolve(createIterResult(undefined, true));
-        return;
-      }
-
       finished(stream, (err) => {
         if (err && err.code !== 'ERR_STREAM_PREMATURE_CLOSE') {
           reject(err);
diff --git a/lib/internal/streams/end-of-stream.js b/lib/internal/streams/end-of-stream.js
index 3f1c0f316c..ca6091fe55 100644
--- a/lib/internal/streams/end-of-stream.js
+++ b/lib/internal/streams/end-of-stream.js
@@ -28,6 +28,25 @@ function eos(stream, opts, callback) {
 
   callback = once(callback);
 
+  const onerror = (err) => {
+    callback.call(stream, err);
+  };
+
+  let writableFinished = stream.writableFinished ||
+    (stream._writableState && stream._writableState.finished);
+  let readableEnded = stream.readableEnded ||
+    (stream._readableState && stream._readableState.endEmitted);
+
+  if (writableFinished || readableEnded || stream.destroyed ||
+      stream.aborted) {
+    if (opts.error !== false) stream.on('error', onerror);
+    // A destroy(err) call emits error in nextTick.
+    process.nextTick(callback.bind(stream));
+    return () => {
+      stream.removeListener('error', onerror);
+    };
+  }
+
   let readable = opts.readable || (opts.readable !== false && stream.readable);
   let writable = opts.writable || (opts.writable !== false && stream.writable);
 
@@ -35,36 +54,23 @@ function eos(stream, opts, callback) {
     if (!stream.writable) onfinish();
   };
 
-  var writableEnded = stream._writableState && stream._writableState.finished;
   const onfinish = () => {
     writable = false;
-    writableEnded = true;
+    writableFinished = true;
     if (!readable) callback.call(stream);
   };
 
-  var readableEnded = stream.readableEnded ||
-    (stream._readableState && stream._readableState.endEmitted);
   const onend = () => {
     readable = false;
     readableEnded = true;
     if (!writable) callback.call(stream);
   };
 
-  const onerror = (err) => {
-    callback.call(stream, err);
-  };
-
   const onclose = () => {
-    let err;
     if (readable && !readableEnded) {
-      if (!stream._readableState || !stream._readableState.ended)
-        err = new ERR_STREAM_PREMATURE_CLOSE();
-      return callback.call(stream, err);
-    }
-    if (writable && !writableEnded) {
-      if (!stream._writableState || !stream._writableState.ended)
-        err = new ERR_STREAM_PREMATURE_CLOSE();
-      return callback.call(stream, err);
+      callback.call(stream, new ERR_STREAM_PREMATURE_CLOSE());
+    } else if (writable && !writableFinished) {
+      callback.call(stream, new ERR_STREAM_PREMATURE_CLOSE());
     }
   };