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>
nodejs · Jan 31, 2023 · cde59606cd7fb436f2dc0dd06e3bbdd5431e61b9 · cde5960
1 parent 896027c
commit cde59606cd7fb436f2dc0dd06e3bbdd5431e61b9
Show file tree

Hide file tree

Showing 5 changed files with 301 additions and 9 deletions.
diff --git a/lib/internal/streams/end-of-stream.js b/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');

 function isRequest(stream) {
@@ -58,14 +61,17 @@ function eos(stream, options, callback) {

  callback = once(callback);

-  const readable = options.readable ?? isReadableNodeStream(stream);
-  const writable = options.writable ?? isWritableNodeStream(stream);
+  if (isReadableStream(stream) || isWritableStream(stream)) {
+    return eosWeb(stream, options, callback);
+  }

  if (!isNodeStream(stream)) {
-    // TODO: Webstreams.
-    throw new ERR_INVALID_ARG_TYPE('stream', 'Stream', stream);
+    throw new ERR_INVALID_ARG_TYPE('stream', ['ReadableStream', 'WritableStream', 'Stream'], stream);
  }

+  const readable = options.readable ?? isReadableNodeStream(stream);
+  const writable = options.writable ?? isWritableNodeStream(stream);
+
  const wState = stream._writableState;
  const rState = stream._readableState;

@@ -255,6 +261,15 @@ function eos(stream, options, callback) {
  return cleanup;
 }

+function eosWeb(stream, opts, callback) {
+  PromisePrototypeThen(
+    stream[kIsClosedPromise].promise,
+    () => process.nextTick(() => callback.call(stream)),
+    (err) => process.nextTick(() => callback.call(stream, err)),
+  );
+  return nop;
+}
+
 function finished(stream, opts) {
  let autoCleanup = false;
  if (opts === null) {

diff --git a/lib/internal/streams/utils.js b/lib/internal/streams/utils.js
@@ -4,13 +4,16 @@ const {
  Symbol,
  SymbolAsyncIterator,
  SymbolIterator,
+  SymbolFor,
 } = primordials;

 const kDestroyed = Symbol('kDestroyed');
 const kIsErrored = Symbol('kIsErrored');
 const kIsReadable = Symbol('kIsReadable');
 const kIsDisturbed = Symbol('kIsDisturbed');

+const kIsClosedPromise = SymbolFor('nodejs.webstream.isClosedPromise');
+
 function isReadableNodeStream(obj, strict = false) {
  return !!(
    obj &&
@@ -55,6 +58,25 @@ function isNodeStream(obj) {
  );
 }

+function isReadableStream(obj) {
+  return !!(
+    obj &&
+    !isNodeStream(obj) &&
+    typeof obj.pipeThrough === 'function' &&
+    typeof obj.getReader === 'function' &&
+    typeof obj.cancel === 'function'
+  );
+}
+
+function isWritableStream(obj) {
+  return !!(
+    obj &&
+    !isNodeStream(obj) &&
+    typeof obj.getWriter === 'function' &&
+    typeof obj.abort === 'function'
+  );
+}
+
 function isIterable(obj, isAsync) {
  if (obj == null) return false;
  if (isAsync === true) return typeof obj[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,

diff --git a/lib/internal/webstreams/readablestream.js b/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) {
 function readableStreamClose(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

diff --git a/lib/internal/webstreams/writablestream.js b/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();
+
    const size = extractSizeAlgorithm(strategy?.size);
    const highWaterMark = 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);
 }