Change data pipe wrappers used by SerialConnection to use message pipe.

extensions/renderer/resources/data_receiver.js

 // Copyright 2014 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.
 
 define('data_receiver', [
-    'async_waiter',
     'device/serial/data_stream.mojom',
     'device/serial/data_stream_serialization.mojom',
     'mojo/public/js/bindings/core',
     'mojo/public/js/bindings/router',
-], function(asyncWaiter, dataStream, serialization, core, router) {
+], function(dataStream, serialization, core, router) {
   /**
    * @module data_receiver
    */
 
   /**
    * A pending receive operation.
    * @constructor
    * @alias module:data_receiver~PendingReceive
    * @private
    */
@@ skipping 36 matching lines @@
    */
   PendingReceive.prototype.dispatchData = function(data) {
     this.dataCallback_(data);
   };
 
   /**
    * Dispatches an error if the offset of the error has been reached.
    * @param {!PendingReceiveError} error The error to dispatch.
    * @param {number} bytesReceived The number of bytes that have been received.
    */
-  PendingReceive.prototype.dispatchError = function(error, bytesReceived) {
-    if (bytesReceived != error.offset)
+  PendingReceive.prototype.dispatchError = function(error) {
+    if (error.queuePosition > 0)
       return false;
 
     var e = new Error();
     e.error = error.error;
     this.errorCallback_(e);
     return true;
   };
 
   /**
    * Unconditionally dispatches an error.
    * @param {number} error The error to dispatch.
    */
   PendingReceive.prototype.dispatchFatalError = function(error) {
     var e = new Error();
     e.error = error;
     this.errorCallback_(e);
   };
 
   /**
    * A DataReceiver that receives data from a DataSource.
    * @param {!MojoHandle} handle The handle to the DataSource.
-   * @param {number} bufferSize How large a buffer the data pipe should use.
+   * @param {number} bufferSize How large a buffer to use.
    * @param {number} fatalErrorValue The receive error value to report in the
    *     event of a fatal error.
    * @constructor
    * @alias module:data_receiver.DataReceiver
    */
   function DataReceiver(handle, bufferSize, fatalErrorValue) {
-    var dataPipeOptions = {
-      flags: core.CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,
-      elementNumBytes: 1,
-      capacityNumBytes: bufferSize,
-    };
-    var receivePipe = core.createDataPipe(dataPipeOptions);
-    this.init_(
-        handle, receivePipe.consumerHandle, fatalErrorValue, 0, null, false);
-    this.source_.init(receivePipe.producerHandle);
+    this.init_(handle, fatalErrorValue, 0, null, [], false);
+    this.source_.init(bufferSize);
   }
 
   DataReceiver.prototype =
       $Object.create(dataStream.DataSourceClient.stubClass.prototype);
 
   /**
    * Closes this DataReceiver.
    */
   DataReceiver.prototype.close = function() {
     if (this.shutDown_)
       return;
     this.shutDown_ = true;
     this.router_.close();
-    this.waiter_.stop();
-    core.close(this.receivePipe_);
     if (this.receive_) {
       this.receive_.dispatchFatalError(this.fatalErrorValue_);
       this.receive_ = null;
     }
   };
 
   /**
    * Initialize this DataReceiver.
    * @param {!MojoHandle} source A handle to the DataSource
-   * @param {!MojoHandle} dataPipe A handle to use for receiving data from the
-   *     DataSource.
    * @param {number} fatalErrorValue The error to dispatch in the event of a
    *     fatal error.
    * @param {number} bytesReceived The number of bytes already received.
    * @param {PendingReceiveError} pendingError The pending error if there is
    * one.
+   * @param {ArrayBuffer[]} pendingData Data received from the DataSource not

[Ken Rockot(use gerrit already), 2014/10/28 04:59:33]

At least the current version of closure compiler doesn't seem to support this
syntax for array type specs. Also it's ambiguous about nullability (unless maybe
you said !(!ArrayBuffer[]) which is possibly also illegal). In any case, an
unambiguous and legal form would be:

{!Array.<!ArrayBuffer>}

[Sam McNally, 2014/10/28 05:11:10]

Done.

+   * yet requested by the client.
    * @param {boolean} paused Whether the DataSource is paused.
    * @private
    */
   DataReceiver.prototype.init_ = function(source,
-                                          dataPipe,
                                           fatalErrorValue,
                                           bytesReceived,
                                           pendingError,
+                                          pendingData,
                                           paused) {
     /**
      * The [Router]{@link module:mojo/public/js/bindings/router.Router} for the
      * connection to the DataSource.
      * @private
      */
     this.router_ = new router.Router(source);
     /**
      * The connection to the DataSource.
      * @private
      */
     this.source_ = new dataStream.DataSource.proxyClass(this.router_);
     this.router_.setIncomingReceiver(this);
     /**
-     * The handle to the data pipe to use for receiving data.
-     * @private
-     */
-    this.receivePipe_ = dataPipe;
-    /**
      * The current receive operation.
      * @type {module:data_receiver~PendingReceive}
      * @private
      */
     this.receive_ = null;
     /**
      * The error to be dispatched in the event of a fatal error.
      * @const {number}
      * @private
      */
     this.fatalErrorValue_ = fatalErrorValue;
     /**
-     * The async waiter used to wait for
-     * |[receivePipe_]{@link module:data_receiver.DataReceiver#receivePipe_}| to
-     *     be readable.
-     * @type {!module:async_waiter.AsyncWaiter}
-     * @private
-     */
-    this.waiter_ = new asyncWaiter.AsyncWaiter(this.receivePipe_,
-                                               core.HANDLE_SIGNAL_READABLE,
-                                               this.onHandleReady_.bind(this));
-    /**
-     * The number of bytes received from the DataSource.
-     * @type {number}
-     * @private
-     */
-    this.bytesReceived_ = bytesReceived;
-    /**
      * The pending error if there is one.
      * @type {PendingReceiveError}
      * @private
      */
     this.pendingError_ = pendingError;
     /**
      * Whether the DataSource is paused.
      * @type {boolean}
      * @private
      */
     this.paused_ = paused;
     /**
+     * A queue of data that has been received from the DataSource, but not
+     * consumed by the client.
+     * @type {module:data_receiver~PendingData[]}
+     * @private
+     */
+    this.pendingDataBuffers_ = pendingData;
+    /**
      * Whether this DataReceiver has shut down.
      * @type {boolean}
      * @private
      */
     this.shutDown_ = false;
   };
 
   /**
    * Serializes this DataReceiver.
    * This will cancel a receive if one is in progress.
    * @return {!Promise.<SerializedDataReceiver>} A promise that will resolve to
    *     the serialization of this DataReceiver. If this DataReceiver has shut
    *     down, the promise will resolve to null.
    */
   DataReceiver.prototype.serialize = function() {
     if (this.shutDown_)
       return Promise.resolve(null);
 
-    this.waiter_.stop();
     if (this.receive_) {
       this.receive_.dispatchFatalError(this.fatalErrorValue_);
       this.receive_ = null;
     }
     var serialized = new serialization.SerializedDataReceiver();
     serialized.source = this.router_.connector_.handle_;
-    serialized.data_pipe = this.receivePipe_;
     serialized.fatal_error_value = this.fatalErrorValue_;
-    serialized.bytes_received = this.bytesReceived_;
     serialized.paused = this.paused_;
     serialized.pending_error = this.pendingError_;
+    serialized.pending_data = [];
+    $Array.forEach(this.pendingDataBuffers_, function(buffer) {
+      serialized.pending_data.push(new Uint8Array(buffer));
+    });
     this.router_.connector_.handle_ = null;
     this.router_.close();
     this.shutDown_ = true;
     return Promise.resolve(serialized);
   };
 
   /**
    * Deserializes a SerializedDataReceiver.
    * @param {SerializedDataReceiver} serialized The serialized DataReceiver.
    * @return {!DataReceiver} The deserialized DataReceiver.
    */
   DataReceiver.deserialize = function(serialized) {
     var receiver = $Object.create(DataReceiver.prototype);
     receiver.deserialize_(serialized);
     return receiver;
   };
 
   /**
    * Deserializes a SerializedDataReceiver into this DataReceiver.
    * @param {SerializedDataReceiver} serialized The serialized DataReceiver.
    * @private
    */
   DataReceiver.prototype.deserialize_ = function(serialized) {
     if (!serialized) {
       this.shutDown_ = true;
       return;
     }
+    var pendingData = [];
+    $Array.forEach(serialized.pending_data, function(data) {
+      var buffer = new Uint8Array(data.length);
+      buffer.set(data);
+      pendingData.push(buffer.buffer);
+    });
     this.init_(serialized.source,
-               serialized.data_pipe,
                serialized.fatal_error_value,
                serialized.bytes_received,
                serialized.pending_error,
+               pendingData,
                serialized.paused);
   };
 
   /**
    * Receive data from the DataSource.
    * @return {Promise.<ArrayBuffer>} A promise to the received data. If an error
    *     occurs, the promise will reject with an Error object with a property
    *     error containing the error code.
    * @throws Will throw if this has encountered a fatal error or another receive
    *     is in progress.
    */
   DataReceiver.prototype.receive = function() {
     if (this.shutDown_)
       throw new Error('DataReceiver has been closed');
     if (this.receive_)
       throw new Error('Receive already in progress.');
     var receive = new PendingReceive();
     var promise = receive.getPromise();
     if (this.pendingError_ &&
-        receive.dispatchError(this.pendingError_, this.bytesReceived_)) {
+        receive.dispatchError(this.pendingError_)) {
       this.pendingError_ = null;
       this.paused_ = true;
       return promise;
     }
     if (this.paused_) {
       this.source_.resume();
       this.paused_ = false;
     }
     this.receive_ = receive;
-    this.waiter_.start();
+    this.dispatchData_();
     return promise;
   };
 
-  /**
-   * Invoked when
-   * |[receivePipe_]{@link module:data_receiver.DataReceiver#receivePipe_}| is
-   * ready to read. Reads from the data pipe if the wait is successful.
-   * @param {number} waitResult The result of the asynchronous wait.
-   * @private
-   */
-  DataReceiver.prototype.onHandleReady_ = function(waitResult) {
-    if (waitResult != core.RESULT_OK || !this.receive_) {
+  DataReceiver.prototype.dispatchData_ = function() {
+    if (!this.receive_) {
       this.close();
       return;
     }
-    var result = core.readData(this.receivePipe_, core.READ_DATA_FLAG_NONE);
-    if (result.result == core.RESULT_OK) {
-      // TODO(sammc): Handle overflow in the same fashion as the C++ receiver.
-      this.bytesReceived_ += result.buffer.byteLength;
-      this.receive_.dispatchData(result.buffer);
+    if (this.pendingDataBuffers_.length) {
+      this.receive_.dispatchData(this.pendingDataBuffers_[0]);
+      this.source_.reportBytesReceived(this.pendingDataBuffers_[0].byteLength);
       this.receive_ = null;
-    } else if (result.result == core.RESULT_SHOULD_WAIT) {
-      this.waiter_.start();
-    } else {
-      this.close();
+      this.pendingDataBuffers_.shift();
+      if (this.pendingError_)
+        this.pendingError_.queuePosition--;
     }
   };
 
   /**
    * Invoked by the DataSource when an error is encountered.
    * @param {number} offset The location at which the error occurred.
    * @param {number} error The error that occurred.
    * @private
    */
-  DataReceiver.prototype.onError = function(offset, error) {
+  DataReceiver.prototype.onError = function(error) {
     if (this.shutDown_)
       return;
 
     var pendingError = new serialization.PendingReceiveError();
     pendingError.error = error;
-    pendingError.offset = offset;
-    if (this.receive_ &&
-        this.receive_.dispatchError(pendingError, this.bytesReceived_)) {
+    pendingError.queuePosition = this.pendingDataBuffers_.length;
+    if (this.receive_ && this.receive_.dispatchError(pendingError)) {
       this.receive_ = null;
-      this.waiter_.stop();
       this.paused_ = true;
       return;
     }
     this.pendingError_ = pendingError;
   };
 
+  DataReceiver.prototype.onData = function(data) {
+    var buffer = new ArrayBuffer(data.length);
+    var uintView = new Uint8Array(buffer);
+    uintView.set(data);
+    this.pendingDataBuffers_.push(buffer);
+    if (this.receive_)
+      this.dispatchData_();
+  };
+
   return {DataReceiver: DataReceiver};
 });