From 2c1a619cfbbab701f336eb8e95c2c0cd61ff01b1 Mon Sep 17 00:00:00 2001 From: Debadree Chatterjee Date: Fri, 20 Jan 2023 02:01:54 +0530 Subject: [PATCH] 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 Reviewed-By: James M Snell Reviewed-By: Robert Nagy --- lib/internal/streams/duplexify.js | 38 +++++---- test/parallel/test-stream-duplex-from.js | 102 +++++++++++++++++++++++ 2 files changed, 125 insertions(+), 15 deletions(-) diff --git a/lib/internal/streams/duplexify.js b/lib/internal/streams/duplexify.js index 13404e43bf1bae..24be3aa4a3a81d 100644 --- a/lib/internal/streams/duplexify.js +++ b/lib/internal/streams/duplexify.js @@ -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' || diff --git a/test/parallel/test-stream-duplex-from.js b/test/parallel/test-stream-duplex-from.js index c3f3dd756b2e66..e3c117ff8dedb0 100644 --- a/test/parallel/test-stream-duplex-from.js +++ b/test/parallel/test-stream-duplex-from.js @@ -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); + })); +}