stream: extract Readable.from in its own file

See: nodejs/readable-stream#420

PR-URL: #30140
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: Colin Ihrig <cjihrig@gmail.com>
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>
Reviewed-By: Gus Caplan <me@gus.host>
Reviewed-By: Beth Griggs <Bethany.Griggs@uk.ibm.com>

Author: mcollina

lib/_stream_readable.js

@@ -47,6 +47,7 @@ const {
 // Lazy loaded to improve the startup performance.
 let StringDecoder;
 let createReadableStreamAsyncIterator;
+let from;
 
 Object.setPrototypeOf(Readable.prototype, Stream.prototype);
 Object.setPrototypeOf(Readable, Stream);
@@ -1209,40 +1210,8 @@ function endReadableNT(state, stream) {
 }
 
 Readable.from = function(iterable, opts) {
-  let iterator;
-  if (iterable && iterable[Symbol.asyncIterator])
-    iterator = iterable[Symbol.asyncIterator]();
-  else if (iterable && iterable[Symbol.iterator])
-    iterator = iterable[Symbol.iterator]();
-  else
-    throw new ERR_INVALID_ARG_TYPE('iterable', ['Iterable'], iterable);
-
-  const readable = new Readable({
-    objectMode: true,
-    ...opts
-  });
-  // Reading boolean to protect against _read
-  // being called before last iteration completion.
-  let reading = false;
-  readable._read = function() {
-    if (!reading) {
-      reading = true;
-      next();
-    }
-  };
-  async function next() {
-    try {
-      const { value, done } = await iterator.next();
-      if (done) {
-        readable.push(null);
-      } else if (readable.push(await value)) {
-        next();
-      } else {
-        reading = false;
-      }
-    } catch (err) {
-      readable.destroy(err);
-    }
+  if (from === undefined) {
+    from = require('internal/streams/from');
   }
-  return readable;
+  return from(Readable, iterable, opts);
 };

lib/internal/streams/from.js

@@ -0,0 +1,46 @@
+'use strict';
+
+const {
+  ERR_INVALID_ARG_TYPE
+} = require('internal/errors').codes;
+
+function from(Readable, iterable, opts) {
+  let iterator;
+  if (iterable && iterable[Symbol.asyncIterator])
+    iterator = iterable[Symbol.asyncIterator]();
+  else if (iterable && iterable[Symbol.iterator])
+    iterator = iterable[Symbol.iterator]();
+  else
+    throw new ERR_INVALID_ARG_TYPE('iterable', ['Iterable'], iterable);
+
+  const readable = new Readable({
+    objectMode: true,
+    ...opts
+  });
+  // Reading boolean to protect against _read
+  // being called before last iteration completion.
+  let reading = false;
+  readable._read = function() {
+    if (!reading) {
+      reading = true;
+      next();
+    }
+  };
+  async function next() {
+    try {
+      const { value, done } = await iterator.next();
+      if (done) {
+        readable.push(null);
+      } else if (readable.push(await value)) {
+        next();
+      } else {
+        reading = false;
+      }
+    } catch (err) {
+      readable.destroy(err);
+    }
+  }
+  return readable;
+}
+
+module.exports = from;

node.gyp

@@ -205,6 +205,7 @@
       'lib/internal/streams/async_iterator.js',
       'lib/internal/streams/buffer_list.js',
       'lib/internal/streams/duplexpair.js',
+      'lib/internal/streams/from.js',
       'lib/internal/streams/legacy.js',
       'lib/internal/streams/destroy.js',
       'lib/internal/streams/state.js',