Implementation of ReadableStream pipeTo and pipeThrough

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

 // Copyright 2016 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.
 
 // Implementation of WritableStream for Blink.  See
 // https://streams.spec.whatwg.org/#ws. The implementation closely follows the
 // standard, except where required for performance or integration with Blink. In
 // particular, classes, methods and abstract operations are implemented in the
 // same order as in the standard, to simplify side-by-side reading.
 
@@ skipping 346 matching lines @@
     }
     if (backpressure) {
       writer[_readyPromise] = v8.createPromise();
     } else {
       TEMP_ASSERT(backpressure === false,
                   'backpressure is false.');
       v8.resolvePromise(writer[_readyPromise], undefined);
     }
   }
 
+  // Functions to expose internals for ReadableStream.pipeTo. These are not
+  // part of the standard.
+  function isWritableStreamErrored(stream) {
+    TEMP_ASSERT(
+        IsWritableStream(stream), '! IsWritableStream(stream) is true.');
+    return stream[_state] === ERRORED;
+  }
+
+  function isWritableStreamClosingOrClosed(stream) {
+    TEMP_ASSERT(
+        IsWritableStream(stream), '! IsWritableStream(stream) is true.');
+    return stream[_state] === CLOSING || stream[_state] === CLOSED;
+  }
+
+  function getWritableStreamStoredError(stream) {
+    TEMP_ASSERT(
+        IsWritableStream(stream), '! IsWritableStream(stream) is true.');
+    return stream[_storedError];
+  }
+
   class WritableStreamDefaultWriter {
     constructor(stream) {
       if (!IsWritableStream(stream)) {
         throw new TypeError(streamErrors.illegalConstructor);
       }
       if (IsWritableStreamLocked(stream)) {
         throw new TypeError(streamErrors.illegalConstructor);
       }
       this[_ownerWritableStream] = stream;
       stream[_writer] = this;
@@ skipping 120 matching lines @@
     stream[_pendingCloseRequest] = v8.createPromise();
     if (WritableStreamDefaultControllerGetBackpressure(
         stream[_writableStreamController])) {
       v8.resolvePromise(writer[_readyPromise], undefined);
     }
     stream[_state] = CLOSING;
     WritableStreamDefaultControllerClose(stream[_writableStreamController]);
     return stream[_pendingCloseRequest];
   }
 
+  function WritableStreamDefaultWriterCloseWithErrorPropagation(writer) {
+    const stream = writer[_ownerWritableStream];
+    TEMP_ASSERT(stream !== undefined, 'stream is not undefined.');
+    const state = stream[_state];
+    if (state === CLOSING || state === CLOSED) {
+      return Promise_resolve(undefined);
+    }
+    if (state === ERRORED) {
+      return Promise_reject(stream[_storedError]);
+    }
+    TEMP_ASSERT(state === WRITABLE, 'state is "writable".');
+    return WritableStreamDefaultWriterClose(writer);
+  }
+
   function WritableStreamDefaultWriterGetDesiredSize(writer) {
     const stream = writer[_ownerWritableStream];
     const state = stream[_state];
     if (state === ERRORED) {
       return null;
     }
     if (state === CLOSED) {
       return 0;
     }
     return WritableStreamDefaultControllerGetDesiredSize(
@@ skipping 39 matching lines @@
           createCannotActionOnStateStreamError('write to', state));
     }
     TEMP_ASSERT(state === WRITABLE,
                 'state is "writable".');
     const promise = WritableStreamAddWriteRequest(stream);
     WritableStreamDefaultControllerWrite(stream[_writableStreamController],
                                          chunk);
     return promise;
   }
 
+  // Functions to expose internals for ReadableStream.pipeTo. These do not
+  // appear in the standard.
+  function getWritableStreamDefaultWriterClosedPromise(writer) {
+    TEMP_ASSERT(
+        IsWritableStreamDefaultWriter(writer),
+        'writer is a WritableStreamDefaultWriter.');
+    return writer[_closedPromise];
+  }
+
+  function getWritableStreamDefaultWriterReadyPromise(writer) {
+    TEMP_ASSERT(
+        IsWritableStreamDefaultWriter(writer),
+        'writer is a WritableStreamDefaultWriter.');
+    return writer[_readyPromise];
+  }
+
   class WritableStreamDefaultController {
     constructor(stream, underlyingSink, size, highWaterMark) {
       if (!IsWritableStream(stream)) {
         throw new TypeError(streamErrors.illegalConstructor);
       }
       if (stream[_writableStreamController] !== undefined) {
         throw new TypeError(streamErrors.illegalConstructor);
       }
       this[_controlledWritableStream] = stream;
       this[_underlyingSink] = underlyingSink;
@@ skipping 351 matching lines @@
   //
 
   defineProperty(global, 'WritableStream', {
     value: WritableStream,
     enumerable: false,
     configurable: true,
     writable: true
   });
 
   // TODO(ricea): Exports to Blink
+
+  // Exports for ReadableStream
+  binding.AcquireWritableStreamDefaultWriter =
+      AcquireWritableStreamDefaultWriter;
+  binding.IsWritableStream = IsWritableStream;
+  binding.isWritableStreamClosingOrClosed = isWritableStreamClosingOrClosed;
+  binding.isWritableStreamErrored = isWritableStreamErrored;
+  binding.IsWritableStreamLocked = IsWritableStreamLocked;
+  binding.WritableStreamAbort = WritableStreamAbort;
+  binding.WritableStreamDefaultWriterCloseWithErrorPropagation =
+      WritableStreamDefaultWriterCloseWithErrorPropagation;
+  binding.getWritableStreamDefaultWriterClosedPromise =
+      getWritableStreamDefaultWriterClosedPromise;
+  binding.WritableStreamDefaultWriterGetDesiredSize =
+      WritableStreamDefaultWriterGetDesiredSize;
+  binding.getWritableStreamDefaultWriterReadyPromise =
+      getWritableStreamDefaultWriterReadyPromise;
+  binding.WritableStreamDefaultWriterRelease =
+      WritableStreamDefaultWriterRelease;
+  binding.WritableStreamDefaultWriterWrite = WritableStreamDefaultWriterWrite;
+  binding.getWritableStreamStoredError = getWritableStreamStoredError;
 });