add pipeline, addAbortSignal, and finished for webstreams with passin…

…g tests
nodejs · ghost · Jul 24, 2021 · Jul 25, 2021 · Jul 25, 2021 · 2e232114b858e5f4ac6b8fd8d03fb56458e4a143
commit 2e232114b858e5f4ac6b8fd8d03fb56458e4a143
diff --git a/doc/api/webstreams.md b/doc/api/webstreams.md
@@ -33,6 +33,9 @@ There are three primary types of objects
 * `WritableStream` - Represents a destination for streaming data.
 * `TransformStream` - Represents an algorithm for transforming streaming data.
 
+Additionally, this module includes the utility functions
+[`pipeline()`](#pipeline), [`finished()`](#finished), and [`addAbortSignal()`](#addabortsignal).
+
 ### Example `ReadableStream`
 
 This example creates a simple `ReadableStream` that pushes the current

diff --git a/lib/internal/webstreams/add-abort-signal.js b/lib/internal/webstreams/add-abort-signal.js
@@ -0,0 +1,55 @@
+'use strict';
+
+const { validateAbortSignal } = require("internal/validators");
+const { isWebstream } = require('internal/webstreams/util');
+const { finished } = require('internal/webstreams/finished');
+const { AbortError, codes } = require("internal/errors");
+const { ERR_INVALID_ARG_TYPE } = codes;
+
+function addAbortSignal(signal, stream) {
+    validateInputs(signal, stream);
+
+    const onAbort = () => {
+        stream.abort(new AbortError());
+    }
+
+    if (signalIsAborted(signal)){
+        onAbort();
+    } else {
+        addAbortCallbackToSignal(signal, stream, onAbort);
+    }
+
+    return stream;
+}
+
+function validateInputs(signal, stream) {
+    validateAbortSignal(signal, 'signal');
+    throwErrorIfNotWebstream(stream);
+}
+
+function throwErrorIfNotWebstream(stream) {
+    if (isWebstream(stream)) {
+      return;
+    }
+
+    throw new ERR_INVALID_ARG_TYPE(
+      'stream',
+      'ReadableStream|WritableStream|TransformStream',
+      stream,
+    );
+}
+
+function signalIsAborted(signal) {
+    return signal.aborted;
+}
+
+function addAbortCallbackToSignal(signal, stream, callback) {
+    signal.addEventListener('abort', callback);
+    finished(stream, () => {
+        signal.removeEventListener('abort', callback); 
+    });
+}
+
+module.exports = {
+    addAbortSignal
+}
diff --git a/lib/internal/webstreams/finished.js b/lib/internal/webstreams/finished.js
@@ -0,0 +1,118 @@
+'use strict';
+
+const { validateFunction } = require('internal/validators');
+const { isWebstream } = require('internal/webstreams/util');
+const { isTransformStream } = require('internal/webstreams/transformstream');
+const {
+  isWritableStreamClosed,
+  isWritableStreamErrored,
+  writableStreamAddFinishedCallback,
+  writableStreamRemoveAllFinishedCallbacks,
+} = require('internal/webstreams/writablestream');
+const {
+  isReadableStream,
+  isReadableStreamClosed,
+  isReadableStreamErrored,
+  readableStreamAddFinishedCallback,
+  readableStreamRemoveAllFinishedCallbacks,
+} = require('internal/webstreams/readablestream');
+const { once } = require('internal/util');
+const { codes } = require('internal/errors');
+const { ERR_INVALID_ARG_TYPE, ERR_MISSING_ARGS } = codes;
+
+function finished(stream, callback) {
+  validateAndFixInputs(stream, callback);
+
+  if (isTransformStream(stream)) {
+    return finishedForTransformStream(stream, callback);
+  }
+
+  addCallbackToStream(stream, callback);
+  callCallbackIfStreamIsFinished(stream, callback);
+
+  return getCleanupFuncForStream(stream);
+}
+
+function validateAndFixInputs(stream, callback) {
+  throwErrorIfNullStream(stream);
+  callback = replaceWithNoOpIfNull(callback);
+  validateInputs(stream, callback);
+  return stream, once(callback);
+}
+
+function throwErrorIfNullStream(stream) {
+  if (!stream) {
+    throw new ERR_MISSING_ARGS('stream');
+  }
+}
+
+function replaceWithNoOpIfNull(callback) {
+  if (!callback) {
+    callback = () => {};
+  }
+  return callback;
+}
+
+function validateInputs(stream, callback) {
+  validateFunction(callback, 'callback');
+  throwErrorIfNotWebstream(stream);
+}
+
+function throwErrorIfNotWebstream(stream) {
+  if (isWebstream(stream)) {
+    return;
+  }
+
+  throw new ERR_INVALID_ARG_TYPE(
+    'stream',
+    'ReadableStream|WritableStream|TransformStream',
-    'ReadableStream|WritableStream|TransformStream',
+    ['ReadableStream', 'WritableStream', 'TransformStream'],
-    'ReadableStream|WritableStream|TransformStream',
+    ['ReadableStream', 'WritableStream', 'TransformStream'],
+    stream
+  );
+}
+
+function finishedForTransformStream(transformStream, callback) {
+  return finished(transformStream.readable, callback);
+}
+
+function addCallbackToStream(stream, callback) {
+  if (isReadableStream(stream)) {
+    readableStreamAddFinishedCallback(stream, callback);
+    return;
+  }
+  writableStreamAddFinishedCallback(stream, callback);
+}
+
+function callCallbackIfStreamIsFinished(stream, callback) {
+  if (isWebstreamFinished(stream)) {
+    callback();
+  }
+}
+
+function isWebstreamFinished(stream) {
+  if (isReadableStream(stream)) {
+    return isReadableStreamFinished(stream);
+  }
+  return isWritableStreamFinished(stream);
+}
+
+function isReadableStreamFinished(rs) {
+  return isReadableStreamClosed(rs) || isReadableStreamErrored(rs);
+}
+
+function isWritableStreamFinished(ws) {
+  return isWritableStreamClosed(ws) || isWritableStreamErrored(ws);
+}
+
+function getCleanupFuncForStream(stream) {
+  if (isReadableStream(stream)) {
+    return () => {
+      readableStreamRemoveAllFinishedCallbacks(stream);
+    };
+  }
+
+  return () => {
+    writableStreamRemoveAllFinishedCallbacks(stream);
+  };
+}
+
+module.exports = { finished };
diff --git a/lib/internal/webstreams/pipeline.js b/lib/internal/webstreams/pipeline.js
@@ -0,0 +1,132 @@
+'use strict';
+
+const { ArrayIsArray } = primordials;
+
+const { validateCallback } = require('internal/validators');
+const { isTransformStream } = require('internal/webstreams/transformstream');
+const { isReadableStream } = require('internal/webstreams/readablestream');
+const { isWritableStream } = require('internal/webstreams/writablestream');
+const { once } = require('internal/util');
+const { codes } = require('internal/errors');
+const { ERR_INVALID_ARG_TYPE, ERR_MISSING_ARGS } = codes;
+
+function pipeline(...input) {
+  const [streams, callback] = parseAndValidateInput(input);
+  pipeStreams(streams).then(callback);
+}
+
+function parseAndValidateInput(input) {
+  validateInput(input);
+  const [streams, callback] = parseInput(input);
+
+  validateStreams(streams);
+  validateCallback(callback);
+  return [streams, once(callback)];
+}
+
+function validateInput(input) {
+  if (!input || input.length < 2) {
+    throw new ERR_MISSING_ARGS('input');
+  }
+}
+
+function parseInput(input) {
+  const callback = popCallbackFromInput(input);
+  const streams = unnestArrayIfIsArray(input);
+  return [streams, callback];
+}
+
+function popCallbackFromInput(input) {
+  return input.pop();
+}
+
+function unnestArrayIfIsArray(array) {
+  if (ArrayIsArray(array[0]) && array.length === 1) {
+    return array[0];
+  }
+  return array;
+}
+
+function validateStreams(streams) {
+  throwErrorIfInsufficientStreams(streams);
+
+  const allButFirstStream = streams.slice(1);
+  throwErrorIfStreamsNotWritable(allButFirstStream);
+
+  const allButLastStream = streams.slice(0, -1);
+  throwErrorIfStreamsNotReadable(allButLastStream);
+}
+
+function throwErrorIfInsufficientStreams(streams) {
+  if (streams.length < 2) {
+    throw new ERR_MISSING_ARGS('streams');
+  }
+}
+
+function throwErrorIfStreamsNotWritable(streams) {
+  if (streams.some(cannotWriteToStream)) {
+    throw new ERR_INVALID_ARG_TYPE(
+      'streams',
+      'WritableStream|TransformStream',
-      'WritableStream|TransformStream',
+      ['WritableStream', 'TransformStream'],
-      'WritableStream|TransformStream',
+      ['WritableStream', 'TransformStream'],
+      streams
+    );
+  }
+}
+
+function cannotWriteToStream(stream) {
+  return !(isWritableStream(stream) || isTransformStream(stream));
+}
+
+function throwErrorIfStreamsNotReadable(streams) {
+  if (streams.some(cannotReadFromStream)) {
+    throw new ERR_INVALID_ARG_TYPE(
+      'streams',
+      'ReadableStream|TransformStream',
-      'ReadableStream|TransformStream',
+      ['ReadableStream', 'TransformStream'],
-      'ReadableStream|TransformStream',
+      ['ReadableStream', 'TransformStream'],
+      streams
+    );
+  }
+}
+
+function cannotReadFromStream(stream) {
+  return !(isReadableStream(stream) || isTransformStream(stream));
+}
+
+async function pipeStreams(streams) {
+  for (let i = 0; i < streams.length - 1; i++) {
+    const source = streams[i];
+    const dest = streams[i + 1];
+    await pipeStreamPair(source, dest);
+  }
+}
+
+async function pipeStreamPair(source, dest) {
+  if (isTransformStream(source)) {
+    return pipeStreamPairWithTransformSource(source, dest);
+  }
+
+  if (isTransformStream(dest)) {
+    return source.pipeThrough(dest);
+  }
+
+  return source.pipeTo(dest);
+}
+
+async function pipeStreamPairWithTransformSource(source, dest) {
+  throwErrorIfTransformStreamHasNoReadable(source);
+  return pipeStreamPair(source.readable, dest);
+}
+
+function throwErrorIfTransformStreamHasNoReadable(transform) {
+  const readable = transform?.readable;
+  if (!isReadableStream(readable)) {
+    throw new ERR_INVALID_ARG_TYPE(
+      'transform.readable',
+      'ReadableStream',
+      readable
+    );
+  }
+}
+
+module.exports = {
+  pipeline,
+};