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 (checkInitialState()) {
+      // Need to detect closing and error when we are not reading.
+      thenPromise(reader[_closedPromise], onReaderClosed, readableError);
+      // Need to detect error when we are not writing.
+      thenPromise(
+          binding.getWritableStreamDefaultWriterClosedPromise(writer),
+          undefined, writableError);
+      pump();
+    }
+
+    // Checks the state of the streams and executes the shutdown handlers if
+    // necessary. Returns true if piping can continue.
+    function checkInitialState() {
+      const state = ReadableStreamGetState(readable);
+
+      // Both streams can be errored or closed. To perform the right action the
+      // order of the checks must match the standard.
+      if (state === STATE_ERRORED) {
+        readableError(readable[_storedError]);
+        return false;
+      }
+
+      if (binding.isWritableStreamErrored(dest)) {
+        writableError(binding.getWritableStreamStoredError(dest));
+        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(binding.getWritableStreamStoredError(dest));
+      }
+      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();
+        return;
+      }
+      const write = binding.WritableStreamDefaultWriterWrite(writer, value);
+      thenPromise(write, undefined, writableError);
+      pump();
+    }
+
+    function readRejected() {
+      reading = false;
+      readableError(readable[_storedError]);
+    }
+
+    // If read() is in progress, then wait for it to tell us that the stream is
+    // closed so that we write all the data before shutdown.
+    function onReaderClosed() {
+      if (!reading) {
+        readableClosed();
+      }
+    }
+
+    // These steps are from "Errors must be propagated forward" in the
+    // standard.
+    function readableError(error) {
+      if (!preventAbort) {
+        shutdownWithAction(
+            binding.WritableStreamAbort, [dest, error], error, true);
+      } else {
+        shutdown(error, true);
+      }
+    }
+
+    // These steps are from "Errors must be propagated backward".
+    function writableError(error) {
+      if (!preventCancel) {
+        shutdownWithAction(
+            ReadableStreamCancel, [readable, error], error, true);
+      } else {
+        shutdown(error, true);
+      }
+    }
+
+    // These steps are from "Closing must be propagated forward".
+    function readableClosed() {
+      if (!preventClose) {
+        shutdownWithAction(
+            binding.WritableStreamDefaultWriterCloseWithErrorPropagation,
+            [writer]);
+      } else {
+        shutdown();
+      }
+    }
+
+    // These steps are from "Closing must be propagated backward".
+    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) {
+      if (shuttingDown) {
+        return;
+      }
+      shuttingDown = true;
+      const p = applyFunction(action, undefined, args);
+      thenPromise(
+          p, () => finalize(originalError, errorGiven),
+          newError => finalize(newError, true));
+    }
+
+    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 772 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;
 });