Allow aborting an ongoing pipe operation using AbortSignals

Closes #446.

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 {{AbortSignal}} 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 !
@@ -791,6 +800,16 @@ option. If <code><a for="underlying source">type</a></code> is set to <code>unde
          1. If _preventCancel_ is *false*, <a href="#rs-pipeTo-shutdown-with-action">shutdown with an action</a> of !
             ReadableStreamCancel(*this*, _destClosed_) and with _destClosed_.
          1. Otherwise, <a href="#rs-pipeTo-shutdown">shutdown</a> with _destClosed_.
+     * <strong>Abort signals must stop activity:</strong> if _signal_ is not *undefined*, the following algorithm
+       _abortAlgorithm_ must be <a for="AbortSignal">added</a> to _signal_:
+         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 action of performing !
+            WritableStreamAbort(_dest_, _error_) to _actions_.
+         1. If _preventCancel_ is *false*, <a for="set">append</a> the action of performing !
+            ReadableStreamCancel(*this*, _error_) to _actions_.
+         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_.
      * <i id="rs-pipeTo-shutdown-with-action">Shutdown with an action</i>: if any of the above requirements ask to
        shutdown with an action _action_, optionally with an error _originalError_, then:
        1. If _shuttingDown_ is *true*, abort these substeps.
@@ -817,6 +836,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_.