Update writable streams for the latest spec changes

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.
 
 (function(global, binding, v8) {
   'use strict';
 
   // Private symbols. These correspond to the internal slots in the standard.
   // "[[X]]" in the standard is spelt _X in this implementation.
+  const _pendingWriteRequest = v8.createPrivateSymbol('[[pendingWriteRequest]]');
+  const _pendingCloseRequest = v8.createPrivateSymbol('[[pendingCloseRequest]]');
+  const _pendingAbortRequest = v8.createPrivateSymbol('[[pendingAbortRequest]]');
   const _state = v8.createPrivateSymbol('[[state]]');
   const _storedError = v8.createPrivateSymbol('[[storedError]]');
   const _writer = v8.createPrivateSymbol('[[writer]]');
   const _writableStreamController =
       v8.createPrivateSymbol('[[writableStreamController]]');
   const _writeRequests = v8.createPrivateSymbol('[[writeRequests]]');
   const _closedPromise = v8.createPrivateSymbol('[[closedPromise]]');
   const _ownerWritableStream =
       v8.createPrivateSymbol('[[ownerWritableStream]]');
   const _readyPromise = v8.createPrivateSymbol('[[readyPromise]]');
   const _controlledWritableStream =
       v8.createPrivateSymbol('[[controlledWritableStream]]');
   const _queue = v8.createPrivateSymbol('[[queue]]');
   const _queueSize = v8.createPrivateSymbol('[[queueSize]]');
   const _strategyHWM = v8.createPrivateSymbol('[[strategyHWM]]');
   const _strategySize = v8.createPrivateSymbol('[[strategySize]]');
   const _underlyingSink = v8.createPrivateSymbol('[[underlyingSink]]');
 
   // _defaultControllerFlags combines WritableStreamDefaultController's internal
-  // slots [[started]] and [[writing]] into a single bitmask for efficiency.
+  // slots [[started]], [[writing]], and [[inClose]] into a single bitmask for
+  // efficiency.
   const _defaultControllerFlags =
       v8.createPrivateSymbol('[[defaultControllerFlags]]');
   const FLAG_STARTED = 0b1;
   const FLAG_WRITING = 0b10;
+  const FLAG_INCLOSE = 0b100;
 
   // For efficiency, WritableStream [[state]] contains numeric values.
   const WRITABLE = 0;
   const CLOSING = 1;
   const CLOSED = 2;
   const ERRORED = 3;
 
   // Javascript functions. It is important to use these copies, as the ones on
   // the global object may have been overwritten. See "V8 Extras Design Doc",
   // section "Security Considerations".
   // https://docs.google.com/document/d/1AT5-T0aHGp7Lt29vPWFr2-qG8r3l9CByyvKwEuA8Ec0/edit#heading=h.9yixony1a18r
   const undefined = global.undefined;
 
   const defineProperty = global.Object.defineProperty;
   const hasOwnProperty = v8.uncurryThis(global.Object.hasOwnProperty);
 
-  const Function_call = v8.uncurryThis(global.Function.prototype.call);
   const Function_apply = v8.uncurryThis(global.Function.prototype.apply);
 
   const TypeError = global.TypeError;
   const RangeError = global.RangeError;
 
   const Boolean = global.Boolean;
   const Number = global.Number;
   const Number_isNaN = Number.isNaN;
   const Number_isFinite = Number.isFinite;
 
@@ skipping 55 matching lines @@
   }
 
   function getDefaultControllerWritingFlag(controller) {
     return Boolean(controller[_defaultControllerFlags] & FLAG_WRITING);
   }
 
   function setDefaultControllerWritingFlag(controller, value) {
     setDefaultControllerFlag(controller, FLAG_WRITING, value);
   }
 
+  function getDefaultControllerInCloseFlag(controller) {
+    return Boolean(controller[_defaultControllerFlags] & FLAG_INCLOSE);
+  }
+
+  function setDefaultControllerInCloseFlag(controller, value) {
+    setDefaultControllerFlag(controller, FLAG_INCLOSE, value);
+  }
+
   function rejectPromises(array, e) {
     // array is an InternalPackedArray so forEach won't work.
     for (let i = 0; i < array.length; ++i) {
       v8.rejectPromise(array[i], e);
     }
   }
 
   // https://tc39.github.io/ecma262/#sec-ispropertykey
   // TODO(ricea): Remove this when the asserts using it are removed.
   function IsPropertyKey(argument) {
     return typeof argument === 'string' || typeof argument === 'symbol';
   }
 
   // TODO(ricea): Remove all asserts once the implementation has stabilised.
   function TEMP_ASSERT(predicate, message) {
     if (predicate) {
       return;
     }
     v8.log(`Assertion failed: ${message}\n`);
     v8.logStackTrace();
-    class WritableStreamInternalError {
+    class WritableStreamInternalError extends Error {
+      constructor(message) {
+        super(message);
+      }
     }
-    throw new WritableStreamInternalError();
+    throw new WritableStreamInternalError(message);
   }
 
   class WritableStream {
     constructor(underlyingSink = {}, { size, highWaterMark = 1 } = {}) {
       this[_state] = WRITABLE;
       this[_storedError] = undefined;
       this[_writer] = undefined;
       this[_writableStreamController] = undefined;
+      this[_pendingWriteRequest] = undefined;
+      this[_pendingCloseRequest] = undefined;
+      this[_pendingAbortRequest] = undefined;
       this[_writeRequests] = new v8.InternalPackedArray();
       const type = underlyingSink.type;
       if (type !== undefined) {
         throw new RangeError(streamErrors.invalidType);
       }
       this[_writableStreamController] =
           new WritableStreamDefaultController(this, underlyingSink, size,
                                               highWaterMark);
     }
 
@@ skipping 43 matching lines @@
     if (state === CLOSED) {
       return Promise_resolve(undefined);
     }
     if (state === ERRORED) {
       return Promise_reject(stream[_storedError]);
     }
     TEMP_ASSERT(state === WRITABLE || state === CLOSING,
                'state is "writable" or "closing".');
     const error = new TypeError(errStreamAborted);
     WritableStreamError(stream, error);
-    return WritableStreamDefaultControllerAbort(
-        stream[_writableStreamController], reason);
+
+    const controller = stream[_writableStreamController];
+    TEMP_ASSERT(controller !== undefined,
+                'controller is not undefined');
+
+    const isWriting = getDefaultControllerWritingFlag(controller);
+    if (isWriting || getDefaultControllerInCloseFlag(controller)) {
+      const promise =  v8.createPromise();
+      stream[_pendingAbortRequest] = promise;
+
+      if (isWriting) {
+        return thenPromise(promise, () => {
+          return WritableStreamDefaultControllerAbort(controller, reason);
+        });
+      }
+      return promise;
+    }
+
+    return WritableStreamDefaultControllerAbort(controller, reason);
   }
 
   // Writable Stream Abstract Operations Used by Controllers
 
   function WritableStreamAddWriteRequest(stream) {
     TEMP_ASSERT(IsWritableStreamLocked(stream),
                 '! IsWritableStreamLocked(writer) is true.');
     TEMP_ASSERT(stream[_state] === WRITABLE,
                 'stream.[[state]] is "writable".');
     const promise = v8.createPromise();
     stream[_writeRequests].push(promise);
     return promise;
   }
 
   function WritableStreamError(stream, e) {
-    const state = stream[_state];
-    TEMP_ASSERT(state === WRITABLE || state === CLOSING,
-                'state is "writable" or "closing".');
-    rejectPromises(stream[_writeRequests], e);
-    stream[_writeRequests] = new v8.InternalPackedArray();
+    const oldState = stream[_state];
+    TEMP_ASSERT(oldState === WRITABLE || oldState === CLOSING,
+                'oldState is "writable" or "closing".');
+
+    stream[_state] = ERRORED;
+    stream[_storedError] = e;
+
+    const controller = stream[_writableStreamController];
+    if (controller === undefined ||
+        (!getDefaultControllerWritingFlag(controller) &&
+         !getDefaultControllerInCloseFlag(controller))) {
+      WritableStreamRejectPromisesInReactionToError(stream);
+    }
+
     const writer = stream[_writer];
     if (writer !== undefined) {
-      v8.rejectPromise(writer[_closedPromise], e);
-      if (state === WRITABLE &&
-          WritableStreamDefaultControllerGetBackpressure(
-              stream[_writableStreamController])) {
+      if (oldState === WRITABLE &&
+          WritableStreamDefaultControllerGetBackpressure(controller) === true) {
         v8.rejectPromise(writer[_readyPromise], e);
       } else {
         writer[_readyPromise] = Promise_reject(e);
       }
+      v8.markPromiseAsHandled(writer[_readyPromise]);
     }
-    stream[_state] = ERRORED;
-    stream[_storedError] = e;
   }
 
   function WritableStreamFinishClose(stream) {
-    TEMP_ASSERT(stream[_state] === CLOSING,
-                'stream.[[state]] is "closing".');
-    TEMP_ASSERT(stream[_writer] !== undefined,
-                'stream.[[writer]] is not undefined.');
-    stream[_state] = CLOSED;
-    v8.resolvePromise(stream[_writer][_closedPromise], undefined);
+    const state = stream[_state];
+    TEMP_ASSERT(state === CLOSING || state === ERRORED,
+                'state is "closing" or "errored"');
+
+    if (state === CLOSING) {
+      v8.resolvePromise(stream[_writer][_closedPromise], undefined);
+      stream[_state] = CLOSED;
+    } else {
+      TEMP_ASSERT(state === ERRORED, 'state is "errored"');
+      v8.rejectPromise(stream[_writer][_closedPromise], stream[_storedError]);
+      v8.markPromiseAsHandled(stream[_writer][_closedPromise]);
+    }
+
+    if (stream[_pendingAbortRequest] !== undefined) {
+      v8.resolvePromise(stream[_pendingAbortRequest], undefined);
+      stream[_pendingAbortRequest] = undefined;
+    }
   }
 
-  function WritableStreamFulfillWriteRequest(stream) {
-    TEMP_ASSERT(stream[_writeRequests].length !== 0,
-               'stream.[[writeRequests]] is not empty.');
-    const writeRequest = stream[_writeRequests].shift();
-    v8.resolvePromise(writeRequest, undefined);
+  function WritableStreamRejectPromisesInReactionToError(stream) {
+    TEMP_ASSERT(stream[_state] === ERRORED, 'stream.[[state]] is "errored"');
+    TEMP_ASSERT(stream[_pendingWriteRequest] === undefined,
+                'stream.[[pendingWriteRequest]] is undefined');
+
+    const storedError = stream[_storedError];
+    rejectPromises(stream[_writeRequests], storedError);
+    stream[_writeRequests] = new v8.InternalPackedArray();
+
+    if (stream[_pendingCloseRequest] !== undefined) {
+      TEMP_ASSERT(
+          getDefaultControllerInCloseFlag(stream[_writableStreamController]) ===
+          false, 'stream.[[writableStreamController]].[[inClose]] === false');
+      v8.rejectPromise(stream[_pendingCloseRequest], storedError);
+      stream[_pendingCloseRequest] = undefined;
+    }
+
+    const writer = stream[_writer];
+    if (writer !== undefined) {
+      v8.rejectPromise(writer[_closedPromise], storedError);
+      v8.markPromiseAsHandled(writer[_closedPromise]);
+    }
   }
 
   function WritableStreamUpdateBackpressure(stream, backpressure) {
     TEMP_ASSERT(stream[_state] === WRITABLE,
                 'stream.[[state]] is "writable".');
     const writer = stream[_writer];
     if (writer === undefined) {
       return;
     }
     if (backpressure) {
@@ skipping 17 matching lines @@
       stream[_writer] = this;
       const state = stream[_state];
       if (state === WRITABLE || state === CLOSING) {
         this[_closedPromise] = v8.createPromise();
       } else if (state === CLOSED) {
         this[_closedPromise] = Promise_resolve(undefined);
       } else {
         TEMP_ASSERT(state === ERRORED,
                     'state is "errored".');
         this[_closedPromise] = Promise_reject(stream[_storedError]);
+        v8.markPromiseAsHandled(this[_closedPromise]);
       }
       if (state === WRITABLE &&
           WritableStreamDefaultControllerGetBackpressure(
               stream[_writableStreamController])) {
         this[_readyPromise] = v8.createPromise();
       } else {
         this[_readyPromise] = Promise_resolve(undefined);
       }
     }
 
@@ skipping 90 matching lines @@
     const stream = writer[_ownerWritableStream];
     TEMP_ASSERT(stream !== undefined,
                 'stream is not undefined.');
     const state = stream[_state];
     if (state === CLOSED || state === ERRORED) {
       return Promise_reject(
           createCannotActionOnStateStreamError('close', state));
     }
     TEMP_ASSERT(state === WRITABLE,
                 'state is "writable".');
-    const promise = WritableStreamAddWriteRequest(stream);
+    stream[_pendingCloseRequest] = v8.createPromise();
     if (WritableStreamDefaultControllerGetBackpressure(
         stream[_writableStreamController])) {
       v8.resolvePromise(writer[_readyPromise], undefined);
     }
     stream[_state] = CLOSING;
     WritableStreamDefaultControllerClose(stream[_writableStreamController]);
-    return promise;
+    return stream[_pendingCloseRequest];
   }
 
   function WritableStreamDefaultWriterGetDesiredSize(writer) {
     const stream = writer[_ownerWritableStream];
     const state = stream[_state];
     if (state === ERRORED) {
       return null;
     }
     if (state === CLOSED) {
       return 0;
     }
     return WritableStreamDefaultControllerGetDesiredSize(
         stream[_writableStreamController]);
   }
 
   function WritableStreamDefaultWriterRelease(writer) {
     const stream = writer[_ownerWritableStream];
     TEMP_ASSERT(stream !== undefined,
                 'stream is not undefined.');
     TEMP_ASSERT(stream[_writer] === writer,
                 'stream.[[writer]] is writer.');
     const releasedError = new TypeError(errReleasedWriterClosedPromise);
     const state = stream[_state];
-    if (state === WRITABLE || state === CLOSING) {
+    if (state === WRITABLE || state === CLOSING ||
+        stream[_pendingAbortRequest] !== undefined) {
       v8.rejectPromise(writer[_closedPromise], releasedError);
     } else {
       writer[_closedPromise] = Promise_reject(releasedError);
     }
+    v8.markPromiseAsHandled(writer[_closedPromise]);
+
     if (state === WRITABLE &&
         WritableStreamDefaultControllerGetBackpressure(
             stream[_writableStreamController])) {
       v8.rejectPromise(writer[_readyPromise], releasedError);
     } else {
       writer[_readyPromise] = Promise_reject(releasedError);
     }
+    v8.markPromiseAsHandled(writer[_readyPromise]);
+
     stream[_writer] = undefined;
     writer[_ownerWritableStream] = undefined;
   }
 
   function WritableStreamDefaultWriterWrite(writer, chunk) {
     const stream = writer[_ownerWritableStream];
     TEMP_ASSERT(stream !== undefined,
                 'stream is not undefined.');
     const state = stream[_state];
     if (state === CLOSED || state === ERRORED) {
@@ skipping 76 matching lines @@
   function WritableStreamDefaultControllerGetDesiredSize(controller) {
     const queueSize = GetTotalQueueSizeForController(controller);
     return controller[_strategyHWM] - queueSize;
   }
 
   function WritableStreamDefaultControllerWrite(controller, chunk) {
     const stream = controller[_controlledWritableStream];
     TEMP_ASSERT(stream[_state] === WRITABLE,
                 'stream.[[state]] is "writable".');
     let chunkSize = 1;
-    if (controller[_strategySize] !== undefined) {
+    const strategySize = controller[_strategySize];
+    if (strategySize !== undefined) {
       try {
-        chunkSize = Function_call(controller[_strategySize], undefined, chunk);
+        chunkSize = strategySize(chunk);
       } catch (e) {
         WritableStreamDefaultControllerErrorIfNeeded(controller, e);
-        return Promise_reject(e);
+        return;
       }
     }
     const writeRecord = {chunk};
     const lastBackpressure =
         WritableStreamDefaultControllerGetBackpressure(controller);
     try {
-      const enqueueResult =
-          EnqueueValueWithSizeForController(controller, writeRecord, chunkSize);
+      EnqueueValueWithSizeForController(controller, writeRecord, chunkSize);
     } catch (e) {
       WritableStreamDefaultControllerErrorIfNeeded(controller, e);
-      return Promise_reject(e);
+      return;
     }
     if (stream[_state] === WRITABLE) {
       const backpressure =
           WritableStreamDefaultControllerGetBackpressure(controller);
       if (lastBackpressure !== backpressure) {
         WritableStreamUpdateBackpressure(stream, backpressure);
       }
     }
     WritableStreamDefaultControllerAdvanceQueueIfNeeded(controller);
   }
@@ skipping 28 matching lines @@
     }
   }
 
   function WritableStreamDefaultControllerProcessClose(controller) {
     const stream = controller[_controlledWritableStream];
     TEMP_ASSERT(stream[_state] === CLOSING,
                 'stream.[[state]] is "closing".');
     DequeueValueForController(controller);
     TEMP_ASSERT(controller[_queue].length === 0,
                 'controller.[[queue]] is empty.');
+    setDefaultControllerInCloseFlag(controller, true);
     const sinkClosePromise = PromiseInvokeOrNoop(controller[_underlyingSink],
-                                               'close', [controller]);
+                                                 'close', [controller]);
     thenPromise(sinkClosePromise,
                 () => {
-                  if (stream[_state] !== CLOSING) {
+                  TEMP_ASSERT(getDefaultControllerInCloseFlag(controller)
+                      === true,
+                      'controller.[[inClose]] is true');
+                  setDefaultControllerInCloseFlag(controller, false);
+
+                  if (stream[_state] !== CLOSING &&
+                      stream[_state] !== ERRORED) {
                     return;
                   }
-                  WritableStreamFulfillWriteRequest(stream);
+
+                  TEMP_ASSERT(stream[_pendingCloseRequest] !== undefined);
+                  v8.resolvePromise(stream[_pendingCloseRequest], undefined);
+                  stream[_pendingCloseRequest] = undefined;
+
                   WritableStreamFinishClose(stream);
                 },
-                r => WritableStreamDefaultControllerErrorIfNeeded(controller, r)
+                r => {
+                  TEMP_ASSERT(getDefaultControllerInCloseFlag(controller)
+                      === true,
+                      'controller.[[inClose]] is true');
+                  setDefaultControllerInCloseFlag(controller, false);
+
+                  TEMP_ASSERT(stream[_pendingCloseRequest] !== undefined);
+                  v8.rejectPromise(stream[_pendingCloseRequest], r);
+                  stream[_pendingCloseRequest] = undefined;
+
+                  if (stream[_pendingAbortRequest] !== undefined) {
+                    v8.rejectPromise(stream[_pendingAbortRequest], r);
+                    stream[_pendingAbortRequest] = undefined;
+                  }
+
+                  WritableStreamDefaultControllerErrorIfNeeded(controller, r);
+                }
                );
   }
 
   function WritableStreamDefaultControllerProcessWrite(controller, chunk) {
     setDefaultControllerWritingFlag(controller, true);
+
+    const stream = controller[_controlledWritableStream];
+
+    TEMP_ASSERT(stream[_pendingWriteRequest] === undefined,
+                'stream.[[pendingWriteRequest]] is undefined');
+    TEMP_ASSERT(stream[_writeRequests].length > 0,
+                'stream.[[writeRequests]] is not empty');
+    stream[_pendingWriteRequest] = stream[_writeRequests].shift();
+
     const sinkWritePromise = PromiseInvokeOrNoop(controller[_underlyingSink],
                                                'write', [chunk, controller]);
     thenPromise(
         sinkWritePromise,
         () => {
-          const stream = controller[_controlledWritableStream];
+          TEMP_ASSERT(getDefaultControllerWritingFlag(controller) === true,
+                      'controller.[[writing]] is true');
+          setDefaultControllerWritingFlag(controller, false);
+
+          TEMP_ASSERT(stream[_pendingWriteRequest] !== undefined,
+                      'stream.[[pendingWriteRequest]] is not undefined');
+          v8.resolvePromise(stream[_pendingWriteRequest], undefined);
+          stream[_pendingWriteRequest] = undefined;
+
           const state = stream[_state];
-          if (state === ERRORED || state === CLOSED) {
+          if (state === ERRORED) {
+            WritableStreamRejectPromisesInReactionToError(stream);
+
+            if (stream[_pendingAbortRequest] !== undefined) {
+              v8.resolvePromise(stream[_pendingAbortRequest], undefined);
+              stream[_pendingAbortRequest] = undefined;
+            }
             return;
           }
-          setDefaultControllerWritingFlag(controller, false);
-          WritableStreamFulfillWriteRequest(stream);
+
           const lastBackpressure =
               WritableStreamDefaultControllerGetBackpressure(controller);
           DequeueValueForController(controller);
+
           if (state !== CLOSING) {
             const backpressure =
                 WritableStreamDefaultControllerGetBackpressure(controller);
             if (lastBackpressure !== backpressure) {
               WritableStreamUpdateBackpressure(
                   controller[_controlledWritableStream], backpressure);
             }
           }
+
           WritableStreamDefaultControllerAdvanceQueueIfNeeded(controller);
         },
-        r => WritableStreamDefaultControllerErrorIfNeeded(controller, r)
-        );
+        r => {
+          TEMP_ASSERT(getDefaultControllerWritingFlag(controller) === true,
+                      'controller.[[writing]] is true');
+          setDefaultControllerWritingFlag(controller, false);
+
+          TEMP_ASSERT(stream[_pendingWriteRequest] !== undefined,
+                      'stream.[[pendingWriteRequest]] is not undefined');
+          v8.rejectPromise(stream[_pendingWriteRequest], r);
+          stream[_pendingWriteRequest] = undefined;
+
+          if (stream[_state] === ERRORED) {
+            stream[_storedError] = r;
+            WritableStreamRejectPromisesInReactionToError(stream);
+          }
+
+          if (stream[_pendingAbortRequest] !== undefined) {
+            v8.rejectPromise(stream[_pendingAbortRequest], r);
+            stream[_pendingAbortRequest] = undefined;
+          }
+
+          WritableStreamDefaultControllerErrorIfNeeded(controller, r);
+        }
+    );
   }
 
   function WritableStreamDefaultControllerGetBackpressure(controller) {
     const desiredSize =
         WritableStreamDefaultControllerGetDesiredSize(controller);
     return desiredSize <= 0;
   }
 
   function WritableStreamDefaultControllerError(controller, e) {
     const stream = controller[_controlledWritableStream];
@@ skipping 117 matching lines @@
 
   defineProperty(global, 'WritableStream', {
     value: WritableStream,
     enumerable: false,
     configurable: true,
     writable: true
   });
 
   // TODO(ricea): Exports to Blink
 });