Implementation of ReadableStream pipeTo and pipeThrough

third_party/WebKit/Source/core/streams/ReadableStream.js

 // Copyright 2015 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 (function(global, binding, v8) {
   'use strict';
 
   const _reader = v8.createPrivateSymbol('[[reader]]');
   const _storedError = v8.createPrivateSymbol('[[storedError]]');
   const _controller = v8.createPrivateSymbol('[[controller]]');
@@ skipping 31 matching lines @@
   const PULLING = 0b100;
   const PULL_AGAIN = 0b1000;
   const EXTERNALLY_CONTROLLED = 0b10000;
 
   const undefined = global.undefined;
   const Infinity = global.Infinity;
 
   const defineProperty = global.Object.defineProperty;
   const hasOwnProperty = v8.uncurryThis(global.Object.hasOwnProperty);
   const callFunction = v8.uncurryThis(global.Function.prototype.call);
+  const applyFunction = v8.uncurryThis(global.Function.prototype.apply);
 
   const TypeError = global.TypeError;
   const RangeError = global.RangeError;
 
   const Number = global.Number;
   const Number_isNaN = Number.isNaN;
   const Number_isFinite = Number.isFinite;
 
   const Promise = global.Promise;
   const thenPromise = v8.uncurryThis(Promise.prototype.then);
@@ skipping 24 matching lines @@
       'ReadableStreamReader constructor argument is not a readable stream';
   const errReaderConstructorStreamAlreadyLocked =
       'ReadableStreamReader constructor can only accept readable streams that are not yet locked to a reader';
   const errReleaseReaderWithPendingRead =
       'Cannot release a readable stream reader when it still has outstanding read() calls that have not yet settled';
   const errReleasedReaderClosedPromise =
       'This readable stream reader has been released and cannot be used to monitor the stream\'s state';
 
   const errTmplMustBeFunctionOrUndefined = name =>
       `${name} must be a function or undefined`;
+  const errCannotPipeLockedStream = 'Cannot pipe a locked stream';
+  const errCannotPipeToALockedStream = 'Cannot pipe to a locked stream';
+  const errDestinationStreamClosed = 'Destination stream closed';
 
   class ReadableStream {
     constructor() {
       // TODO(domenic): when V8 gets default parameters and destructuring, all
       // this can be cleaned up.
       const underlyingSource = arguments[0] === undefined ? {} : arguments[0];
       const strategy = arguments[1] === undefined ? {} : arguments[1];
       const size = strategy.size;
       let highWaterMark = strategy.highWaterMark;
       if (highWaterMark === undefined) {
@@ skipping 68 matching lines @@
 
     tee() {
       if (IsReadableStream(this) === false) {
         throw new TypeError(streamErrors.illegalInvocation);
       }
 
       return ReadableStreamTee(this);
     }
   }
 
+  // TODO(ricea): Move this into the class definition once it ships.
+  function ReadableStream_prototype_pipeThrough({writable, readable}, options) {
+    this.pipeTo(writable, options);
+    return readable;
+  }
+
+  // TODO(ricea): Move this into the class definition once it ships.
+  function ReadableStream_prototype_pipeTo(
+      dest, { preventClose, preventAbort, preventCancel } = {}) {
+    if (!IsReadableStream(this)) {
+      return Promise_reject(new TypeError(streamErrors.illegalInvocation));
+    }
+
+    if (!binding.IsWritableStream(dest)) {
+      // TODO(ricea): Think about having a better error message.
+      return Promise_reject(new TypeError(streamErrors.illegalInvocation));
+    }
+
+    preventClose = Boolean(preventClose);
+    preventAbort = Boolean(preventAbort);
+    preventCancel = Boolean(preventCancel);
+
+    const readable = this;
+    if (IsReadableStreamLocked(readable)) {
+      return Promise_reject(new TypeError(errCannotPipeLockedStream));
+    }
+
+    if (binding.IsWritableStreamLocked(dest)) {
+      return Promise_reject(new TypeError(errCannotPipeToALockedStream));
+    }
+
+    const reader = AcquireReadableStreamDefaultReader(readable);
+    const writer = binding.AcquireWritableStreamDefaultWriter(dest);
+    let shuttingDown = false;
+    const promise = v8.createPromise();
+    let reading = false;
+
+    if (initialStateOk()) {
+      // Need to detect closing and error when we are not reading.
+      thenPromise(reader[_closedPromise], readableClosed, readableError);
+      // Need to detect error when we are not writing.
+      thenPromise(binding.WritableStreamDefaultWriterClosedPromise(writer),
+                  undefined, writableError);
+      pump();
+    }
+
+    function initialStateOk() {

[domenic, 2017/01/18 23:51:17]

It might make sense to pull some of these out to top-level in the file (passing
them some arguments) so that they are not re-created every time someone calls
pipeTo. But it's a judgement call as to which is more appropriate, and can
certainly be left for a follow-up.

[Adam Rice, 2017/01/19 14:02:47]

This function calls state-changing functions, which means it would have to be
passed an object encapsulating the current state, which ends up having to bind a
lot of parameters.

I started out implementing pipeTo as a class, but I found I had to bind |this|
to the methods and it was very ugly. So I took the painful step of rewriting it
with nested functions. Having done that I don't really want to go back to
binding stuff again.

This whole thing will change drastically when we start optimising pipeTo anyway,
so I'm not worried.

+      const state = ReadableStreamGetState(readable);
+      if (state === STATE_ERRORED) {
+        readableError();
+        return false;
+      }
+
+      if (binding.IsWritableStreamErrored(dest)) {

[domenic, 2017/01/18 23:51:17]

Maybe since these are not abstract operation names they should go with the more
conventional camel case? isWritableStreamErrored?

[Adam Rice, 2017/01/19 14:02:47]

It's not great for the getters, because I don't want to write WritableStream
with a lowercase W. I added a "get" prefix instead.

+        writableError();
+        return false;
+      }
+
+      if (state === STATE_CLOSED) {
+        readableClosed();
+        return false;
+      }
+
+      if (binding.IsWritableStreamClosingOrClosed(dest)) {
+        writableStartedClosed();
+        return false;
+      }
+
+      return true;
+    }
+
+    function pump() {
+      if (shuttingDown) {
+        return;
+      }
+      const desiredSize =
+          binding.WritableStreamDefaultWriterGetDesiredSize(writer);
+      if (desiredSize === null) {
+        writableError();
+      }
+      if (desiredSize <= 0) {
+        thenPromise(binding.WritableStreamDefaultWriterReadyPromise(writer),
+                    pump, writableError);
+        return;
+      }
+      reading = true;
+      // TODO(ricea): Delay reads heuristically when desiredSize is low.
+      thenPromise(ReadableStreamDefaultReaderRead(reader), readResolved,
+                  readRejected);
+    }
+
+    function readResolved({value, done}) {

[domenic, 2017/01/18 23:51:17]

Nit: "fulfilled" is more correct than "resolved" here.
https://github.com/domenic/promises-unwrapping/blob/master/docs/states-and-fa...

[Adam Rice, 2017/01/19 14:02:47]

I hadn't seen that document before. It's cleared up some stuff that was fuzzy in
my mind. Thanks!

+      reading = false;
+      if (shuttingDown) {
+        return;
+      }
+      if (done) {
+        readableClosed();
+        return;
+      }
+      const write = binding.WritableStreamDefaultWriterWrite(writer, value);
+      thenPromise(write, undefined, writableError);
+      pump();
+    }
+
+    function readRejected() {
+      reading = false;
+      readableError();
+    }
+
+    function readableError() {
+      if (!preventAbort) {
+        shutdownWithAction(binding.WritableStreamAbort,
+                           [dest, readable[_storedError]],
+                           readable[_storedError]);
+      } else {
+        shutdown(readable[_storedError]);
+      }
+    }
+
+    function writableError() {
+      const storedError = binding.WritableStreamStoredError(dest);
+      if (!preventCancel) {
+        shutdownWithAction(ReadableStreamCancel, [readable,
+                                                  storedError], storedError);
+      } else {
+        shutdown(storedError);
+      }
+    }
+
+    function readableClosed() {
+      if (reading) {
+        // Handle the close status from the read() method rather than the
+        // [[closedPromise]].
+        return;
+      }
+      if (!preventClose) {
+        shutdownWithAction(
+            binding.WritableStreamDefaultWriterCloseWithErrorPropagation,
+            [writer]);
+      } else {
+        shutdown();
+      }
+    }
+
+    function writableStartedClosed() {
+      const destClosed = new TypeError(errDestinationStreamClosed);
+      if (!preventCancel) {
+        shutdownWithAction(ReadableStreamCancel, [readable,
+                                                  destClosed], destClosed);
+      } else {
+        shutdown(destClosed);
+      }
+    }
+
+    function shutdownWithAction(action, args, originalError = undefined) {
+      if (shuttingDown) {
+        return;
+      }
+      shuttingDown = true;
+      const p = applyFunction(action, undefined, args);
+      thenPromise(p,
+                  () => finalize(originalError),
+                  newError => finalize(newError));
+    }
+
+    function shutdown(error = undefined) {
+      if (shuttingDown) {
+        return;
+      }
+      shuttingDown = true;
+      finalize(error);
+    }
+
+    function finalize(error) {
+      binding.WritableStreamDefaultWriterRelease(writer);
+      ReadableStreamReaderGenericRelease(reader);
+      if (error !== undefined) {

[domenic, 2017/01/18 23:51:17]

This seems like it will not work correctly when the error is actually undefined,
explicitly. We should have tests for that case; I guess we do not. In the
reference implementation I had to solve this by passing along an additional
boolean flag.

[Adam Rice, 2017/01/19 14:02:47]

Yep. I went with passing booleans around as well. I have written tests and will
submit them to the web-platform-tests repository.

+        v8.rejectPromise(promise, error);
+      } else {
+        v8.resolvePromise(promise, undefined);
+      }
+    }
+
+    return promise;
+  }
+
   class ReadableStreamDefaultController {
     constructor(stream, underlyingSource, size, highWaterMark, isExternallyControlled) {
       if (IsReadableStream(stream) === false) {
         throw new TypeError(streamErrors.illegalConstructor);
       }
 
       if (stream[_controller] !== undefined) {
         throw new TypeError(streamErrors.illegalConstructor);
       }
 
@@ skipping 710 matching lines @@
   binding.ReadableStreamDefaultControllerClose = ReadableStreamDefaultControllerClose;
   binding.ReadableStreamDefaultControllerGetDesiredSize = ReadableStreamDefaultControllerGetDesiredSize;
   binding.ReadableStreamDefaultControllerEnqueue = ReadableStreamDefaultControllerEnqueue;
   binding.ReadableStreamDefaultControllerError = ReadableStreamDefaultControllerError;
 
   binding.createReadableStreamWithExternalController =
       (underlyingSource, strategy) => {
         return new ReadableStream(
             underlyingSource, strategy, createWithExternalControllerSentinel);
       };
+
+  // Temporary exports while pipeTo() and pipeThrough() are behind flags
+  binding.ReadableStream_prototype_pipeThrough =
+      ReadableStream_prototype_pipeThrough;
+  binding.ReadableStream_prototype_pipeTo = ReadableStream_prototype_pipeTo;
 });