[mojo-bindings] Use Watch API instead of MessagePumpMojo

mojo/public/cpp/bindings/lib/connector.cc

Index: mojo/public/cpp/bindings/lib/connector.cc
diff --git a/mojo/public/cpp/bindings/lib/connector.cc b/mojo/public/cpp/bindings/lib/connector.cc
index 7f9a9c30856e1a35fcc25b3e1d28d6007d02221b..812b4c1133f56db708729a82055cbb24e9671456 100644
--- a/mojo/public/cpp/bindings/lib/connector.cc
+++ b/mojo/public/cpp/bindings/lib/connector.cc
@@ -11,6 +11,7 @@
 #include "base/logging.h"
 #include "base/macros.h"
 #include "base/synchronization/lock.h"
+#include "base/thread_task_runner_handle.h"
 #include "mojo/public/cpp/bindings/lib/sync_handle_watcher.h"
 
 namespace mojo {
@@ -247,7 +248,7 @@ bool Connector::RunSyncHandleWatch(const bool* should_stop) {
   return SyncHandleWatcher::current()->WatchAllHandles(should_stop_array, 2);
 }
 
-void Connector::OnHandleWatcherHandleReady(MojoResult result) {
+void Connector::OnWatcherHandleReady(MojoResult result) {
   OnHandleReadyInternal(result);
 }
 
@@ -273,12 +274,21 @@ void Connector::OnHandleReadyInternal(MojoResult result) {
 }
 
 void Connector::WaitToReadMore() {
-  CHECK(!handle_watcher_.is_watching());
   CHECK(!paused_);
-  handle_watcher_.Start(message_pipe_.get(), MOJO_HANDLE_SIGNAL_READABLE,
-                        MOJO_DEADLINE_INDEFINITE,
-                        base::Bind(&Connector::OnHandleWatcherHandleReady,
-                                   base::Unretained(this)));
+  DCHECK(!handle_watcher_.IsWatching());
+
+  MojoResult rv = handle_watcher_.Start(
+      message_pipe_.get(), MOJO_HANDLE_SIGNAL_READABLE,
+      base::Bind(&Connector::OnWatcherHandleReady,
+                 base::Unretained(this)));
+
+  if (rv != MOJO_RESULT_OK) {
+    // If the watch failed because the handle is invalid or its conditions can
+    // no longer be met, we signal the error asynchronously to avoid reentry.
+    base::ThreadTaskRunnerHandle::Get()->PostTask(
+        FROM_HERE, base::Bind(&Connector::OnWatcherHandleReady,
+                              weak_factory_.GetWeakPtr(), rv));
+  }
 
   if (register_sync_handle_watch_count_ > 0 &&
       !registered_with_sync_handle_watcher_) {
@@ -304,13 +314,6 @@ bool Connector::ReadSingleMessage(MojoResult* read_result) {
   *read_result = rv;
 
   if (rv == MOJO_RESULT_OK) {
-    // Dispatching the message may spin in a nested message loop. To ensure we
-    // continue dispatching messages when this happens start listening for
-    // messagse now.
-    if (!handle_watcher_.is_watching()) {
-      // TODO: Need to evaluate the perf impact of this.
-      WaitToReadMore();
-    }
     receiver_result =
         incoming_receiver_ && incoming_receiver_->Accept(&message);
   }
@@ -344,21 +347,13 @@ void Connector::ReadAllAvailableMessages() {
     if (paused_)
       return;
 
-    if (rv == MOJO_RESULT_SHOULD_WAIT) {
-      // ReadSingleMessage could end up calling HandleError which resets
-      // message_pipe_ to a dummy one that is closed. The old EDK will see the
-      // that the peer is closed immediately, while the new one is asynchronous
-      // because of thread hops. In that case, there'll still be an async
-      // waiter.
-      if (!handle_watcher_.is_watching())
-        WaitToReadMore();
+    if (rv == MOJO_RESULT_SHOULD_WAIT)
       break;
-    }
   }
 }
 
 void Connector::CancelWait() {
-  handle_watcher_.Stop();
+  handle_watcher_.Cancel();
 
   if (registered_with_sync_handle_watcher_) {
     SyncHandleWatcher::current()->UnregisterHandle(message_pipe_.get());
@@ -394,8 +389,6 @@ void Connector::HandleError(bool force_pipe_reset, bool force_async_handler) {
   }
 
   if (force_async_handler) {
-    // |dummy_pipe.handle1| has been destructed. Reading the pipe will
-    // eventually cause a read error on |message_pipe_| and set error state.
     if (!paused_)
       WaitToReadMore();
   } else {