Revert of [mojo-bindings] Use Watch API instead of MessagePumpMojo (patchset #10 id:180001 of https://codereview.chromium.org/1759783003/ )

Reason for revert:
Breaks iOS GN which is apparently a tree closer despite having no CQ coverage

Original issue's description:
> [mojo-bindings] Use Watch API instead of MessagePumpMojo
>
> This removes the C++ bindings dependency on MessagePumpMojo,
> consuming the new Watch API instead.
>
> For convenience a new mojo::Watcher is added to the public
> Mojo C++ API library, and this is used by Connector.
>
> BUG=590495
> R=yzshen@chromium.org
> TBR=blundell@chromium.org for rename affecting components/message_port.gypi
> TBR=jam@chromium.org - added a missing header to new url tests
>
> Committed: https://crrev.com/d06373e7cd8b4ad725ed5c64c958f2de13585add
> Cr-Commit-Position: refs/heads/master@{#379402}

TBR=jam@chromium.org,yzshen@chromium.org
# Skipping CQ checks because original CL landed less than 1 days ago.
NOPRESUBMIT=true
NOTREECHECKS=true
NOTRY=true
BUG=590495

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

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

3 files changed, 30 insertions, 22 deletions

diff --git a/mojo/public/cpp/bindings/BUILD.gn b/mojo/public/cpp/bindings/BUILD.gn
index 137d8d3..b172621 100644
--- a/mojo/public/cpp/bindings/BUILD.gn
+++ b/mojo/public/cpp/bindings/BUILD.gn
@@ -101,6 +101,7 @@ source_set("bindings") {
 
   deps = [
     "//base",
+    "//mojo/message_pump",
     "//mojo/public/interfaces/bindings:bindings_cpp_sources",
   ]
 }
diff --git a/mojo/public/cpp/bindings/lib/connector.cc b/mojo/public/cpp/bindings/lib/connector.cc
index 812b4c11..7f9a9c3 100644
--- a/mojo/public/cpp/bindings/lib/connector.cc
+++ b/mojo/public/cpp/bindings/lib/connector.cc
@@ -11,7 +11,6 @@
 #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 {
@@ -248,7 +247,7 @@ bool Connector::RunSyncHandleWatch(const bool* should_stop) {
   return SyncHandleWatcher::current()->WatchAllHandles(should_stop_array, 2);
 }
 
-void Connector::OnWatcherHandleReady(MojoResult result) {
+void Connector::OnHandleWatcherHandleReady(MojoResult result) {
   OnHandleReadyInternal(result);
 }
 
@@ -274,21 +273,12 @@ void Connector::OnHandleReadyInternal(MojoResult result) {
 }
 
 void Connector::WaitToReadMore() {
+  CHECK(!handle_watcher_.is_watching());
   CHECK(!paused_);
-  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));
-  }
+  handle_watcher_.Start(message_pipe_.get(), MOJO_HANDLE_SIGNAL_READABLE,
+                        MOJO_DEADLINE_INDEFINITE,
+                        base::Bind(&Connector::OnHandleWatcherHandleReady,
+                                   base::Unretained(this)));
 
   if (register_sync_handle_watch_count_ > 0 &&
       !registered_with_sync_handle_watcher_) {
@@ -314,6 +304,13 @@ 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);
   }
@@ -347,13 +344,21 @@ void Connector::ReadAllAvailableMessages() {
     if (paused_)
       return;
 
-    if (rv == MOJO_RESULT_SHOULD_WAIT)
+    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();
       break;
+    }
   }
 }
 
 void Connector::CancelWait() {
-  handle_watcher_.Cancel();
+  handle_watcher_.Stop();
 
   if (registered_with_sync_handle_watcher_) {
     SyncHandleWatcher::current()->UnregisterHandle(message_pipe_.get());
@@ -389,6 +394,8 @@ 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 {
diff --git a/mojo/public/cpp/bindings/lib/connector.h b/mojo/public/cpp/bindings/lib/connector.h
index 2989d70..2d2a5cc 100644
--- a/mojo/public/cpp/bindings/lib/connector.h
+++ b/mojo/public/cpp/bindings/lib/connector.h
@@ -9,10 +9,10 @@
 #include "base/memory/scoped_ptr.h"
 #include "base/memory/weak_ptr.h"
 #include "base/threading/thread_checker.h"
+#include "mojo/message_pump/handle_watcher.h"
 #include "mojo/public/cpp/bindings/callback.h"
 #include "mojo/public/cpp/bindings/message.h"
 #include "mojo/public/cpp/system/core.h"
-#include "mojo/public/cpp/system/watcher.h"
 
 namespace base {
 class Lock;
@@ -141,8 +141,8 @@ class Connector : public MessageReceiver {
   }
 
  private:
-  // Callback of mojo::Watcher.
-  void OnWatcherHandleReady(MojoResult result);
+  // Callback of mojo::common::HandleWatcher.
+  void OnHandleWatcherHandleReady(MojoResult result);
   // Callback of SyncHandleWatcher.
   void OnSyncHandleWatcherHandleReady(MojoResult result);
   void OnHandleReadyInternal(MojoResult result);
@@ -169,7 +169,7 @@ class Connector : public MessageReceiver {
   ScopedMessagePipeHandle message_pipe_;
   MessageReceiver* incoming_receiver_;
 
-  Watcher handle_watcher_;
+  common::HandleWatcher handle_watcher_;
 
   bool error_;
   bool drop_writes_;