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]>

Author: ronag

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
   });
 

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());