mirror of
https://github.com/zebrajr/node.git
synced 2026-01-15 12:15:26 +00:00
stream: extract Readable.from in its own file
See: https://github.com/nodejs/readable-stream/pull/420 PR-URL: https://github.com/nodejs/node/pull/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>
This commit is contained in:
@@ -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);
|
||||
};
|
||||
|
||||
46
lib/internal/streams/from.js
Normal file
46
lib/internal/streams/from.js
Normal file
@@ -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;
|
||||
Reference in New Issue
Block a user