[x] stream: error multiple iterators

doc/api/errors.md

@@ -1703,6 +1703,11 @@ An attempt was made to call [`stream.pipe()`][] on a [`Writable`][] stream.
 A stream method was called that cannot complete because the stream was
 destroyed using `stream.destroy()`.
 
+<a id="ERR_STREAM_ITERATOR_EXISTS"></a>
+### ERR_STREAM_ITERATOR_EXISTS
+
+Stream cannot be consumed by multiple iterators.
+
 <a id="ERR_STREAM_NULL_VALUES"></a>
 ### ERR_STREAM_NULL_VALUES
 

lib/internal/errors.js

@@ -1130,6 +1130,7 @@ E('ERR_SRI_PARSE',
   SyntaxError);
 E('ERR_STREAM_CANNOT_PIPE', 'Cannot pipe, not readable', Error);
 E('ERR_STREAM_DESTROYED', 'Cannot call %s after a stream was destroyed', Error);
+E('ERR_STREAM_ITERATOR_EXISTS', 'Iterator already exists', Error);
 E('ERR_STREAM_NULL_VALUES', 'May not write null values to stream', TypeError);
 E('ERR_STREAM_PREMATURE_CLOSE', 'Premature close', Error);
 E('ERR_STREAM_PUSH_AFTER_EOF', 'stream.push() after EOF', Error);

lib/internal/streams/async_iterator.js

@@ -4,13 +4,18 @@ const { Object } = primordials;
 
 const finished = require('internal/streams/end-of-stream');
 
+const {
+  ERR_STREAM_ITERATOR_EXISTS
+} = require('internal/errors').codes;
+
 const kLastResolve = Symbol('lastResolve');
 const kLastReject = Symbol('lastReject');
 const kError = Symbol('error');
 const kEnded = Symbol('ended');
 const kLastPromise = Symbol('lastPromise');
 const kHandlePromise = Symbol('handlePromise');
 const kStream = Symbol('stream');
+const kIterator = Symbol('iterator');
 
 function createIterResult(value, done) {
   return { value, done };
@@ -126,6 +131,10 @@ const ReadableStreamAsyncIteratorPrototype = Object.setPrototypeOf({
 }, AsyncIteratorPrototype);
 
 const createReadableStreamAsyncIterator = (stream) => {
+  if (stream[kIterator]) {
+    throw new ERR_STREAM_ITERATOR_EXISTS();
+  }
+
   const iterator = Object.create(ReadableStreamAsyncIteratorPrototype, {
     [kStream]: { value: stream, writable: true },
     [kLastResolve]: { value: null, writable: true },
@@ -182,6 +191,8 @@ const createReadableStreamAsyncIterator = (stream) => {
 
   stream.on('readable', onReadable.bind(null, iterator));
 
+  stream[kIterator] = true;
+
   return iterator;
 };
 

test/parallel/test-readline-async-iterators-destroy.js

@@ -58,24 +58,14 @@ async function testMutualDestroy() {
       crlfDelay: Infinity
     });
 
-    const expectedLines = fileContent.split('\n');
-    if (expectedLines[expectedLines.length - 1] === '') {
-      expectedLines.pop();
-    }
-    expectedLines.splice(2);
-
-    const iteratedLines = [];
     for await (const k of rli) {
-      iteratedLines.push(k);
-      for await (const l of rli) {
-        iteratedLines.push(l);
-        break;
+      try {
+        for await (const l of rli) {
+        }
+      } catch (err) {
+        assert.strictEqual(err.code, 'ERR_STREAM_ITERATOR_EXISTS');
       }
-      assert.deepStrictEqual(iteratedLines, expectedLines);
     }
-
-    assert.deepStrictEqual(iteratedLines, expectedLines);
-
     rli.close();
     readable.destroy();
   }

test/parallel/test-readline-async-iterators.js

@@ -43,35 +43,4 @@ async function testSimple() {
   }
 }
 
-async function testMutual() {
-  for (const fileContent of testContents) {
-    fs.writeFileSync(filename, fileContent);
-
-    const readable = fs.createReadStream(filename);
-    const rli = readline.createInterface({
-      input: readable,
-      crlfDelay: Infinity
-    });
-
-    const expectedLines = fileContent.split('\n');
-    if (expectedLines[expectedLines.length - 1] === '') {
-      expectedLines.pop();
-    }
-    const iteratedLines = [];
-    let iterated = false;
-    for await (const k of rli) {
-      // This outer loop should only iterate once.
-      assert.strictEqual(iterated, false);
-      iterated = true;
-
-      iteratedLines.push(k);
-      for await (const l of rli) {
-        iteratedLines.push(l);
-      }
-      assert.deepStrictEqual(iteratedLines, expectedLines);
-    }
-    assert.deepStrictEqual(iteratedLines, expectedLines);
-  }
-}
-
-testSimple().then(testMutual).then(common.mustCall());
+testSimple().then(common.mustCall());

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

@@ -375,11 +375,26 @@ async function tests() {
     // eslint-disable-next-line no-unused-vars
     for await (const a of r) {
     }
-    // eslint-disable-next-line no-unused-vars
-    for await (const b of r) {
+    try {
+      // eslint-disable-next-line no-unused-vars
+      for await (const b of r) {
+      }
+    } catch (err) {
+      assert.strictEqual(err.code, 'ERR_STREAM_ITERATOR_EXISTS');
     }
   }
 
+  {
+    console.log('creating multiple iterators');
+    const r = new Readable();
+    r[Symbol.asyncIterator]();
+    common.expectsError(() => {
+      r[Symbol.asyncIterator]();
+    }, {
+      code: 'ERR_STREAM_ITERATOR_EXISTS'
+    });
+  }
+
   {
     console.log('destroy mid-stream does not error');
     const r = new Readable({