Handle error in connector and close handle and swap with dummy handle.

mojo/public/js/connector.js

Index: mojo/public/js/connector.js
diff --git a/mojo/public/js/connector.js b/mojo/public/js/connector.js
index d51b429d7d012e03283a43baf7fd34fc7481c39c..662c65dae1f518e4909d42a1817fc5d79f5ddfd6 100644
--- a/mojo/public/js/connector.js
+++ b/mojo/public/js/connector.js
@@ -7,7 +7,8 @@ define("mojo/public/js/connector", [
   "mojo/public/js/codec",
   "mojo/public/js/core",
   "mojo/public/js/support",
-], function(buffer, codec, core, support) {
+  "mojo/public/js/validator",
+], function(buffer, codec, core, support, validator) {
 
   function Connector(handle) {
     if (!core.isHandle(handle))
@@ -20,18 +21,11 @@ define("mojo/public/js/connector", [
     this.errorHandler_ = null;
     this.paused_ = false;
 
-    if (handle) {
-      this.readWatcher_ = support.watch(handle,
-                                        core.HANDLE_SIGNAL_READABLE,
-                                        this.readMore_.bind(this));
-    }
+    this.waitToReadMore();
   }
 
   Connector.prototype.close = function() {
-    if (this.readWatcher_) {
-      support.cancelWatch(this.readWatcher_);
-      this.readWatcher_ = null;
-    }
+    this.cancelWait();
     if (this.handle_ != null) {
       core.close(this.handle_);
       this.handle_ = null;
@@ -43,11 +37,7 @@ define("mojo/public/js/connector", [
       return;
     }
     this.paused_= true;
-
-    if (this.readWatcher_) {
-      support.cancelWatch(this.readWatcher_);
-      this.readWatcher_ = null;
-    }
+    this.cancelWait();
   };
 
   Connector.prototype.resumeIncomingMethodCallProcessing = function() {
@@ -55,12 +45,7 @@ define("mojo/public/js/connector", [
       return;
     }
     this.paused_= false;
-
-    if (this.handle_) {
-      this.readWatcher_ = support.watch(this.handle_,
-                                        core.HANDLE_SIGNAL_READABLE,
-                                        this.readMore_.bind(this));
-    }
+    this.waitToReadMore();
   };
 
   Connector.prototype.accept = function(message) {
@@ -122,18 +107,71 @@ define("mojo/public/js/connector", [
       if (read.result == core.RESULT_SHOULD_WAIT)
         return;
       if (read.result != core.RESULT_OK) {
-        // TODO(wangjimmy): Add a handleError method to swap the handle to be
-        // closed with a dummy handle in the case when
-        // read.result != MOJO_RESULT_FAILED_PRECONDITION
-        this.error_ = true;
-        if (this.errorHandler_)
-          this.errorHandler_.onError();
+        this.handleError(read.result !== core.RESULT_FAILED_PRECONDITION,
+            false);
         return;
       }
       var messageBuffer = new buffer.Buffer(read.buffer);
       var message = new codec.Message(messageBuffer, read.handles);
-      if (this.incomingReceiver_)
-        this.incomingReceiver_.accept(message);
+      var receiverResult = this.incomingReceiver_ &&
+          this.incomingReceiver_.accept(message);
+
+      // Handle invalid incoming message.
+      if (!validator.isTestingMode() && !receiverResult) {
+        // TODO(yzshen): Consider notifying the embedder.
+        this.handleError(true, false);
+      }
+    }
+  };
+
+  Connector.prototype.cancelWait = function() {
+    if (this.readWatcher_) {
+      support.cancelWatch(this.readWatcher_);
+      this.readWatcher_ = null;
+    }
+  };
+
+  Connector.prototype.waitToReadMore = function() {
+    if (this.handle_) {
+      this.readWatcher_ = support.watch(this.handle_,
+                                        core.HANDLE_SIGNAL_READABLE,
+                                        this.readMore_.bind(this));
+    }
+  };
+
+  Connector.prototype.handleError = function(forcePipeReset,
+                                             forceAsyncHandler) {
+    if (this.error_ || this.handle_ === null) {
+      return;
+    }
+
+    if (this.paused_) {
+      // Enforce calling the error handler asynchronously if the user has
+      // paused receiving messages. We need to wait until the user starts
+      // receiving messages again.
+      forceAsyncHandler = true;
+    }
+
+    if (!forcePipeReset && forceAsyncHandler) {
+      forcePipeReset = true;
+    }
+
+    this.cancelWait();
+    if (forcePipeReset) {
+      core.close(this.handle_);
+      var dummyPipe = core.createMessagePipe();
+      this.handle_ = dummyPipe.handle0;
+    }
+
+    if (forceAsyncHandler) {
+      if (!this.paused_) {
+        this.waitToReadMore();
+      }
+    } else {
+      this.error_ = true;
+      if (this.errorHandler_) {
+        this.errorHandler_.onError();
+      }
     }
   };