Implement the client side of Serial I/O on data pipe.

This change implements chrome.serial.send and the
chrome.serial.onReceive(Error) events using DataSender and DataReceiver,
respectively. It also adds a TimeoutManager to allow control over when
callbacks queued by setTimeout are invoked to facilitate testing of send
and receive timeouts.

BUG=389016

Review URL: https://codereview.chromium.org/509813002

Cr-Commit-Position: refs/heads/master@{#293092}

1 files changed, 160 insertions, 20 deletions

diff --git a/extensions/renderer/resources/serial_service.js b/extensions/renderer/resources/serial_service.js
index b8b0970..60b0afd 100644
--- a/extensions/renderer/resources/serial_service.js
+++ b/extensions/renderer/resources/serial_service.js
@@ -4,10 +4,17 @@
 
 define('serial_service', [
     'content/public/renderer/service_provider',
+    'data_receiver',
+    'data_sender',
     'device/serial/serial.mojom',
     'mojo/public/js/bindings/core',
     'mojo/public/js/bindings/router',
-], function(serviceProvider, serialMojom, core, routerModule) {
+], function(serviceProvider,
+            dataReceiver,
+            dataSender,
+            serialMojom,
+            core,
+            routerModule) {
   /**
    * A Javascript client for the serial service and connection Mojo services.
    *
@@ -59,6 +66,20 @@ define('serial_service', [
     'odd': serialMojom.ParityBit.ODD,
     'even': serialMojom.ParityBit.EVEN,
   };
+  var SEND_ERROR_TO_MOJO = {
+    undefined: serialMojom.SendError.NONE,
+    'disconnected': serialMojom.SendError.DISCONNECTED,
+    'pending': serialMojom.SendError.PENDING,
+    'timeout': serialMojom.SendError.TIMEOUT,
+    'system_error': serialMojom.SendError.SYSTEM_ERROR,
+  };
+  var RECEIVE_ERROR_TO_MOJO = {
+    undefined: serialMojom.ReceiveError.NONE,
+    'disconnected': serialMojom.ReceiveError.DISCONNECTED,
+    'device_lost': serialMojom.ReceiveError.DEVICE_LOST,
+    'timeout': serialMojom.ReceiveError.TIMEOUT,
+    'system_error': serialMojom.ReceiveError.SYSTEM_ERROR,
+  };
 
   function invertMap(input) {
     var output = {};
@@ -73,6 +94,8 @@ define('serial_service', [
   var DATA_BITS_FROM_MOJO = invertMap(DATA_BITS_TO_MOJO);
   var STOP_BITS_FROM_MOJO = invertMap(STOP_BITS_TO_MOJO);
   var PARITY_BIT_FROM_MOJO = invertMap(PARITY_BIT_TO_MOJO);
+  var SEND_ERROR_FROM_MOJO = invertMap(SEND_ERROR_TO_MOJO);
+  var RECEIVE_ERROR_FROM_MOJO = invertMap(RECEIVE_ERROR_TO_MOJO);
 
   function getServiceOptions(options) {
     var out = {};
@@ -103,29 +126,44 @@ define('serial_service', [
     };
   }
 
-  function Connection(remoteConnection, router, id, options) {
+  function Connection(
+      remoteConnection, router, receivePipe, sendPipe, id, options) {
     this.remoteConnection_ = remoteConnection;
     this.router_ = router;
-    this.id_ = id;
-    getConnections().then(function(connections) {
-      connections[this.id_] = this;
-    }.bind(this));
-    this.paused_ = false;
     this.options_ = {};
     for (var key in DEFAULT_CLIENT_OPTIONS) {
       this.options_[key] = DEFAULT_CLIENT_OPTIONS[key];
     }
     this.setClientOptions_(options);
+    this.receivePipe_ =
+        new dataReceiver.DataReceiver(receivePipe,
+                                      this.options_.bufferSize,
+                                      serialMojom.ReceiveError.DISCONNECTED);
+    this.sendPipe_ = new dataSender.DataSender(
+        sendPipe, this.options_.bufferSize, serialMojom.SendError.DISCONNECTED);
+    this.id_ = id;
+    getConnections().then(function(connections) {
+      connections[this.id_] = this;
+    }.bind(this));
+    this.paused_ = false;
+    this.sendInProgress_ = false;
+
+    // queuedReceiveData_ or queuedReceiveError will store the receive result or
+    // error, respectively, if a receive completes or fails while this
+    // connection is paused. At most one of the the two may be non-null: a
+    // receive completed while paused will only set one of them, no further
+    // receives will be performed while paused and a queued result is dispatched
+    // before any further receives are initiated when unpausing.
+    this.queuedReceiveData_ = null;
+    this.queuedReceiveError = null;
+
+    this.startReceive_();
   }
 
   Connection.create = function(path, options) {
     options = options || {};
     var serviceOptions = getServiceOptions(options);
     var pipe = core.createMessagePipe();
-    // Note: These two are created and closed because the service implementation
-    // requires that we provide valid message pipes for the data source and
-    // sink. Currently the client handles are immediately closed; the real
-    // implementation will come later.
     var sendPipe = core.createMessagePipe();
     var receivePipe = core.createMessagePipe();
     service.connect(path,
@@ -133,21 +171,24 @@ define('serial_service', [
                     pipe.handle0,
                     sendPipe.handle0,
                     receivePipe.handle0);
-    core.close(sendPipe.handle1);
-    core.close(receivePipe.handle1);
     var router = new routerModule.Router(pipe.handle1);
     var connection = new serialMojom.ConnectionProxy(router);
-    return connection.getInfo().then(convertServiceInfo).then(
-        function(info) {
+    return connection.getInfo().then(convertServiceInfo).then(function(info) {
       return Promise.all([info, allocateConnectionId()]);
     }).catch(function(e) {
       router.close();
+      core.close(sendPipe.handle1);
+      core.close(receivePipe.handle1);
       throw e;
     }).then(function(results) {
       var info = results[0];
       var id = results[1];
-      var serialConnectionClient = new Connection(
-          connection, router, id, options);
+      var serialConnectionClient = new Connection(connection,
+                                                  router,
+                                                  receivePipe.handle1,
+                                                  sendPipe.handle1,
+                                                  id,
+                                                  options);
       var clientInfo = serialConnectionClient.getClientInfo_();
       for (var key in clientInfo) {
         info[key] = clientInfo[key];
@@ -161,8 +202,12 @@ define('serial_service', [
 
   Connection.prototype.close = function() {
     this.router_.close();
+    this.receivePipe_.close();
+    this.sendPipe_.close();
+    clearTimeout(this.receiveTimeoutId_);
+    clearTimeout(this.sendTimeoutId_);
     return getConnections().then(function(connections) {
-      delete connections[this.id_]
+      delete connections[this.id_];
       return true;
     }.bind(this));
   };
@@ -171,7 +216,7 @@ define('serial_service', [
     var info = {
       connectionId: this.id_,
       paused: this.paused_,
-    }
+    };
     for (var key in this.options_) {
       info[key] = this.options_[key];
     }
@@ -253,6 +298,99 @@ define('serial_service', [
 
   Connection.prototype.setPaused = function(paused) {
     this.paused_ = paused;
+    if (paused) {
+      clearTimeout(this.receiveTimeoutId_);
+      this.receiveTimeoutId_ = null;
+    } else if (!this.receiveInProgress_) {
+      this.startReceive_();
+    }
+  };
+
+  Connection.prototype.send = function(data) {
+    if (this.sendInProgress_)
+      return Promise.resolve({bytesSent: 0, error: 'pending'});
+
+    if (this.options_.sendTimeout) {
+      this.sendTimeoutId_ = setTimeout(function() {
+        this.sendPipe_.cancel(serialMojom.SendError.TIMEOUT);
+      }.bind(this), this.options_.sendTimeout);
+    }
+    this.sendInProgress_ = true;
+    return this.sendPipe_.send(data).then(function(bytesSent) {
+      return {bytesSent: bytesSent};
+    }).catch(function(e) {
+      return {
+        bytesSent: e.bytesSent,
+        error: SEND_ERROR_FROM_MOJO[e.error],
+      };
+    }).then(function(result) {
+      if (this.sendTimeoutId_)
+        clearTimeout(this.sendTimeoutId_);
+      this.sendTimeoutId_ = null;
+      this.sendInProgress_ = false;
+      return result;
+    }.bind(this));
+  };
+
+  Connection.prototype.startReceive_ = function() {
+    this.receiveInProgress_ = true;
+    var receivePromise = null;
+    // If we have a queued receive result, dispatch it immediately instead of
+    // starting a new receive.
+    if (this.queuedReceiveData_) {
+      receivePromise = Promise.resolve(this.queuedReceiveData_);
+      this.queuedReceiveData_ = null;
+    } else if (this.queuedReceiveError) {
+      receivePromise = Promise.reject(this.queuedReceiveError);
+      this.queuedReceiveError = null;
+    } else {
+      receivePromise = this.receivePipe_.receive();
+    }
+    receivePromise.then(this.onDataReceived_.bind(this)).catch(
+        this.onReceiveError_.bind(this));
+    this.startReceiveTimeoutTimer_();
+  };
+
+  Connection.prototype.onDataReceived_ = function(data) {
+    this.startReceiveTimeoutTimer_();
+    this.receiveInProgress_ = false;
+    if (this.paused_) {
+      this.queuedReceiveData_ = data;
+      return;
+    }
+    if (this.onData) {
+      this.onData(data);
+    }
+    if (!this.paused_) {
+      this.startReceive_();
+    }
+  };
+
+  Connection.prototype.onReceiveError_ = function(e) {
+    clearTimeout(this.receiveTimeoutId_);
+    this.receiveInProgress_ = false;
+    if (this.paused_) {
+      this.queuedReceiveError = e;
+      return;
+    }
+    var error = e.error;
+    this.paused_ = true;
+    if (this.onError)
+      this.onError(RECEIVE_ERROR_FROM_MOJO[error]);
+  };
+
+  Connection.prototype.startReceiveTimeoutTimer_ = function() {
+    clearTimeout(this.receiveTimeoutId_);
+    if (this.options_.receiveTimeout && !this.paused_) {
+      this.receiveTimeoutId_ = setTimeout(this.onReceiveTimeout_.bind(this),
+                                          this.options_.receiveTimeout);
+    }
+  };
+
+  Connection.prototype.onReceiveTimeout_ = function() {
+    if (this.onError)
+      this.onError('timeout');
+    this.startReceiveTimeoutTimer_();
   };
 
   var connections_ = {};
@@ -268,7 +406,7 @@ define('serial_service', [
   function getConnection(id) {
     return getConnections().then(function(connections) {
       if (!connections[id])
-        throw new Error ('Serial connection not found.');
+        throw new Error('Serial connection not found.');
       return connections[id];
     });
   }
@@ -282,5 +420,7 @@ define('serial_service', [
     createConnection: Connection.create,
     getConnection: getConnection,
     getConnections: getConnections,
+    // For testing.
+    Connection: Connection,
   };
 });