lib: add webstreams to Duplex.from()

Refs: https://github.com/nodejs/node/pull/39519
PR-URL: https://github.com/nodejs/node/pull/46190
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Robert Nagy <ronagy@icloud.com>
This commit is contained in:
Debadree Chatterjee authored and GitHub committed 2023-01-19 20:31:54 +00:00
1 parent e7e12553fd
commit 49413ad8ae
2 files changed
+125 -15

No files matched your search

+23 -15
View File
@@ -20,6 +20,7 @@ const {
const { destroyer } = require('internal/streams/destroy');
const Duplex = require('internal/streams/duplex');
const Readable = require('internal/streams/readable');
const Writable = require('internal/streams/writable');
const { createDeferredPromise } = require('internal/util');
const from = require('internal/streams/from');
@@ -32,6 +33,16 @@ const {
FunctionPrototypeCall
} = primordials;
const {
isBrandCheck,
} = require('internal/webstreams/util');
const isReadableStream =
isBrandCheck('ReadableStream');
const isWritableStream =
isBrandCheck('WritableStream');
// This is needed for pre node 17.
class Duplexify extends Duplex {
constructor(options) {
@@ -71,15 +82,13 @@ module.exports = function duplexify(body, name) {
return _duplexify({ writable: false, readable: false });
}
// TODO: Webstreams
// if (isReadableStream(body)) {
// return _duplexify({ readable: Readable.fromWeb(body) });
// }
if (isReadableStream(body)) {
return _duplexify({ readable: Readable.fromWeb(body) });
}
// TODO: Webstreams
// if (isWritableStream(body)) {
// return _duplexify({ writable: Writable.fromWeb(body) });
// }
if (isWritableStream(body)) {
return _duplexify({ writable: Writable.fromWeb(body) });
}
if (typeof body === 'function') {
const { value, write, final, destroy } = fromAsyncGen(body);
@@ -146,13 +155,12 @@ module.exports = function duplexify(body, name) {
});
}
// TODO: Webstreams.
// if (
// isReadableStream(body?.readable) &&
// isWritableStream(body?.writable)
// ) {
// return Duplexify.fromWeb(body);
// }
if (
isReadableStream(body?.readable) &&
isWritableStream(body?.writable)
) {
return Duplexify.fromWeb(body);
}
if (
typeof body?.writable === 'object' ||
+102
View File
@@ -3,6 +3,7 @@
const common = require('../common');
const assert = require('assert');
const { Duplex, Readable, Writable, pipeline, PassThrough } = require('stream');
const { ReadableStream, WritableStream } = require('stream/web');
const { Blob } = require('buffer');
{
@@ -299,3 +300,104 @@ const { Blob } = require('buffer');
assert.strictEqual(res, 'foobar');
})).on('close', common.mustCall());
}
function makeATestReadableStream(value) {
return new ReadableStream({
start(controller) {
controller.enqueue(value);
controller.close();
}
});
}
function makeATestWritableStream(writeFunc) {
return new WritableStream({
write(chunk) {
writeFunc(chunk);
}
});
}
{
const d = Duplex.from({
readable: makeATestReadableStream('foo'),
});
assert.strictEqual(d.readable, true);
assert.strictEqual(d.writable, false);
d.on('data', common.mustCall((data) => {
assert.strictEqual(data.toString(), 'foo');
}));
d.on('end', common.mustCall(() => {
assert.strictEqual(d.readable, false);
}));
}
{
const d = Duplex.from(makeATestReadableStream('foo'));
assert.strictEqual(d.readable, true);
assert.strictEqual(d.writable, false);
d.on('data', common.mustCall((data) => {
assert.strictEqual(data.toString(), 'foo');
}));
d.on('end', common.mustCall(() => {
assert.strictEqual(d.readable, false);
}));
}
{
let ret = '';
const d = Duplex.from({
writable: makeATestWritableStream((chunk) => ret += chunk),
});
assert.strictEqual(d.readable, false);
assert.strictEqual(d.writable, true);
d.end('foo');
d.on('finish', common.mustCall(() => {
assert.strictEqual(ret, 'foo');
assert.strictEqual(d.writable, false);
}));
}
{
let ret = '';
const d = Duplex.from(makeATestWritableStream((chunk) => ret += chunk));
assert.strictEqual(d.readable, false);
assert.strictEqual(d.writable, true);
d.end('foo');
d.on('finish', common.mustCall(() => {
assert.strictEqual(ret, 'foo');
assert.strictEqual(d.writable, false);
}));
}
{
let ret = '';
const d = Duplex.from({
readable: makeATestReadableStream('foo'),
writable: makeATestWritableStream((chunk) => ret += chunk),
});
d.end('bar');
d.on('data', common.mustCall((data) => {
assert.strictEqual(data.toString(), 'foo');
}));
d.on('end', common.mustCall(() => {
assert.strictEqual(d.readable, false);
}));
d.on('finish', common.mustCall(() => {
assert.strictEqual(ret, 'bar');
assert.strictEqual(d.writable, false);
}));
}