Allow aborting an ongoing pipe operation using AbortSignals

Closes #446.

Co-authored-by: Adam Rice <[email protected]>

Author: domenic

index.bs

@@ -401,7 +401,7 @@ like
     <a href="#rs-cancel">cancel</a>(reason)
     <a href="#rs-get-reader">getReader</a>()
     <a href="#rs-pipe-through">pipeThrough</a>({ writable, readable }, options)
-    <a href="#rs-pipe-to">pipeTo</a>(dest, { preventClose, preventAbort, preventCancel } = {})
+    <a href="#rs-pipe-to">pipeTo</a>(dest, { preventClose, preventAbort, preventCancel, signal } = {})
     <a href="#rs-tee">tee</a>()
   }
 </code></pre>
@@ -701,8 +701,9 @@ option. If <code><a for="underlying source">type</a></code> is set to <code>unde
   </code></pre>
 </div>
 
-<h5 id="rs-pipe-to" method for="ReadableStream" lt="pipeTo(dest, options)">pipeTo(<var ignore>dest</var>, {
-<var ignore>preventClose</var>, <var ignore>preventAbort</var>, <var ignore>preventCancel</var> } = {})</h5>
+<h5 id="rs-pipe-to" method for="ReadableStream" lt="pipeTo(dest, options)">pipeTo(<var ignore>dest</var>,
+{ <var ignore>preventClose</var>, <var ignore>preventAbort</var>, <var ignore>preventCancel</var>,
+<var ignore>signal</var> } = {})</h5>
 
 <div class="note">
   The <code>pipeTo</code> method <a lt="piping">pipes</a> this <a>readable stream</a> to a given <a>writable
@@ -733,13 +734,21 @@ option. If <code><a for="underlying source">type</a></code> is set to <code>unde
     promise will be rejected with an error indicating piping to a closed stream failed, or with any error that occurs
     during canceling the source.</p></li>
   </ul>
+
+  The <code>signal</code> option can be set to an {{AbortSignal}} to allow aborting an ongoing pipe operation via the
+  corresponding {{AbortController}}. In this case, the source <a>readable stream</a> will be <a lt="cancel a readable
+  stream">canceled</a>, and the destination <a>writable stream</a> <a lt="abort a writable stream">aborted</a>, unless
+  the respective options <code>preventCancel</code> or <code>preventAbort</code> are set.
+
 </div>
 
 <emu-alg>
   1. If ! IsReadableStream(*this*) is *false*, return <a>a promise rejected with</a> a *TypeError* exception.
   1. If ! IsWritableStream(_dest_) is *false*, return <a>a promise rejected with</a> a *TypeError* exception.
   1. Set _preventClose_ to ! ToBoolean(_preventClose_), set _preventAbort_ to ! ToBoolean(_preventAbort_), and set
      _preventCancel_ to ! ToBoolean(_preventCancel_).
+  1. If _signal_ is not *undefined*, and _signal_ is not an instance of the `<a idl>AbortSignal</a>` interface, return
+     <a>a promise rejected with</a> a *TypeError* exception.
   1. If ! IsReadableStreamLocked(*this*) is *true*, return <a>a promise rejected with</a> a *TypeError* exception.
   1. If ! IsWritableStreamLocked(_dest_) is *true*, return <a>a promise rejected with</a> a *TypeError* exception.
   1. If ! IsReadableByteStreamController(*this*.[[readableStreamController]]) is *true*, let _reader_ be either !
@@ -749,6 +758,20 @@ option. If <code><a for="underlying source">type</a></code> is set to <code>unde
   1. Let _writer_ be ! AcquireWritableStreamDefaultWriter(_dest_).
   1. Let _shuttingDown_ be *false*.
   1. Let _promise_ be <a>a new promise</a>.
+  1. If _signal_ is not *undefined*,
+    1. Let _abortAlgorithm_ be the following steps:
+      1. Let _error_ be a new "`<a idl>AbortError</a>`" `<a idl>DOMException</a>`.
+      1. Let _actions_ be an empty <a>ordered set</a>.
+      1. If _preventAbort_ is *false*, <a for="set">append</a> the following action to _actions_:
+        1. If _dest_.[[_state_]] is `"writable"`, return ! WritableStreamAbort(_dest_, _error_).
+        1. Otherwise, return <a>a promise resolved with</a> *undefined*.
+      1. If _preventCancel_ is *false*, <a for="set">append</a> the following action action to _actions_:
+        1. If *this*.[[_state_]] is `"readable"`, return ! ReadableStreamCancel(*this*, _error_).
+        1. Otherwise, return <a>a promise resolved with</a> *undefined*.
+      1. <a href="#rs-pipeTo-shutdown-with-action">Shutdown with an action</a> consisting of <a>waiting for all</a>
+         of the actions in _actions_, and with _error_.
+    1. If _signal_'s <a for=AbortSignal>aborted flag</a> is set, perform _abortAlgorithm_ and return _promise_.
+    1. <a for="AbortSignal">Add</a> _abortAlgorithm_ to _signal_.
   1. <a>In parallel</a> <span class="XXX">but not really; see <a
      href="https://github.com/whatwg/streams/issues/905">#905</a></span>, using _reader_ and _writer_, read all
      <a>chunks</a> from *this* and write them to _dest_. Due to the locking provided by the reader and writer, the exact
@@ -817,6 +840,7 @@ option. If <code><a for="underlying source">type</a></code> is set to <code>unde
        an error _error_, which means to perform the following steps:
        1. Perform ! WritableStreamDefaultWriterRelease(_writer_).
        1. Perform ! ReadableStreamReaderGenericRelease(_reader_).
+       1. If _signal_ is not *undefined*, <a for="AbortSignal">remove</a> _abortAlgorithm_ from _signal_.
        1. If _error_ was given, <a>reject</a> _promise_ with _error_.
        1. Otherwise, <a>resolve</a> _promise_ with *undefined*.
   1. Return _promise_.

reference-implementation/lib/helpers.js

@@ -156,3 +156,67 @@ exports.MakeSizeAlgorithmFromSizeFunction = size => {
   }
   return chunk => size(chunk);
 };
+
+exports.PerformPromiseThen = (promise, onFulfilled, onRejected) => {
+  // There doesn't appear to be any way to correctly emulate the behaviour from JavaScript, so this is just an
+  // approximation.
+  return Promise.prototype.then.call(promise, onFulfilled, onRejected);
+};
+
+exports.WaitForAll = (promises, successSteps, failureSteps) => {
+  let rejected = false;
+  const rejectionHandler = arg => {
+    if (rejected === false) {
+      rejected = true;
+      failureSteps(arg);
+    }
+  };
+  let index = 0;
+  let fulfilledCount = 0;
+  const total = promises.length;
+  const result = new Array(total);
+  for (const promise of promises) {
+    const promiseIndex = index;
+    const fulfillmentHandler = arg => {
+      result[promiseIndex] = arg;
+      ++fulfilledCount;
+      if (fulfilledCount === total) {
+        successSteps(result);
+      }
+    };
+    exports.PerformPromiseThen(promise, fulfillmentHandler, rejectionHandler);
+    ++index;
+  }
+};
+
+exports.WaitForAllPromise = (promises, successSteps, failureSteps = undefined) => {
+  let resolvePromise;
+  let rejectPromise;
+  const promise = new Promise((resolve, reject) => {
+    resolvePromise = resolve;
+    rejectPromise = reject;
+  });
+  if (failureSteps === undefined) {
+    failureSteps = arg => {
+      throw arg;
+    };
+  }
+  const successStepsWrapper = results => {
+    try {
+      const stepsResult = successSteps(results);
+      resolvePromise(stepsResult);
+    } catch (e) {
+      rejectPromise(e);
+    }
+  };
+  const failureStepsWrapper = reason => {
+    try {
+      const stepsResult = failureSteps(reason);
+      resolvePromise(stepsResult);
+    } catch (e) {
+      rejectPromise(e);
+    }
+  };
+  exports.WaitForAll(promises, successStepsWrapper, failureStepsWrapper);
+  return promise;
+};

reference-implementation/lib/readable-stream.js

@@ -1,8 +1,11 @@
 'use strict';
+/* global AbortSignal:false */
+
 const assert = require('better-assert');
 const { ArrayBufferCopy, CreateAlgorithmFromUnderlyingMethod, IsFiniteNonNegativeNumber, InvokeOrNoop,
         IsDetachedBuffer, TransferArrayBuffer, ValidateAndNormalizeHighWaterMark, IsNonNegativeNumber,
-        MakeSizeAlgorithmFromSizeFunction, createArrayFromList, typeIsObject } = require('./helpers.js');
+        MakeSizeAlgorithmFromSizeFunction, createArrayFromList, typeIsObject, WaitForAllPromise } =
+      require('./helpers.js');
 const { rethrowAssertionErrorRejection } = require('./utils.js');
 const { DequeueValue, EnqueueValueWithSize, ResetQueue } = require('./queue-with-sizes.js');
 const { AcquireWritableStreamDefaultWriter, IsWritableStream, IsWritableStreamLocked,
@@ -97,7 +100,7 @@ class ReadableStream {
     return readable;
   }
 
-  pipeTo(dest, { preventClose, preventAbort, preventCancel } = {}) {
+  pipeTo(dest, { preventClose, preventAbort, preventCancel, signal } = {}) {
     if (IsReadableStream(this) === false) {
       return Promise.reject(streamBrandCheckException('pipeTo'));
     }
@@ -110,6 +113,10 @@ class ReadableStream {
     preventAbort = Boolean(preventAbort);
     preventCancel = Boolean(preventCancel);
 
+    if (signal !== undefined && !isAbortSignal(signal)) {
+      return Promise.reject(new TypeError('ReadableStream.prototype.pipeTo\'s signal option must be an AbortSignal'));
+    }
+
     if (IsReadableStreamLocked(this) === true) {
       return Promise.reject(new TypeError('ReadableStream.prototype.pipeTo cannot be used on a locked ReadableStream'));
     }
@@ -126,6 +133,38 @@ class ReadableStream {
     let currentWrite = Promise.resolve();
 
     return new Promise((resolve, reject) => {
+      let abortAlgorithm;
+      if (signal !== undefined) {
+        abortAlgorithm = () => {
+          const error = new DOMException('Aborted', 'AbortError');
+          const actions = [];
+          if (preventAbort === false) {
+            actions.push(() => {
+              if (dest._state === 'writable') {
+                return WritableStreamAbort(dest, error);
+              }
+              return Promise.resolve();
+            });
+          }
+          if (preventCancel === false) {
+            actions.push(() => {
+              if (this._state === 'readable') {
+                return ReadableStreamCancel(this, error);
+              }
+              return Promise.resolve();
+            });
+          }
+          shutdownWithAction(() => WaitForAllPromise(actions.map(action => action()), results => results), true, error);
+        };
+
+        if (signal.aborted === true) {
+          abortAlgorithm();
+          return;
+        }
+
+        signal.addEventListener('abort', abortAlgorithm);
+      }
+
       // Using reader and writer, read all chunks from this and write them to dest
       // - Backpressure must be enforced
       // - Shutdown must stop all activity
@@ -250,6 +289,9 @@ class ReadableStream {
         WritableStreamDefaultWriterRelease(writer);
         ReadableStreamReaderGenericRelease(reader);
 
+        if (signal !== undefined) {
+          signal.removeEventListener('abort', abortAlgorithm);
+        }
         if (isError) {
           reject(error);
         } else {
@@ -1952,6 +1994,21 @@ function SetUpReadableStreamBYOBRequest(request, controller, view) {
 
 // Helper functions for the ReadableStream.
 
+function isAbortSignal(value) {
+  if (typeof value !== 'object' || value === null) {
+    return false;
+  }
+
+  // Use the brand check to distinguish a real AbortSignal from a fake one.
+  const aborted = Object.getOwnPropertyDescriptor(AbortSignal.prototype, 'aborted').get;
+  try {
+    aborted.call(value);
+    return true;
+  } catch (e) {
+    return false;
+  }
+}
+
 function streamBrandCheckException(name) {
   return new TypeError(`ReadableStream.prototype.${name} can only be used on a ReadableStream`);
 }

reference-implementation/package.json

@@ -18,7 +18,7 @@
     "minimatch": "^3.0.4",
     "nyc": "^13.0.1",
     "opener": "^1.5.1",
-    "wpt-runner": "^2.2.0"
+    "wpt-runner": "^2.6.0"
   },
   "nyc": {
     "include": [