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 <matteo.collina@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: Ruben Bridgewater <ruben@bridgewater.de>

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