stream: implement throw for async iterator

PR-URL: #31316 Reviewed-By: Anna Henningsen <[email protected]> Reviewed-By: Matteo Collina <[email protected]> Reviewed-By: Rich Trott <[email protected]> Reviewed-By: Minwoo Jung <[email protected]>
nodejs · Feb 17, 2020 · 3046648 · 3046648
1 parent 5a95fa4
commit 3046648
Show file tree

Hide file tree

Showing 2 changed files with 51 additions and 20 deletions.
diff --git a/lib/internal/streams/async_iterator.js b/lib/internal/streams/async_iterator.js
@@ -71,6 +71,31 @@ function wrapForNext(lastPromise, iter) {
 const AsyncIteratorPrototype = ObjectGetPrototypeOf(
   ObjectGetPrototypeOf(async function* () {}).prototype);
 
+function finish(self, err) {
+  return new Promise((resolve, reject) => {
+    const stream = self[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);
+      } else {
+        resolve(createIterResult(undefined, true));
+      }
+    });
+    destroy(stream, err);
+  });
+}
+
 const ReadableStreamAsyncIteratorPrototype = ObjectSetPrototypeOf({
   get stream() {
     return this[kStream];
@@ -131,27 +156,11 @@ const ReadableStreamAsyncIteratorPrototype = ObjectSetPrototypeOf({
   },
 
   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;
-      }
+    return finish(this);
+  },
 
-      finished(stream, (err) => {
-        if (err && err.code !== 'ERR_STREAM_PREMATURE_CLOSE') {
-          reject(err);
-        } else {
-          resolve(createIterResult(undefined, true));
-        }
-      });
-      destroy(stream);
-    });
+  throw(err) {
+    return finish(this, err);
   },
 }, AsyncIteratorPrototype);
 

diff --git a/test/parallel/test-stream-readable-async-iterators.js b/test/parallel/test-stream-readable-async-iterators.js
@@ -282,6 +282,28 @@ async function tests() {
     assert.strictEqual(received, 1);
   }
 
+  {
+    // Iterator throw.
+
+    const readable = new Readable({
+      objectMode: true,
+      read() {
+        this.push('hello');
+      }
+    });
+
+    readable.on('error', common.mustCall((err) => {
+      assert.strictEqual(err.message, 'kaboom');
+    }));
+
+    const it = readable[Symbol.asyncIterator]();
+    it.throw(new Error('kaboom')).catch(common.mustCall((err) => {
+      assert.strictEqual(err.message, 'kaboom');
+    }));
+
+    assert.strictEqual(readable.destroyed, true);
+  }
+
   {
     console.log('destroyed by throw');
     const readable = new Readable({