stream: asyncIterator without for..await

nodejs · Dec 19, 2017 · 02b6336 · 02b6336
1 parent 6395581
commit 02b6336
Show file tree

Hide file tree

Showing 2 changed files with 45 additions and 9 deletions.
diff --git a/lib/internal/streams/async_iterator.js b/lib/internal/streams/async_iterator.js
@@ -30,6 +30,7 @@ class ReadableAsyncIterator {
     this.lastReject = null;
     this.error = null;
     this.ended = false;
+    this.lastPromise = null;
 
     stream.on('readable', () => {
       // we wait for the next tick, because it might
@@ -60,12 +61,18 @@ class ReadableAsyncIterator {
     // the function passed to new Promise
     // is cached so we avoid allocating a new
     // closure at every run
-    this._handlePromise = (resolve, reject) => {
+    this._handlePromise = (resolve, reject, promise) => {
+      if (promise === this.lastPromise) {
+        this.lastPromise = null;
+      } else if (promise === undefined && this.lastPromise !== null) {
+        const boundFn =
+          this._handlePromise.bind(this, resolve, reject, this.lastPromise);
+        this.lastPromise.then(boundFn, boundFn);
+        return;
+      }
       const data = this.stream.read();
       if (data) {
         resolve(new Item(data, false));
-      } else if (this.lastResolve !== null) {
-        throw new Error('next can be called only once');
       } else {
         this.lastResolve = resolve;
         this.lastReject = reject;
@@ -76,15 +83,13 @@ class ReadableAsyncIterator {
   next() {
     // if we have detected an error in the meanwhile
     // reject straight away
-    if (this.error !== null) {
-      return Promise.reject(this.error);
-    }
+    if (this.error !== null)
+      throw Promise.reject(this.error);
 
-    if (this.ended) {
+    if (this.ended)
       return Promise.resolve(new Item(null, true));
-    }
 
-    return new Promise(this._handlePromise);
+    return this.lastPromise = new Promise(this._handlePromise);
   }
 
   async return() {

diff --git a/test/parallel/test-stream-readable-async-iterators.js b/test/parallel/test-stream-readable-async-iterators.js
@@ -7,6 +7,37 @@ const assert = require('assert');
 common.crashOnUnhandledRejection();
 
 async function tests () {
+  await (async function() {
+    console.log('read without for..await');
+    const max = 5;
+    let readed = 0;
+    let received = 0;
+    const readable = new Readable({
+      objectMode: true,
+      read() {}
+    });
+
+    const iter = readable[Symbol.asyncIterator]();
+    const values = [];
+    for (var i = 0; i < max; i++) {
+      values.push(iter.next());
+    }
+    Promise.all(values).then(common.mustCall((values) => {
+      values.forEach(common.mustCall(
+        item => assert.strictEqual(item.value, 'hello'), 5));
+    }));
+
+    readable.push('hello');
+    readable.push('hello');
+    readable.push('hello');
+    readable.push('hello');
+    readable.push('hello');
+    readable.push(null);
+
+    const last = await iter.next();
+    assert.strictEqual(last.done, true);
+  })();
+
   await (async function() {
     console.log('read object mode');
     const max = 42;