Add the JS data pipe client to be used to implement serial send.

extensions/renderer/resources/data_sender.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_sender', [
+    'async_waiter',
+    'device/serial/data_stream.mojom',
+    'mojo/public/js/bindings/core',
+    'mojo/public/js/bindings/router',
+], function(asyncWaiter, dataStreamMojom, core, routerModule) {
+  /**
+   * @module data_sender
+   */
+
+  /**
+   * A pending send operation.
+   * @param {ArrayBuffer} data The data to be sent.
+   * @constructor
+   * @alias module:data_sender~PendingSend
+   * @private
+   */
+  function PendingSend(data) {
+    /**
+     * The remaining data to be sent.
+     * @type {ArrayBuffer}
+     * @private
+     */
+    this.data_ = data;
+    /**
+     * The total length of data to be sent.
+     * @type {number}
+     * @private
+     */
+    this.length_ = data.byteLength;
+    /**
+     * The number of bytes that have been received by the DataSink.
+     * @type {number}
+     * @private
+     */
+    this.bytesSent_ = 0;

[benwells, 2014/08/20 06:13:47]

This name is confusing. It isn't sent bytes, it is bytes that have been received
at the other end. To me this is bytes sent - bytes in transit.

Can you rename it to something communicating that?

[Sam McNally, 2014/08/20 06:30:56]

Done.

+    /**
+     * The promise that will be resolved or rejected when this send completes
+     * or fails, respectively.
+     * @type {Promise.<number>}
+     * @private
+     */
+    this.promise_ = new Promise(function(resolve, reject) {
+      /**
+       * The callback to call on success.
+       * @type {Function}
+       * @private
+       */
+      this.successCallback_ = resolve;
+      /**
+       * The callback to call with the error on failure.
+       * @type {Function}
+       * @private
+       */
+      this.errorCallback_ = reject;
+    }.bind(this));
+  }
+
+  /**
+   * Returns the promise that will be resolved when this operation completes or
+   * rejected if an error occurs.
+   * @return {Promise.<number>} A promise to the number of bytes sent.
+   */
+  PendingSend.prototype.getPromise = function() {
+    return this.promise_;
+  };
+
+  /**
+   * @typedef module:data_sender~PendingSend.ReportBytesResult
+   * @property {number} bytesLeft The number of bytes reported that were not
+   *     sent as part of the send.
+   * @property {boolean?} done Whether this send has completed.
+   * @property {number?} bytesToFlush The number of bytes to flush in the event
+   *     of an error.
+   */
+
+  /**
+   * Invoked when the DataSink reports that bytes have been sent. Resolves the
+   * promise returned by
+   * [getPromise()]{@link module:data_sender~PendingSend#getPromise} once all
+   * bytes have been reported as sent.
+   * @param {number} numBytes The number of bytes sent.
+   * @return {module:data_sender~PendingSend.ReportBytesResult}
+   */
+  PendingSend.prototype.reportBytesSent = function(numBytes) {
+    var result = this.reportBytesSentInternal_(numBytes);
+    if (this.bytesSent_ == this.length_) {
+      result.done = true;
+      this.successCallback_(this.bytesSent_);
+    }
+    return result;
+  };
+
+  /**
+   * Invoked when the DataSink reports an error. Rejects the promise returned by
+   * [getPromise()]{@link module:data_sender~PendingSend#getPromise} unless the
+   * error occurred after this send, that is, unless numBytes is greater than
+   * the nubmer of outstanding bytes.
+   * @param {number} numBytes The number of bytes sent.
+   * @param {number} error The error reported by the DataSink.
+   * @return {module:data_sender~PendingSend.ReportBytesResult}
+   */
+  PendingSend.prototype.reportBytesSentAndError = function(numBytes, error) {
+    var result = this.reportBytesSentInternal_(numBytes);
+    // If there are remaining bytes to report, the error occurred after this
+    // PendingSend so we should report success.
+    if (result.bytesLeft > 0) {
+      this.successCallback_(this.bytesSent_);
+      return result;
+    }
+
+    var e = new Error();
+    e.error = error;
+    e.bytesSent = this.bytesSent_;
+    this.errorCallback_(e);
+    this.done = true;
+    result.bytesToFlush =
+        this.length_ - this.data_.byteLength - this.bytesSent_;
+    return result;
+  };
+
+  /**
+   * Updates the internal state in response to a report from the DataSink.
+   * @param {number} numBytes The number of bytes sent.
+   * @return {module:data_sender~PendingSend.ReportBytesResult}
+   * @private
+   */
+  PendingSend.prototype.reportBytesSentInternal_ = function(numBytes) {
+    this.bytesSent_ += numBytes;
+    var result = {bytesLeft: 0, bytesToFlush: 0};

[benwells, 2014/08/20 06:13:47]

Nit: bytesToFlush isn't used in the normal (non-error) flow at all. Consider
removing it from this object and adding it in reportBytesSentAndError.

[Sam McNally, 2014/08/20 06:30:56]

Done.

+    if (this.bytesSent_ > this.length_) {
+      result.bytesLeft = this.bytesSent_ - this.length_;
+      this.bytesSent_ = this.length_;
+    }
+    result.done = false;
+    return result;
+  };
+
+  /**
+   * Writes pending data into the data pipe.
+   * @param {MojoHandle} handle The handle to the data pipe.
+   * @return {number} The Mojo result corresponding to the outcome:
+   *     <ul>
+   *     <li>RESULT_OK if the write completes successfully;
+   *     <li>RESULT_SHOULD_WAIT if some, but not all data was written; or
+   *     <li>the data pipe error if the write failed.
+   *     </ul>
+   */
+  PendingSend.prototype.sendData = function(handle) {
+    var result = core.writeData(
+        handle, new Int8Array(this.data_), core.WRITE_DATA_FLAG_NONE);
+    if (result.result != core.RESULT_OK)
+      return result.result;
+    this.data_ = this.data_.slice(result.numBytes);
+    return this.data_.byteLength ? core.RESULT_SHOULD_WAIT : core.RESULT_OK;
+  };
+
+  /**
+   * A DataSender that sends data to a DataSink.
+   * @param {MojoHandle} handle The handle to the DataSink.
+   * @param {number} bufferSize How large a buffer the data pipe should use.
+   * @param {number} fatalErrorValue The send error value to report in the
+   *     event of a fatal error.
+   * @constructor
+   * @alias module:data_sender.DataSender
+   */
+  function DataSender(handle, bufferSize, fatalErrorValue) {
+    /**
+     * The [Router]{@link module:mojo/public/js/bindings/router.Router} for the
+     * connection to the DataSink.
+     * @private
+     */
+    this.router_ = new routerModule.Router(handle);
+    /**
+     * The connection to the DataSink.
+     * @private
+     */
+    this.sink_ = new dataStreamMojom.DataSinkProxy(this.router_);
+    this.router_.setIncomingReceiver(this);
+    var dataPipeOptions = {
+      flags: core.CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,
+      elementNumBytes: 1,
+      capacityNumBytes: bufferSize,
+    };
+    var sendPipe = core.createDataPipe(dataPipeOptions);
+    this.sink_.init(sendPipe.consumerHandle);
+    /**
+     * The handle to the data pipe to use for sending data.
+     * @private
+     */
+    this.sendPipe_ = sendPipe.producerHandle;
+    /**
+     * The error to be dispatched in the event of a fatal error.
+     * @type {number}
+     * @private
+     */
+    this.fatalErrorValue_ = fatalErrorValue;
+    /**
+     * The async waiter used to wait for
+     * {@link module:data_sender.DataSender#sendPipe_} to be writable.
+     * @type module:async_waiter.AsyncWaiter
+     * @private
+     */
+    this.waiter_ = new asyncWaiter.AsyncWaiter(
+        this.sendPipe_, core.HANDLE_SIGNAL_WRITABLE,
+        this.onHandleReady_.bind(this));
+    /**
+     * A queue of sends that have not fully written their data to the data pipe.
+     * @type module:data_sender~PendingSend[]
+     * @private
+     */
+    this.pendingSends_ = [];
+    /**
+     * A queue of sends that have written their data to the data pipe, but have
+     * not been received by the DataSink.
+     * @type module:data_sender~PendingSend[]
+     * @private
+     */
+    this.sendsAwaitingAck_ = [];
+    /**
+     * The callback that will resolve a pending cancel if one is in progress.
+     * @type Function
+     * @private
+     */
+    this.pendingCancel_ = null;
+    /**
+     * Whether this DataReceiver has shut down.
+     * @type {boolean}
+     * @private
+     */
+    this.shutDown_ = false;
+  }
+
+  DataSender.prototype =
+      $Object.create(dataStreamMojom.DataSinkClientStub.prototype);
+
+  /**
+   * Closes this DataSender.
+   */
+  DataSender.prototype.close = function() {
+    if (this.shutDown_)
+      return;
+    this.shutDown_ = true;
+    this.waiter_.stop();
+    this.router_.close();
+    core.close(this.sendPipe_);
+    while (this.pendingSends_.length) {
+      this.pendingSends_.pop().reportBytesSentAndError(
+          0, this.fatalErrorValue_);
+    }
+    while (this.sendsAwaitingAck_.length) {
+      this.sendsAwaitingAck_.pop().reportBytesSentAndError(
+          0, this.fatalErrorValue_);
+    }
+    if (this.pendingCancel_) {
+      this.pendingCancel_();
+      this.pendingCancel_ = null;
+    }
+  };
+
+  /**
+   * Sends data to the DataSink.
+   * @return {Promise.<number>} A promise to the number of bytes sent. 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 a cancel is in
+   *     progress.
+   */
+  DataSender.prototype.send = function(data) {
+    if (this.shutDown_)
+      throw new Error('DataSender has been closed');
+    if (this.pendingCancel_)
+      throw new Error('Cancel in progress');
+    var send = new PendingSend(data);
+    this.pendingSends_.push(send);
+    if (!this.waiter_.isWaiting())
+      this.waiter_.start();
+    return send.getPromise();
+  };
+
+  /**
+   * Requests the cancellation of any in-progress sends. Calls to
+   * [send()]{@link module:data_sender.DataSender#send} will fail until the
+   * cancel has completed.
+   * @param {number} error The error to report for cancelled sends.
+   * @return {Promise} A promise that will resolve when the cancel completes.
+   * @throws Will throw if this has encountered a fatal error or another cancel
+   *     is in progress.
+   */
+  DataSender.prototype.cancel = function(error) {
+    if (this.shutDown_)
+      throw new Error('DataSender has been closed');
+    if (this.pendingCancel_)
+      throw new Error('Cancel already in progress');
+    if (this.pendingSends_.length + this.sendsAwaitingAck_.length == 0)
+      return Promise.resolve();
+
+    this.sink_.cancel(error);
+    return new Promise(function(resolve) {
+      this.pendingCancel_ = resolve;
+    }.bind(this));
+  };
+
+  /**
+   * Invoked when |handle_| is ready to write. Writes to the data pipe if the
+   * wait is successful.
+   * @param {number} waitResult The result of the asynchronous wait.
+   * @private
+   */
+  DataSender.prototype.onHandleReady_ = function(result) {
+    if (result != core.RESULT_OK) {
+      this.close();
+      return;
+    }
+    while (this.pendingSends_.length) {
+      var result = this.pendingSends_[0].sendData(this.sendPipe_);
+      if (result == core.RESULT_OK) {
+        this.sendsAwaitingAck_.push(this.pendingSends_.shift());
+      } else if (result == core.RESULT_SHOULD_WAIT) {
+        this.waiter_.start();
+        return;
+      } else {
+        this.close();
+        return;
+      }
+    }
+  };
+
+  /**
+   * Calls and clears the pending cancel callback if one is pending.
+   * @private
+   */
+  DataSender.prototype.callCancelCallback_ = function() {
+    if (this.pendingCancel_) {
+      this.pendingCancel_();
+      this.pendingCancel_ = null;
+    }
+  };
+
+  /**
+   * Invoked by the DataSink to report that data has been successfully sent.
+   * @param {number} numBytes The number of bytes sent.
+   * @private
+   */
+  DataSender.prototype.reportBytesSent = function(numBytes) {
+    while (numBytes > 0 && this.sendsAwaitingAck_.length) {
+      var result = this.sendsAwaitingAck_[0].reportBytesSent(numBytes);
+      numBytes = result.bytesLeft;

[benwells, 2014/08/20 06:13:47]

*Now* I understand bytesLeft. In the context of PendingSend it is confusing - I
thought it was bytes left to be sent by that PendingSend. Maybe bytesUnreported
would be a better name.

Anyway - up to you.

[Sam McNally, 2014/08/20 06:30:56]

Done.

+      if (result.done)
+        this.sendsAwaitingAck_.shift();
+    }
+    if (numBytes > 0 && this.pendingSends_.length) {
+      var result = this.pendingSends_[0].reportBytesSent(numBytes);
+      numBytes = result.bytesLeft;
+    }
+    // A cancel is completed when all of the sends that were in progress have
+    // completed or failed. This is the case where all sends complete
+    // successfully.
+    if (this.pendingSends_.length + this.sendsAwaitingAck_.length == 0)
+      this.callCancelCallback_();
+  };
+
+  /**
+   * Invoked by the DataSink to report an error in sending data.
+   * @param {number} numBytes The number of bytes sent.
+   * @param {number} error The error reported by the DataSink.
+   * @private
+   */
+  DataSender.prototype.reportBytesSentAndError = function(numBytes, error) {
+    var bytesToFlush = 0;
+    while (this.sendsAwaitingAck_.length) {
+      var result = this.sendsAwaitingAck_[0].reportBytesSentAndError(
+          numBytes, error);
+      numBytes = result.bytesLeft;
+      this.sendsAwaitingAck_.shift();
+      bytesToFlush += result.bytesToFlush;
+    }
+    while (this.pendingSends_.length) {
+      var result = this.pendingSends_[0].reportBytesSentAndError(
+          numBytes, error);
+      numBytes = result.bytesLeft;
+      this.pendingSends_.shift();
+      // Only the first PendingSend in |pendingSends_| will have data to flush

[benwells, 2014/08/20 06:13:47]

Nit: this comment is informative but doesn't seem to relate to the code around
it - the first pendingSend isn't treated specially as far as I can see. Maybe
preface it with 'Note:' so people trying to understand the code won't get
confused but it?

[Sam McNally, 2014/08/20 06:30:56]

Done.

+      // as only the first can have written data to the data pipe.
+      bytesToFlush += result.bytesToFlush;
+    }
+    this.callCancelCallback_();
+    return Promise.resolve({bytes_to_flush: bytesToFlush});
+  };
+
+  return {DataSender: DataSender};
+});