Ship ReadableStream pipeTo()

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 158 matching lines @@
         throw new TypeError(errGetReaderNotByteStream);
       }
 
       if (mode === undefined) {
         return AcquireReadableStreamDefaultReader(this);
       }
 
       throw new RangeError(errGetReaderBadMode);
     }
 
+    pipeThrough({writable, readable}, options) {
+      this.pipeTo(writable, options);
+      return readable;

[domenic, 2017/04/03 05:49:30]

Shouldn't this mark the promise as handled?

[Adam Rice, 2017/04/03 15:08:15]

I want to update the implementation in a separate CL like I'm doing for
WritableStream. I created a CL which implements this change:

https://codereview.chromium.org/2796633002/

+    }
+
+    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);
+
+      if (IsReadableStreamLocked(this)) {
+        return Promise_reject(new TypeError(errCannotPipeLockedStream));
+      }
+
+      if (binding.IsWritableStreamLocked(dest)) {
+        return Promise_reject(new TypeError(errCannotPipeToALockedStream));
+      }
+
+      return ReadableStreamPipeTo(this, dest, preventClose, preventAbort,
+                                  preventCancel);
+    }
+
     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));
-    }
+  function ReadableStreamPipeTo(readable, dest, preventClose, preventAbort,
+                                preventCancel) {
+    // assert(IsReadableStream(readable));
+    // assert(binding.IsWritableStream(dest));
+    // assert(!IsReadableStreamLocked(readable));
+    // assert(!binding.IsWritableStreamLocked(dest));

[tyoshino (SeeGerritForStatus), 2017/04/05 06:08:41]

Please write some comment in natural language to explain the intention of these
lines.

[Adam Rice, 2017/04/05 07:03:47]

Done.

 
     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);
@@ skipping 873 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;
 });