stream: implement finished() for ReadableStream and WritableStream

Refs: #39316
PR-URL: #46205
Reviewed-By: Robert Nagy <ronagy@icloud.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Darshan Sen <raisinten@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>

Author: debadree25

lib/internal/streams/end-of-stream.js

@@ -22,20 +22,23 @@ const {
  validateBoolean
 }=require('internal/validators');
 
-const{ Promise }=primordials;
+const{ Promise, PromisePrototypeThen}=primordials;
 
 const{
  isClosed,
  isReadable,
  isReadableNodeStream,
+ isReadableStream,
  isReadableFinished,
  isReadableErrored,
  isWritable,
  isWritableNodeStream,
+ isWritableStream,
  isWritableFinished,
  isWritableErrored,
  isNodeStream,
 willEmitClose: _willEmitClose,
+ kIsClosedPromise,
 }=require('internal/streams/utils');
 
 functionisRequest(stream){
@@ -58,14 +61,17 @@ function eos(stream, options, callback) {
 
 callback=once(callback);
 
-constreadable=options.readable??isReadableNodeStream(stream);
-constwritable=options.writable??isWritableNodeStream(stream);
+if(isReadableStream(stream)||isWritableStream(stream)){
+returneosWeb(stream,options,callback);
+}
 
 if(!isNodeStream(stream)){
-// TODO: Webstreams.
-thrownewERR_INVALID_ARG_TYPE('stream','Stream',stream);
+thrownewERR_INVALID_ARG_TYPE('stream',['ReadableStream','WritableStream','Stream'],stream);
 }
 
+constreadable=options.readable??isReadableNodeStream(stream);
+constwritable=options.writable??isWritableNodeStream(stream);
+
 constwState=stream._writableState;
 constrState=stream._readableState;
 
@@ -255,6 +261,15 @@ function eos(stream, options, callback) {
 returncleanup;
 }
 
+functioneosWeb(stream,opts,callback){
+PromisePrototypeThen(
+stream[kIsClosedPromise].promise,
+()=>process.nextTick(()=>callback.call(stream)),
+(err)=>process.nextTick(()=>callback.call(stream,err)),
+);
+returnnop;
+}
+
 functionfinished(stream,opts){
 letautoCleanup=false;
 if(opts===null){

lib/internal/streams/utils.js

@@ -4,13 +4,16 @@ const {
  Symbol,
  SymbolAsyncIterator,
  SymbolIterator,
+ SymbolFor,
 }=primordials;
 
 constkDestroyed=Symbol('kDestroyed');
 constkIsErrored=Symbol('kIsErrored');
 constkIsReadable=Symbol('kIsReadable');
 constkIsDisturbed=Symbol('kIsDisturbed');
 
+constkIsClosedPromise=SymbolFor('nodejs.webstream.isClosedPromise');
+
 functionisReadableNodeStream(obj,strict=false){
 return!!(
 obj&&
@@ -55,6 +58,25 @@ function isNodeStream(obj) {
 );
 }
 
+functionisReadableStream(obj){
+return!!(
+obj&&
+!isNodeStream(obj)&&
+typeofobj.pipeThrough==='function'&&
+typeofobj.getReader==='function'&&
+typeofobj.cancel==='function'
+);
+}
+
+functionisWritableStream(obj){
+return!!(
+obj&&
+!isNodeStream(obj)&&
+typeofobj.getWriter==='function'&&
+typeofobj.abort==='function'
+);
+}
+
 functionisIterable(obj,isAsync){
 if(obj==null)returnfalse;
 if(isAsync===true)returntypeofobj[SymbolAsyncIterator]==='function';
@@ -269,18 +291,21 @@ module.exports = {
  kIsErrored,
  isReadable,
  kIsReadable,
+ kIsClosedPromise,
  isClosed,
  isDestroyed,
  isDuplexNodeStream,
  isFinished,
  isIterable,
  isReadableNodeStream,
+ isReadableStream,
  isReadableEnded,
  isReadableFinished,
  isReadableErrored,
  isNodeStream,
  isWritable,
  isWritableNodeStream,
+ isWritableStream,
  isWritableEnded,
  isWritableFinished,
  isWritableErrored,

lib/internal/webstreams/readablestream.js

@@ -85,6 +85,7 @@ const {
  kIsDisturbed,
  kIsErrored,
  kIsReadable,
+ kIsClosedPromise,
 }=require('internal/streams/utils');
 
 const{
@@ -258,9 +259,11 @@ class ReadableStream {
 port1: undefined,
 port2: undefined,
 promise: undefined,
-}
+},
 };
 
+this[kIsClosedPromise]=createDeferredPromise();
+
 // The spec requires handling of the strategy first
 // here. Specifically, if getting the size and
 // highWaterMark from the strategy fail, that has
@@ -652,8 +655,9 @@ function TransferredReadableStream() {
 writable: undefined,
 port: undefined,
 promise: undefined,
-}
+},
 };
+this[kIsClosedPromise]=createDeferredPromise();
 },
 [],ReadableStream));
 }
@@ -1213,8 +1217,9 @@ function createTeeReadableStream(start, pull, cancel) {
 writable: undefined,
 port: undefined,
 promise: undefined,
-}
+},
 };
+this[kIsClosedPromise]=createDeferredPromise();
 setupReadableStreamDefaultControllerFromSource(
 this,
 ObjectCreate(null,{
@@ -1887,6 +1892,7 @@ function readableStreamCancel(stream, reason) {
 functionreadableStreamClose(stream){
 assert(stream[kState].state==='readable');
 stream[kState].state='closed';
+stream[kIsClosedPromise].resolve();
 
 const{
  reader,
@@ -1908,6 +1914,8 @@ function readableStreamError(stream, error) {
 assert(stream[kState].state==='readable');
 stream[kState].state='errored';
 stream[kState].storedError=error;
+stream[kIsClosedPromise].reject(error);
+setPromiseHandled(stream[kIsClosedPromise].promise);
 
 const{
  reader

lib/internal/webstreams/writablestream.js

@@ -69,6 +69,10 @@ const {
  kState,
 }=require('internal/webstreams/util');
 
+const{
+ kIsClosedPromise,
+}=require('internal/streams/utils');
+
 const{
  AbortController,
 }=require('internal/abort_controller');
@@ -191,9 +195,11 @@ class WritableStream {
 port1: undefined,
 port2: undefined,
 promise: undefined,
-}
+},
 };
 
+this[kIsClosedPromise]=createDeferredPromise();
+
 constsize=extractSizeAlgorithm(strategy?.size);
 consthighWaterMark=extractHighWaterMark(strategy?.highWaterMark,1);
 
@@ -363,6 +369,7 @@ function TransferredWritableStream() {
 readable: undefined,
 },
 };
+this[kIsClosedPromise]=createDeferredPromise();
 },
 [],WritableStream));
 }
@@ -742,6 +749,10 @@ function writableStreamRejectCloseAndClosedPromiseIfNeeded(stream) {
 resolve: undefined,
 };
 }
+
+stream[kIsClosedPromise].reject(stream[kState]?.storedError);
+setPromiseHandled(stream[kIsClosedPromise].promise);
+
 const{
  writer,
 }=stream[kState];
@@ -855,6 +866,7 @@ function writableStreamFinishInFlightClose(stream) {
 stream[kState].state='closed';
 if(stream[kState].writer!==undefined)
 stream[kState].writer[kState].close.resolve?.();
+stream[kIsClosedPromise].resolve?.();
 assert(stream[kState].pendingAbortRequest.abort.promise===undefined);
 assert(stream[kState].storedError===undefined);
 }