stream: enable usage of webstreams on compose()

Refs: #39316
PR-URL: #46675
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: Robert Nagy <ronagy@icloud.com>
Reviewed-By: James M Snell <jasnell@gmail.com>

Author: debadree25

doc/api/stream.md

@@ -2796,11 +2796,16 @@ const server = http.createServer((req, res) => {
 
 <!-- YAML
 added: v16.9.0
+changes:
+ - version: REPLACEME
+ pr-url: https://github.com/nodejs/node/pull/46675
+ description: Added support for webstreams.
 -->
 
 > Stability: 1 - `stream.compose` is experimental.
 
-*`streams` {Stream\[]|Iterable\[]|AsyncIterable\[]|Function\[]}
+*`streams` {Stream\[]|Iterable\[]|AsyncIterable\[]|Function\[]|
+ ReadableStream\[]|WritableStream\[]|TransformStream\[]}
 * Returns: {stream.Duplex}
 
 Combines two or more streams into a `Duplex` stream that writes to the

lib/internal/streams/compose.js

@@ -7,6 +7,10 @@ const {
  isNodeStream,
  isReadable,
  isWritable,
+ isWebStream,
+ isTransformStream,
+ isWritableStream,
+ isReadableStream,
 }=require('internal/streams/utils');
 const{
  AbortError,
@@ -15,6 +19,7 @@ const {
 ERR_MISSING_ARGS,
 },
 }=require('internal/errors');
+consteos=require('internal/streams/end-of-stream');
 
 module.exports=functioncompose(...streams){
 if(streams.length===0){
@@ -37,18 +42,32 @@ module.exports = function compose(...streams) {
 }
 
 for(letn=0;n<streams.length;++n){
-if(!isNodeStream(streams[n])){
+if(!isNodeStream(streams[n])&&!isWebStream(streams[n])){
 // TODO(ronag): Add checks for non streams.
 continue;
 }
-if(n<streams.length-1&&!isReadable(streams[n])){
+if(
+n<streams.length-1&&
+!(
+isReadable(streams[n])||
+isReadableStream(streams[n])||
+isTransformStream(streams[n])
+)
+){
 thrownewERR_INVALID_ARG_VALUE(
 `streams[${n}]`,
 orgStreams[n],
 'must be readable',
 );
 }
-if(n>0&&!isWritable(streams[n])){
+if(
+n>0&&
+!(
+isWritable(streams[n])||
+isWritableStream(streams[n])||
+isTransformStream(streams[n])
+)
+){
 thrownewERR_INVALID_ARG_VALUE(
 `streams[${n}]`,
 orgStreams[n],
@@ -79,8 +98,16 @@ module.exports = function compose(...streams) {
 consthead=streams[0];
 consttail=pipeline(streams,onfinished);
 
-constwritable=!!isWritable(head);
-constreadable=!!isReadable(tail);
+constwritable=!!(
+isWritable(head)||
+isWritableStream(head)||
+isTransformStream(head)
+);
+constreadable=!!(
+isReadable(tail)||
+isReadableStream(tail)||
+isTransformStream(tail)
+);
 
 // TODO(ronag): Avoid double buffering.
 // Implement Writable/Readable/Duplex traits.
@@ -94,28 +121,55 @@ module.exports = function compose(...streams) {
 });
 
 if(writable){
-d._write=function(chunk,encoding,callback){
-if(head.write(chunk,encoding)){
-callback();
-}else{
-ondrain=callback;
-}
-};
-
-d._final=function(callback){
-head.end();
-onfinish=callback;
-};
+if(isNodeStream(head)){
+d._write=function(chunk,encoding,callback){
+if(head.write(chunk,encoding)){
+callback();
+}else{
+ondrain=callback;
+}
+};
+
+d._final=function(callback){
+head.end();
+onfinish=callback;
+};
+
+head.on('drain',function(){
+if(ondrain){
+constcb=ondrain;
+ondrain=null;
+cb();
+}
+});
+}elseif(isWebStream(head)){
+constwritable=isTransformStream(head) ? head.writable : head;
+constwriter=writable.getWriter();
+
+d._write=asyncfunction(chunk,encoding,callback){
+try{
+awaitwriter.ready;
+writer.write(chunk).catch(()=>{});
+callback();
+}catch(err){
+callback(err);
+}
+};
+
+d._final=asyncfunction(callback){
+try{
+awaitwriter.ready;
+writer.close().catch(()=>{});
+onfinish=callback;
+}catch(err){
+callback(err);
+}
+};
+}
 
-head.on('drain',function(){
-if(ondrain){
-constcb=ondrain;
-ondrain=null;
-cb();
-}
-});
+consttoRead=isTransformStream(tail) ? tail.readable : tail;
 
-tail.on('finish',function(){
+eos(toRead,()=>{
 if(onfinish){
 constcb=onfinish;
 onfinish=null;
@@ -125,32 +179,54 @@ module.exports = function compose(...streams) {
 }
 
 if(readable){
-tail.on('readable',function(){
-if(onreadable){
-constcb=onreadable;
-onreadable=null;
-cb();
-}
-});
-
-tail.on('end',function(){
-d.push(null);
-});
-
-d._read=function(){
-while(true){
-constbuf=tail.read();
-
-if(buf===null){
-onreadable=d._read;
-return;
+if(isNodeStream(tail)){
+tail.on('readable',function(){
+if(onreadable){
+constcb=onreadable;
+onreadable=null;
+cb();
 }
-
-if(!d.push(buf)){
-return;
+});
+
+tail.on('end',function(){
+d.push(null);
+});
+
+d._read=function(){
+while(true){
+constbuf=tail.read();
+if(buf===null){
+onreadable=d._read;
+return;
+}
+
+if(!d.push(buf)){
+return;
+}
 }
-}
-};
+};
+}elseif(isWebStream(tail)){
+constreadable=isTransformStream(tail) ? tail.readable : tail;
+constreader=readable.getReader();
+d._read=asyncfunction(){
+while(true){
+try{
+const{ value, done }=awaitreader.read();
+
+if(!d.push(value)){
+return;
+}
+
+if(done){
+d.push(null);
+return;
+}
+}catch{
+return;
+}
+}
+};
+}
 }
 
 d._destroy=function(err,callback){
@@ -166,7 +242,9 @@ module.exports = function compose(...streams) {
 callback(err);
 }else{
 onclose=callback;
-destroyer(tail,err);
+if(isNodeStream(tail)){
+destroyer(tail,err);
+}
 }
 };
 

lib/internal/streams/pipeline.js

@@ -286,7 +286,7 @@ function pipelineImpl(streams, callback, opts) {
 thrownewERR_INVALID_RETURN_VALUE(
 'Iterable, AsyncIterable or Stream','source',ret);
 }
-}elseif(isIterable(stream)||isReadableNodeStream(stream)){
+}elseif(isIterable(stream)||isReadableNodeStream(stream)||isTransformStream(stream)){
 ret=stream;
 }else{
 ret=Duplex.from(stream);
@@ -385,6 +385,7 @@ function pipelineImpl(streams, callback, opts) {
 finishCount++;
 pumpToWeb(ret,stream,finish,{ end });
 }elseif(isTransformStream(ret)){
+finishCount++;
 pumpToWeb(ret.readable,stream,finish,{ end });
 }else{
 thrownewERR_INVALID_ARG_TYPE(