lib: add webstreams to Duplex.from()

Refs: #39519
PR-URL: #46190
Backport-PR-URL: #46314
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Robert Nagy <ronagy@icloud.com>

Author: debadree25

lib/internal/streams/duplexify.js

@@ -20,6 +20,7 @@ const {
 const{ destroyer }=require('internal/streams/destroy');
 constDuplex=require('internal/streams/duplex');
 constReadable=require('internal/streams/readable');
+constWritable=require('internal/streams/writable');
 const{ createDeferredPromise }=require('internal/util');
 constfrom=require('internal/streams/from');
 
@@ -32,6 +33,16 @@ const {
  FunctionPrototypeCall,
 }=primordials;
 
+
+const{
+ isBrandCheck,
+}=require('internal/webstreams/util');
+
+constisReadableStream=
+isBrandCheck('ReadableStream');
+constisWritableStream=
+isBrandCheck('WritableStream');
+
 // This is needed for pre node 17.
 classDuplexifyextendsDuplex{
 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(typeofbody==='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)
+){
+returnDuplexify.fromWeb(body);
+}
 
 if(
 typeofbody?.writable==='object'||

test/parallel/test-stream-duplex-from.js

@@ -3,6 +3,7 @@
 constcommon=require('../common');
 constassert=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());
 }
+
+functionmakeATestReadableStream(value){
+returnnewReadableStream({
+start(controller){
+controller.enqueue(value);
+controller.close();
+}
+});
+}
+
+functionmakeATestWritableStream(writeFunc){
+returnnewWritableStream({
+write(chunk){
+writeFunc(chunk);
+}
+});
+}
+
+{
+constd=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);
+}));
+}
+
+{
+constd=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);
+}));
+}
+
+{
+letret='';
+constd=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);
+}));
+}
+
+{
+letret='';
+constd=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);
+}));
+}
+
+{
+letret='';
+constd=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);
+}));
+}