stream: add null check in Readable.from

Throws `ERR_STREAM_NULL_VALUES` error if a null value is passed to
`Readable.from`. Also added docs for the same.

Co-Authored-By: 扩散性百万甜面包 <himself65@outlook.com>
Fixes: #32845
PR-URL: #32873
Reviewed-By: Ruben Bridgewater <ruben@bridgewater.de>
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: Robert Nagy <ronagy@icloud.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>

doc/api/stream.md

@@ -1621,7 +1621,8 @@ added:
 -->
 
 * `iterable` {Iterable} Object implementing the `Symbol.asyncIterator` or
-  `Symbol.iterator` iterable protocol.
+  `Symbol.iterator` iterable protocol. Emits an 'error' event if a null
+   value is passed.
 * `options` {Object} Options provided to `new stream.Readable([options])`.
   By default, `Readable.from()` will set `options.objectMode` to `true`, unless
   this is explicitly opted out by setting `options.objectMode` to `false`.

lib/internal/streams/from.js

@@ -7,7 +7,8 @@ const {
 const { Buffer } = require('buffer');
 
 const {
-  ERR_INVALID_ARG_TYPE
+  ERR_INVALID_ARG_TYPE,
+  ERR_STREAM_NULL_VALUES
 } = require('internal/errors').codes;
 
 function from(Readable, iterable, opts) {
@@ -73,15 +74,20 @@ function from(Readable, iterable, opts) {
       needToClose = false;
       const { value, done } = await iterator.next();
       needToClose = !done;
-      const resolved = await value;
       if (done) {
         readable.push(null);
       } else if (readable.destroyed) {
         await close();
-      } else if (readable.push(resolved)) {
-        next();
       } else {
-        reading = false;
+        const res = await value;
+        if (res === null) {
+          reading = false;
+          throw new ERR_STREAM_NULL_VALUES();
+        } else if (readable.push(res)) {
+          next();
+        } else {
+          reading = false;
+        }
       }
     } catch (err) {
       readable.destroy(err);

test/parallel/test-readable-from-iterator-closing.js

@@ -168,18 +168,17 @@ async function closeAfterNullYielded() {
   const finallyMustCall = mustCall();
   const dataMustCall = mustCall(3);
 
-  function* infiniteGenerate() {
+  function* generate() {
     try {
       yield 'a';
       yield 'a';
       yield 'a';
-      while (true) yield null;
     } finally {
       finallyMustCall();
     }
   }
 
-  const stream = Readable.from(infiniteGenerate());
+  const stream = Readable.from(generate());
 
   stream.on('data', (chunk) => {
     dataMustCall();

test/parallel/test-stream-readable-next-no-null.js

@@ -0,0 +1,19 @@
+'use strict';
+const { mustNotCall, expectsError } = require('../common');
+const { Readable } = require('stream');
+
+async function* generate() {
+  yield null;
+}
+
+const stream = Readable.from(generate());
+
+stream.on('error', expectsError({
+  code: 'ERR_STREAM_NULL_VALUES',
+  name: 'TypeError',
+  message: 'May not write null values to stream'
+}));
+
+stream.on('data', mustNotCall((chunk) => {}));
+
+stream.on('end', mustNotCall());