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) {
  /**
   * @module data_receiver
   */

  /**
   * A pending receive operation.
   * @constructor
   * @alias module:data_receiver~PendingReceive
   * @private
   */
  function PendingReceive() {
    /**
     * The promise that will be resolved or rejected when this receive completes
     * or fails, respectively.
     * @type {!Promise.<ArrayBuffer>}
     * @private
     */
    this.promise_ = new Promise(function(resolve, reject) {
      /**
       * The callback to call with the data received on success.
       * @type {Function}
       * @private
       */
      this.dataCallback_ = 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.<ArrayBuffer>} A promise to the data received.
   */
  PendingReceive.prototype.getPromise = function() {
    return this.promise_;
  };

  /**
   * Dispatches received data to the promise returned by
   * [getPromise]{@link module:data_receiver.PendingReceive#getPromise}.
   * @param {!ArrayBuffer} data The data to dispatch.
   */
  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)
      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} 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);
  }

  DataReceiver.prototype =
      $Object.create(dataStream.DataSourceClientStub.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 {boolean} paused Whether the DataSource is paused.
   * @private
   */
  DataReceiver.prototype.init_ = function(source,
                                          dataPipe,
                                          fatalErrorValue,
                                          bytesReceived,
                                          pendingError,
                                          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.DataSourceProxy(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;
    /**
     * 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_;
    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;
    }
    this.init_(serialized.source,
               serialized.data_pipe,
               serialized.fatal_error_value,
               serialized.bytes_received,
               serialized.pending_error,
               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_)) {
      this.pendingError_ = null;
      this.paused_ = true;
      return promise;
    }
    if (this.paused_) {
      this.source_.resume();
      this.paused_ = false;
    }
    this.receive_ = receive;
    this.waiter_.start();
    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_) {
      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);
      this.receive_ = null;
    } else if (result.result == core.RESULT_SHOULD_WAIT) {
      this.waiter_.start();
    } else {
      this.close();
    }
  };

  /**
   * 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) {
    if (this.shutDown_)
      return;

    var pendingError = new serialization.PendingReceiveError();
    pendingError.error = error;
    pendingError.offset = offset;
    if (this.receive_ &&
        this.receive_.dispatchError(pendingError, this.bytesReceived_)) {
      this.receive_ = null;
      this.waiter_.stop();
      this.paused_ = true;
      return;
    }
    this.pendingError_ = pendingError;
  };

  return {DataReceiver: DataReceiver};
});