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.getWritableStreamDefaultWriterClosedPromise(writer),
+                  undefined, writableError);
+      pump();
+    }
+
+    function initialStateOk() {

[tyoshino (SeeGerritForStatus), 2017/01/23 10:42:57]

This function name sounds like checking states and returning go/no-go without
side-effect, but this is actually handling some initial state values to perform
state transition.

Maybe you factored out the following logic and named this function like this for
describing the meaning of the return value while avoiding redundant code. I
understand that motivation but to follow our usual convention, how about name
this like checkInitialState() and add a small comment about the return value?

[Adam Rice, 2017/01/23 13:24:30]

Done.

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

[tyoshino (SeeGerritForStatus), 2017/01/23 10:42:57]

are these error check blocks for optimization?

[Adam Rice, 2017/01/23 13:24:30]

They are to perform the checks in the correct order. This is the order that
passes the tests; it's not currently specified by the standard.

The uncommitted pull request https://github.com/whatwg/streams/pull/647 contains
more details.

[tyoshino (SeeGerritForStatus), 2017/01/24 09:52:33]

Oh, thanks for the pointer. Sorry for being unaware of that.

+
+      if (state === STATE_CLOSED) {
+        readableClosed();

[tyoshino (SeeGerritForStatus), 2017/01/23 10:42:57]

how about calling factoring out the latter half of readableClosed() and call it
here?

[Adam Rice, 2017/01/23 13:24:30]

The first check in readableClosed() didn't belong in that operation, so I have
moved it to a separate onReaderClosed() function instead.

+        return false;
+      }

[tyoshino (SeeGerritForStatus), 2017/01/23 10:42:57]

Is this an optimization or to have pipeTo finish successfully even if the dest
is errored/closed/closing if preventAbort and preventClose are specified?

if the latter point is concerned, this should go before the block for checking
dest's error (L241).

[Adam Rice, 2017/01/23 13:24:30]

It's purely to perform the checks in the correct order to pass the tests.

+
+      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.getWritableStreamDefaultWriterReadyPromise(writer),
+                    pump, writableError);
+        return;
+      }
+      reading = true;
+      // TODO(ricea): Delay reads heuristically when desiredSize is low.
+      thenPromise(ReadableStreamDefaultReaderRead(reader), readFulfilled,
+                  readRejected);
+    }
+
+    function readFulfilled({value, done}) {
+      reading = false;
+      if (shuttingDown) {
+        return;
+      }
+      if (done) {
+        readableClosed();

[tyoshino (SeeGerritForStatus), 2017/01/23 10:42:57]

how about calling factoring out the latter half of readableClosed() and call it
here?

[Adam Rice, 2017/01/23 13:24:30]

I removed the first half instead, to make the operations match the ones in the
standard.

+        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], true);
+      } else {
+        shutdown(readable[_storedError], true);
+      }
+    }
+
+    function writableError() {
+      const storedError = binding.getWritableStreamStoredError(dest);

[tyoshino (SeeGerritForStatus), 2017/01/23 10:42:57]

can't we use the rejection value? or intentionally using this for better
correspondence with the spec text?

same about readableError()

[Adam Rice, 2017/01/23 13:24:30]

I'm not sure about this, because now we're looking up the [[storedError]] in two
different places instead of inside this function. PTAL.

+      if (!preventCancel) {
+        shutdownWithAction(ReadableStreamCancel, [readable, storedError],
+                           storedError, true);
+      } else {
+        shutdown(storedError, true);
+      }
+    }
+
+    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, true);
+      } else {
+        shutdown(destClosed, true);
+      }
+    }
+
+    function shutdownWithAction(action, args, originalError = undefined,
+                                errorGiven = false) {

[tyoshino (SeeGerritForStatus), 2017/01/23 10:42:57]

default argument is used for better correspondence with the spec text where
"optionally ..." idiom is used?

[Adam Rice, 2017/01/23 13:24:30]

Exactly. This worked better before Domenic pointed out that I needed to add the
errorGiven boolean.

+      if (shuttingDown) {
+        return;
+      }
+      shuttingDown = true;
+      const p = applyFunction(action, undefined, args);
+      thenPromise(p,
+                  () => finalize(originalError, errorGiven),
+                  newError => finalize(newError, true));
+    }

[tyoshino (SeeGerritForStatus), 2017/01/23 10:42:57]

ah, we need to update the reference implementation by removing the logic to wait
for ongoing write().

[Adam Rice, 2017/01/23 13:24:30]

Yes. I discussed this with Domenic but it looks like I never filed an issue
about it. I just filed https://github.com/whatwg/streams/issues/668 for the
standard text change.

+
+    function shutdown(error = undefined, errorGiven = false) {
+      if (shuttingDown) {
+        return;
+      }
+      shuttingDown = true;
+      finalize(error, errorGiven);
+    }
+
+    function finalize(error, errorGiven) {
+      binding.WritableStreamDefaultWriterRelease(writer);
+      ReadableStreamReaderGenericRelease(reader);
+      if (errorGiven) {
+        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;
 });