stream: add null push transform in async_iterator

when the readable side of a transform ends any for await
loop on that transform stream should also complete. This
fix prevents for await loop on a transform stream
from hanging indefinitely.

PR-URL: https://github.com/nodejs/node/pull/28566
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: Jeremiah Senkpiel <fishrock123@rocketmail.com>
Reviewed-By: Ruben Bridgewater <ruben@bridgewater.de>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Rich Trott <rtrott@gmail.com>

2 files changed, 26 insertions, 2 deletions

diff --git a/lib/internal/streams/async_iterator.js b/lib/internal/streams/async_iterator.js
index defba235ad..89a1dae7fd 100644
--- a/lib/internal/streams/async_iterator.js
+++ b/lib/internal/streams/async_iterator.js
@@ -155,7 +155,7 @@ const createReadableStreamAsyncIterator = (stream) => {
   });
   iterator[kLastPromise] = null;
 
-  finished(stream, (err) => {
+  finished(stream, { writable: false }, (err) => {
     if (err && err.code !== 'ERR_STREAM_PREMATURE_CLOSE') {
       const reject = iterator[kLastReject];
       // Reject if we are waiting for data in the Promise returned by next() and
diff --git a/test/parallel/test-stream-readable-async-iterators.js b/test/parallel/test-stream-readable-async-iterators.js
index b50a6b1734..12971cb236 100644
--- a/test/parallel/test-stream-readable-async-iterators.js
+++ b/test/parallel/test-stream-readable-async-iterators.js
@@ -1,7 +1,7 @@
 'use strict';
 
 const common = require('../common');
-const { Readable, PassThrough, pipeline } = require('stream');
+const { Readable, Transform, PassThrough, pipeline } = require('stream');
 const assert = require('assert');
 
 async function tests() {
@@ -397,6 +397,30 @@ async function tests() {
   }
 
   {
+    console.log('readable side of a transform stream pushes null');
+    const transform = new Transform({
+      objectMode: true,
+      transform: (chunk, enc, cb) => { cb(null, chunk); }
+    });
+    transform.push(0);
+    transform.push(1);
+    process.nextTick(() => {
+      transform.push(null);
+    });
+
+    const mustReach = [ common.mustCall(), common.mustCall() ];
+
+    const iter = transform[Symbol.asyncIterator]();
+    assert.strictEqual((await iter.next()).value, 0);
+
+    for await (const d of iter) {
+      assert.strictEqual(d, 1);
+      mustReach[0]();
+    }
+    mustReach[1]();
+  }
+
+  {
     console.log('all next promises must be resolved on end');
     const r = new Readable({
       objectMode: true,