Use SimpleQueue in WritableStream implementation

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 123 matching lines @@
   }
 
   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);
-    }
+  function rejectPromises(queue, e) {
+    queue.forEach(promise => v8.rejectPromise(promise, 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) {
@@ skipping 12 matching lines @@
 
   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();
+      this[_writeRequests] = new binding.SimpleQueue();
       const type = underlyingSink.type;
       if (type !== undefined) {
         throw new RangeError(streamErrors.invalidType);
       }
       this[_writableStreamController] =
           new WritableStreamDefaultController(this, underlyingSink, size,
                                               highWaterMark);
     }
 
     get locked() {
@@ skipping 132 matching lines @@
     }
   }
 
   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();
+    stream[_writeRequests] = new binding.SimpleQueue();
 
     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];
@@ skipping 271 matching lines @@
   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;
-      this[_queue] = new v8.InternalPackedArray();
+      this[_queue] = new binding.SimpleQueue();
       this[_queueSize] = 0;
       this[_defaultControllerFlags] = 0;
       const normalizedStrategy =
           ValidateAndNormalizeQueuingStrategy(size, highWaterMark);
       this[_strategySize] = normalizedStrategy.size;
       this[_strategyHWM] = normalizedStrategy.highWaterMark;
       const backpressure = WritableStreamDefaultControllerGetBackpressure(this);
       if (backpressure) {
         WritableStreamUpdateBackpressure(stream, backpressure);
       }
@@ skipping 21 matching lines @@
     }
   }
 
   // Writable Stream Default Controller Abstract Operations
 
   function IsWritableStreamDefaultController(x) {
     return hasOwnProperty(x, _underlyingSink);
   }
 
   function WritableStreamDefaultControllerAbort(controller, reason) {
-    controller[_queue] = v8.InternalPackedArray();
+    controller[_queue] = new binding.SimpleQueue();
     controller[_queueSize] = 0;
     const sinkAbortPromise =
         PromiseInvokeOrNoop(controller[_underlyingSink], 'abort', [reason]);
     return thenPromise(sinkAbortPromise, () => undefined);
   }
 
   function WritableStreamDefaultControllerClose(controller) {
     EnqueueValueWithSizeForController(controller, 'close', 0);
     WritableStreamDefaultControllerAdvanceQueueIfNeeded(controller);
   }
@@ skipping 119 matching lines @@
 
     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]);
+                                                 'write', [chunk, controller]);
     thenPromise(
         sinkWritePromise,
         () => {
           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);
@@ skipping 55 matching lines @@
         WritableStreamDefaultControllerGetDesiredSize(controller);
     return desiredSize <= 0;
   }
 
   function WritableStreamDefaultControllerError(controller, e) {
     const stream = controller[_controlledWritableStream];
     const state = stream[_state];
     TEMP_ASSERT(state === WRITABLE || state === CLOSING,
                 'stream.[[state]] is "writable" or "closing".');
     WritableStreamError(stream, e);
-    controller[_queue] = new v8.InternalPackedArray();
+    controller[_queue] = new binding.SimpleQueue();
     controller[_queueSize] = 0;
   }
 
   // Queue-with-Sizes Operations
   //
-  // These differ from the versions in the standard: they take a controller
-  // argument in order to cache the total queue size. This is necessary to avoid
-  // O(N^2) behaviour.
-  //
   // TODO(ricea): Share these operations with ReadableStream.js.
   function DequeueValueForController(controller) {
     TEMP_ASSERT(controller[_queue].length !== 0,
                 'queue is not empty.');
     const result = controller[_queue].shift();
     controller[_queueSize] -= result.size;
     return result.value;
   }
 
   function EnqueueValueWithSizeForController(controller, value, size) {
     size = Number(size);
     if (!IsFiniteNonNegativeNumber(size)) {
       throw new RangeError(streamErrors.invalidSize);
     }
 
     controller[_queueSize] += size;
     controller[_queue].push({value, size});
   }
 
   function GetTotalQueueSizeForController(controller) {
     return controller[_queueSize];
   }
 
   function PeekQueueValue(queue) {
     TEMP_ASSERT(queue.length !== 0,
                 'queue is not empty.');
-    return queue[0].value;
+    return queue.peek().value;
   }
 
   // Miscellaneous Operations
 
   // This differs from "CallOrNoop" in the ReadableStream implementation in
   // that it takes the arguments as an array, so that multiple arguments can be
   // passed.
   //
   // TODO(ricea): Consolidate with ReadableStream implementation.
   function InvokeOrNoop(O, P, args) {
@@ skipping 68 matching lines @@
       getWritableStreamDefaultWriterClosedPromise;
   binding.WritableStreamDefaultWriterGetDesiredSize =
       WritableStreamDefaultWriterGetDesiredSize;
   binding.getWritableStreamDefaultWriterReadyPromise =
       getWritableStreamDefaultWriterReadyPromise;
   binding.WritableStreamDefaultWriterRelease =
       WritableStreamDefaultWriterRelease;
   binding.WritableStreamDefaultWriterWrite = WritableStreamDefaultWriterWrite;
   binding.getWritableStreamStoredError = getWritableStreamStoredError;
 });