stream: make from read one at a time

Currently from will eagerly buffer up items which means that errors are also eagerly encountered and items which are buffer when an error occurs will be discarded, which is inconsistent with how generators work. Fixes: #29428 PR-URL: #33201 Reviewed-By: Matteo Collina <[email protected]> Reviewed-By: James M Snell <[email protected]> Reviewed-By: Benjamin Gruenbaum <[email protected]> Reviewed-By: Ruben Bridgewater <[email protected]>
nodejs · May 6, 2020 · 8607f9e · 8607f9e
1 parent 3662b0c
commit 8607f9e
Show file tree

Hide file tree

Showing 2 changed files with 27 additions and 1 deletion.
diff --git a/lib/internal/streams/from.js b/lib/internal/streams/from.js
@@ -33,6 +33,8 @@ function from(Readable, iterable, opts) {
 
   const readable = new Readable({
     objectMode: true,
+    highWaterMark: 1,
+    // TODO(ronag): What options should be allowed?
     ...opts
   });
 

diff --git a/test/parallel/test-readable-from.js b/test/parallel/test-readable-from.js
@@ -159,6 +159,29 @@ async function asTransformStream() {
   }
 }
 
+async function endWithError() {
+  async function* generate() {
+    yield 1;
+    yield 2;
+    yield Promise.reject('Boum');
+  }
+
+  const stream = Readable.from(generate());
+
+  const expected = [1, 2];
+
+  try {
+    for await (const chunk of stream) {
+      strictEqual(chunk, expected.shift());
+    }
+    throw new Error();
+  } catch (err) {
+    strictEqual(expected.length, 0);
+    strictEqual(err, 'Boum');
+  }
+}
+
+
 Promise.all([
   toReadableBasicSupport(),
   toReadableSyncIterator(),
@@ -168,5 +191,6 @@ Promise.all([
   toReadableOnData(),
   toReadableOnDataNonObject(),
   destroysTheStreamWhenThrowing(),
-  asTransformStream()
+  asTransformStream(),
+  endWithError()
 ]).then(mustCall());