stream: added experimental support for for-await

Adds support for Symbol.asyncIterator into the Readable class.
The stream is destroyed when the loop terminates with break or throw.

Fixes: #15709

PR-URL: #17755
Fixes: #15709
Reviewed-By: Benjamin Gruenbaum <[email protected]>
Reviewed-By: Anatoli Papirovski <[email protected]>
Reviewed-By: James M Snell <[email protected]>
Reviewed-By: Vse Mozhet Byt <[email protected]>
Reviewed-By: Michaël Zasso <[email protected]>

Author: mcollina

doc/api/stream.md

@@ -1165,6 +1165,31 @@ readable stream will release any internal resources.
 Implementors should not override this method, but instead implement
 [`readable._destroy`][readable-_destroy].
 
+##### readable[@@asyncIterator]
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+Returns an [AsyncIterator][async-iterator] to fully consume the stream.
+
+```js
+async function print(readable) {
+  readable.setEncoding('utf8');
+  let data = '';
+  for await (const k of readable) {
+    data += k;
+  }
+  console.log(data);
+}
+
+print(fs.createReadStream('file')).catch(console.log);
+```
+
+If the loop terminates with a `break` or a `throw`, the stream will be destroyed.
+In other terms, iterating over a stream will consume the stream fully.
+
 ### Duplex and Transform Streams
 
 #### Class: stream.Duplex
@@ -2328,3 +2353,4 @@ contain multi-byte characters.
 [readable-destroy]: #stream_readable_destroy_error
 [writable-_destroy]: #stream_writable_destroy_err_callback
 [writable-destroy]: #stream_writable_destroy_error
+[async-iterator]: https://github.com/tc39/proposal-async-iteration

lib/_stream_readable.js

@@ -32,6 +32,8 @@ const debug = util.debuglog('stream');
 const BufferList = require('internal/streams/BufferList');
 const destroyImpl = require('internal/streams/destroy');
 const errors = require('internal/errors');
+const ReadableAsyncIterator = require('internal/streams/async_iterator');
+const { emitExperimentalWarning } = require('internal/util');
 var StringDecoder;
 
 util.inherits(Readable, Stream);
@@ -922,6 +924,12 @@ Readable.prototype.wrap = function(stream) {
   return this;
 };
 
+Readable.prototype[Symbol.asyncIterator] = function() {
+  emitExperimentalWarning('Readable[Symbol.asyncIterator]');
+
+  return new ReadableAsyncIterator(this);
+};
+
 Object.defineProperty(Readable.prototype, 'readableHighWaterMark', {
   // making it explicit this property is not enumerable
   // because otherwise some prototype manipulation in

lib/internal/streams/async_iterator.js

@@ -0,0 +1,159 @@
+'use strict';
+
+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 AsyncIteratorRecord = class AsyncIteratorRecord {
+  constructor(value, done) {
+    this.done = done;
+    this.value = value;
+  }
+};
+
+function readAndResolve(iter) {
+  const resolve = iter[kLastResolve];
+  if (resolve !== null) {
+    const data = iter[kStream].read();
+    // we defer if data is null
+    // we can be expecting either 'end' or
+    // 'error'
+    if (data !== null) {
+      iter[kLastPromise] = null;
+      iter[kLastResolve] = null;
+      iter[kLastReject] = null;
+      resolve(new AsyncIteratorRecord(data, false));
+    }
+  }
+}
+
+function onReadable(iter) {
+  // we wait for the next tick, because it might
+  // emit an error with process.nextTick
+  process.nextTick(readAndResolve, iter);
+}
+
+function onEnd(iter) {
+  const resolve = iter[kLastResolve];
+  if (resolve !== null) {
+    iter[kLastPromise] = null;
+    iter[kLastResolve] = null;
+    iter[kLastReject] = null;
+    resolve(new AsyncIteratorRecord(null, true));
+  }
+  iter[kEnded] = true;
+}
+
+function onError(iter, err) {
+  const reject = iter[kLastReject];
+  // reject if we are waiting for data in the Promise
+  // returned by next() and store the error
+  if (reject !== null) {
+    iter[kLastPromise] = null;
+    iter[kLastResolve] = null;
+    iter[kLastReject] = null;
+    reject(err);
+  }
+  iter.error = err;
+}
+
+function wrapForNext(lastPromise, iter) {
+  return function(resolve, reject) {
+    lastPromise.then(function() {
+      iter[kHandlePromise](resolve, reject);
+    }, reject);
+  };
+}
+
+const ReadableAsyncIterator = class ReadableAsyncIterator {
+  constructor(stream) {
+    this[kStream] = stream;
+    this[kLastResolve] = null;
+    this[kLastReject] = null;
+    this[kError] = null;
+    this[kEnded] = false;
+    this[kLastPromise] = null;
+
+    stream.on('readable', onReadable.bind(null, this));
+    stream.on('end', onEnd.bind(null, this));
+    stream.on('error', onError.bind(null, this));
+
+    // the function passed to new Promise
+    // is cached so we avoid allocating a new
+    // closure at every run
+    this[kHandlePromise] = (resolve, reject) => {
+      const data = this[kStream].read();
+      if (data) {
+        this[kLastPromise] = null;
+        this[kLastResolve] = null;
+        this[kLastReject] = null;
+        resolve(new AsyncIteratorRecord(data, false));
+      } else {
+        this[kLastResolve] = resolve;
+        this[kLastReject] = reject;
+      }
+    };
+  }
+
+  get stream() {
+    return this[kStream];
+  }
+
+  next() {
+    // if we have detected an error in the meanwhile
+    // reject straight away
+    const error = this[kError];
+    if (error !== null) {
+      return Promise.reject(error);
+    }
+
+    if (this[kEnded]) {
+      return Promise.resolve(new AsyncIteratorRecord(null, true));
+    }
+
+    // if we have multiple next() calls
+    // we will wait for the previous Promise to finish
+    // this logic is optimized to support for await loops,
+    // where next() is only called once at a time
+    const lastPromise = this[kLastPromise];
+    let promise;
+
+    if (lastPromise) {
+      promise = new Promise(wrapForNext(lastPromise, this));
+    } else {
+      // fast path needed to support multiple this.push()
+      // without triggering the next() queue
+      const data = this[kStream].read();
+      if (data !== null) {
+        return Promise.resolve(new AsyncIteratorRecord(data, false));
+      }
+
+      promise = new Promise(this[kHandlePromise]);
+    }
+
+    this[kLastPromise] = promise;
+
+    return promise;
+  }
+
+  return() {
+    // destroy(err, cb) is a private API
+    // we can guarantee we have that here, because we control the
+    // Readable class this is attached to
+    return new Promise((resolve, reject) => {
+      this[kStream].destroy(null, (err) => {
+        if (err) {
+          reject(err);
+          return;
+        }
+        resolve(new AsyncIteratorRecord(null, true));
+      });
+    });
+  }
+};
+
+module.exports = ReadableAsyncIterator;

node.gyp

@@ -138,6 +138,7 @@
       'lib/internal/v8_prof_polyfill.js',
       'lib/internal/v8_prof_processor.js',
       'lib/internal/streams/lazy_transform.js',
+      'lib/internal/streams/async_iterator.js',
       'lib/internal/streams/BufferList.js',
       'lib/internal/streams/legacy.js',
       'lib/internal/streams/destroy.js',

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

@@ -0,0 +1,298 @@
+'use strict';
+
+const common = require('../common');
+const { Readable } = require('stream');
+const assert = require('assert');
+
+common.crashOnUnhandledRejection();
+
+async function tests() {
+  await (async function() {
+    console.log('read without for..await');
+    const max = 5;
+    const readable = new Readable({
+      objectMode: true,
+      read() {}
+    });
+
+    const iter = readable[Symbol.asyncIterator]();
+    assert.strictEqual(iter.stream, readable);
+    const values = [];
+    for (let i = 0; i < max; i++) {
+      values.push(iter.next());
+    }
+    Promise.all(values).then(common.mustCall((values) => {
+      values.forEach(common.mustCall(
+        (item, i) => assert.strictEqual(item.value, 'hello-' + i), 5));
+    }));
+
+    readable.push('hello-0');
+    readable.push('hello-1');
+    readable.push('hello-2');
+    readable.push('hello-3');
+    readable.push('hello-4');
+    readable.push(null);
+
+    const last = await iter.next();
+    assert.strictEqual(last.done, true);
+  })();
+
+  await (async function() {
+    console.log('read without for..await deferred');
+    const readable = new Readable({
+      objectMode: true,
+      read() {}
+    });
+
+    const iter = readable[Symbol.asyncIterator]();
+    assert.strictEqual(iter.stream, readable);
+    let values = [];
+    for (let i = 0; i < 3; i++) {
+      values.push(iter.next());
+    }
+
+    readable.push('hello-0');
+    readable.push('hello-1');
+    readable.push('hello-2');
+
+    let k = 0;
+    const results1 = await Promise.all(values);
+    results1.forEach(common.mustCall(
+      (item) => assert.strictEqual(item.value, 'hello-' + k++), 3));
+
+    values = [];
+    for (let i = 0; i < 2; i++) {
+      values.push(iter.next());
+    }
+
+    readable.push('hello-3');
+    readable.push('hello-4');
+    readable.push(null);
+
+    const results2 = await Promise.all(values);
+    results2.forEach(common.mustCall(
+      (item) => assert.strictEqual(item.value, 'hello-' + k++), 2));
+
+    const last = await iter.next();
+    assert.strictEqual(last.done, true);
+  })();
+
+  await (async function() {
+    console.log('read without for..await with errors');
+    const max = 3;
+    const readable = new Readable({
+      objectMode: true,
+      read() {}
+    });
+
+    const iter = readable[Symbol.asyncIterator]();
+    assert.strictEqual(iter.stream, readable);
+    const values = [];
+    const errors = [];
+    let i;
+    for (i = 0; i < max; i++) {
+      values.push(iter.next());
+    }
+    for (i = 0; i < 2; i++) {
+      errors.push(iter.next());
+    }
+
+    readable.push('hello-0');
+    readable.push('hello-1');
+    readable.push('hello-2');
+
+    const resolved = await Promise.all(values);
+
+    resolved.forEach(common.mustCall(
+      (item, i) => assert.strictEqual(item.value, 'hello-' + i), max));
+
+    errors.forEach((promise) => {
+      promise.catch(common.mustCall((err) => {
+        assert.strictEqual(err.message, 'kaboom');
+      }));
+    });
+
+    readable.destroy(new Error('kaboom'));
+  })();
+
+  await (async function() {
+    console.log('read object mode');
+    const max = 42;
+    let readed = 0;
+    let received = 0;
+    const readable = new Readable({
+      objectMode: true,
+      read() {
+        this.push('hello');
+        if (++readed === max) {
+          this.push(null);
+        }
+      }
+    });
+
+    for await (const k of readable) {
+      received++;
+      assert.strictEqual(k, 'hello');
+    }
+
+    assert.strictEqual(readed, received);
+  })();
+
+  await (async function() {
+    console.log('destroy sync');
+    const readable = new Readable({
+      objectMode: true,
+      read() {
+        this.destroy(new Error('kaboom from read'));
+      }
+    });
+
+    let err;
+    try {
+      // eslint-disable-next-line no-unused-vars
+      for await (const k of readable) {}
+    } catch (e) {
+      err = e;
+    }
+    assert.strictEqual(err.message, 'kaboom from read');
+  })();
+
+  await (async function() {
+    console.log('destroy async');
+    const readable = new Readable({
+      objectMode: true,
+      read() {
+        if (!this.pushed) {
+          this.push('hello');
+          this.pushed = true;
+
+          setImmediate(() => {
+            this.destroy(new Error('kaboom'));
+          });
+        }
+      }
+    });
+
+    let received = 0;
+
+    let err = null;
+    try {
+      // eslint-disable-next-line no-unused-vars
+      for await (const k of readable) {
+        received++;
+      }
+    } catch (e) {
+      err = e;
+    }
+
+    assert.strictEqual(err.message, 'kaboom');
+    assert.strictEqual(received, 1);
+  })();
+
+  await (async function() {
+    console.log('destroyed by throw');
+    const readable = new Readable({
+      objectMode: true,
+      read() {
+        this.push('hello');
+      }
+    });
+
+    let err = null;
+    try {
+      for await (const k of readable) {
+        assert.strictEqual(k, 'hello');
+        throw new Error('kaboom');
+      }
+    } catch (e) {
+      err = e;
+    }
+
+    assert.strictEqual(err.message, 'kaboom');
+    assert.strictEqual(readable.destroyed, true);
+  })();
+
+  await (async function() {
+    console.log('destroyed sync after push');
+    const readable = new Readable({
+      objectMode: true,
+      read() {
+        this.push('hello');
+        this.destroy(new Error('kaboom'));
+      }
+    });
+
+    let received = 0;
+
+    let err = null;
+    try {
+      for await (const k of readable) {
+        assert.strictEqual(k, 'hello');
+        received++;
+      }
+    } catch (e) {
+      err = e;
+    }
+
+    assert.strictEqual(err.message, 'kaboom');
+    assert.strictEqual(received, 1);
+  })();
+
+  await (async function() {
+    console.log('push async');
+    const max = 42;
+    let readed = 0;
+    let received = 0;
+    const readable = new Readable({
+      objectMode: true,
+      read() {
+        setImmediate(() => {
+          this.push('hello');
+          if (++readed === max) {
+            this.push(null);
+          }
+        });
+      }
+    });
+
+    for await (const k of readable) {
+      received++;
+      assert.strictEqual(k, 'hello');
+    }
+
+    assert.strictEqual(readed, received);
+  })();
+
+  await (async function() {
+    console.log('push binary async');
+    const max = 42;
+    let readed = 0;
+    const readable = new Readable({
+      read() {
+        setImmediate(() => {
+          this.push('hello');
+          if (++readed === max) {
+            this.push(null);
+          }
+        });
+      }
+    });
+
+    let expected = '';
+    readable.setEncoding('utf8');
+    readable.pause();
+    readable.on('data', (chunk) => {
+      expected += chunk;
+    });
+
+    let data = '';
+    for await (const k of readable) {
+      data += k;
+    }
+
+    assert.strictEqual(data, expected);
+  })();
+}
+
+// to avoid missing some tests if a promise does not resolve
+tests().then(common.mustCall());