Mojo C++ bindings: remove usage of MojoAsyncWaiter from mojo/public/cpp/bindings

BUG=585942

Committed: https://crrev.com/cb97b6a175c61a3538d724f1214d537b40df573a
Cr-Commit-Position: refs/heads/master@{#377652}

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

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

17 files changed, 92 insertions, 187 deletions

diff --git a/mojo/mojo_edk_tests.gyp b/mojo/mojo_edk_tests.gyp
index 94ba7233..0353703 100644
--- a/mojo/mojo_edk_tests.gyp
+++ b/mojo/mojo_edk_tests.gyp
@@ -113,6 +113,7 @@
         'mojo_edk.gyp:mojo_run_all_unittests',
         'mojo_public.gyp:mojo_cpp_bindings',
         'mojo_public.gyp:mojo_environment_standalone',
+        'mojo_public.gyp:mojo_message_pump_lib',
         'mojo_public.gyp:mojo_public_bindings_test_utils',
         'mojo_public.gyp:mojo_public_test_interfaces',
         'mojo_public.gyp:mojo_public_test_utils',
diff --git a/mojo/mojo_public.gyp b/mojo/mojo_public.gyp
index 6490657..a02144a 100644
--- a/mojo/mojo_public.gyp
+++ b/mojo/mojo_public.gyp
@@ -187,6 +187,7 @@
       ],
       'dependencies': [
         '../base/base.gyp:base',
+        'mojo_message_pump_lib',
         'mojo_interface_bindings_cpp_sources',
       ],
     },
diff --git a/mojo/public/cpp/bindings/BUILD.gn b/mojo/public/cpp/bindings/BUILD.gn
index 56b11bf..b172621 100644
--- a/mojo/public/cpp/bindings/BUILD.gn
+++ b/mojo/public/cpp/bindings/BUILD.gn
@@ -101,7 +101,7 @@ source_set("bindings") {
 
   deps = [
     "//base",
-    "//mojo/public/cpp/environment",
+    "//mojo/message_pump",
     "//mojo/public/interfaces/bindings:bindings_cpp_sources",
   ]
 }
diff --git a/mojo/public/cpp/bindings/binding.h b/mojo/public/cpp/bindings/binding.h
index 179c1b9..b16638c 100644
--- a/mojo/public/cpp/bindings/binding.h
+++ b/mojo/public/cpp/bindings/binding.h
@@ -8,7 +8,6 @@
 #include <utility>
 
 #include "base/macros.h"
-#include "mojo/public/c/environment/async_waiter.h"
 #include "mojo/public/cpp/bindings/callback.h"
 #include "mojo/public/cpp/bindings/interface_ptr.h"
 #include "mojo/public/cpp/bindings/interface_ptr_info.h"
@@ -49,16 +48,6 @@ class AssociatedGroup;
 //     }
 //   };
 //
-// The caller may specify a |MojoAsyncWaiter| to be used by the connection when
-// waiting for calls to arrive. Normally it is fine to use the default waiter.
-// However, the caller may provide their own implementation if needed. The
-// |Binding| will not take ownership of the waiter, and the waiter must outlive
-// the |Binding|. The provided waiter must be able to signal the implementation
-// which generally means it needs to be able to schedule work on the thread the
-// implementation runs on. If writing library code that has to work on different
-// types of threads callers may need to provide different waiter
-// implementations.
-//
 // This class is thread hostile while bound to a message pipe. All calls to this
 // class must be from the thread that bound it. The interface implementation's
 // methods will be called from the thread that bound this. If a Binding is not
@@ -75,36 +64,25 @@ class Binding {
 
   // Constructs a completed binding of message pipe |handle| to implementation
   // |impl|. Does not take ownership of |impl|, which must outlive the binding.
-  // See class comment for definition of |waiter|.
-  Binding(Interface* impl,
-          ScopedMessagePipeHandle handle,
-          const MojoAsyncWaiter* waiter = Environment::GetDefaultAsyncWaiter())
-      : Binding(impl) {
-    Bind(std::move(handle), waiter);
+  Binding(Interface* impl, ScopedMessagePipeHandle handle) : Binding(impl) {
+    Bind(std::move(handle));
   }
 
   // Constructs a completed binding of |impl| to a new message pipe, passing the
   // client end to |ptr|, which takes ownership of it. The caller is expected to
   // pass |ptr| on to the client of the service. Does not take ownership of any
   // of the parameters. |impl| must outlive the binding. |ptr| only needs to
-  // last until the constructor returns. See class comment for definition of
-  // |waiter|.
-  Binding(Interface* impl,
-          InterfacePtr<Interface>* ptr,
-          const MojoAsyncWaiter* waiter = Environment::GetDefaultAsyncWaiter())
-      : Binding(impl) {
-    Bind(ptr, waiter);
+  // last until the constructor returns.
+  Binding(Interface* impl, InterfacePtr<Interface>* ptr) : Binding(impl) {
+    Bind(ptr);
   }
 
   // Constructs a completed binding of |impl| to the message pipe endpoint in
   // |request|, taking ownership of the endpoint. Does not take ownership of
-  // |impl|, which must outlive the binding. See class comment for definition of
-  // |waiter|.
-  Binding(Interface* impl,
-          InterfaceRequest<GenericInterface> request,
-          const MojoAsyncWaiter* waiter = Environment::GetDefaultAsyncWaiter())
+  // |impl|, which must outlive the binding.
+  Binding(Interface* impl, InterfaceRequest<GenericInterface> request)
       : Binding(impl) {
-    Bind(request.PassMessagePipe(), waiter);
+    Bind(request.PassMessagePipe());
   }
 
   // Tears down the binding, closing the message pipe and leaving the interface
@@ -121,37 +99,28 @@ class Binding {
 
   // Completes a binding that was constructed with only an interface
   // implementation. Takes ownership of |handle| and binds it to the previously
-  // specified implementation. See class comment for definition of |waiter|.
-  void Bind(
-      ScopedMessagePipeHandle handle,
-      const MojoAsyncWaiter* waiter = Environment::GetDefaultAsyncWaiter()) {
-    internal_state_.Bind(std::move(handle), waiter);
+  // specified implementation.
+  void Bind(ScopedMessagePipeHandle handle) {
+    internal_state_.Bind(std::move(handle));
   }
 
   // Completes a binding that was constructed with only an interface
   // implementation by creating a new message pipe, binding one end of it to the
   // previously specified implementation, and passing the other to |ptr|, which
   // takes ownership of it. The caller is expected to pass |ptr| on to the
-  // eventual client of the service. Does not take ownership of |ptr|. See
-  // class comment for definition of |waiter|.
-  void Bind(
-      InterfacePtr<Interface>* ptr,
-      const MojoAsyncWaiter* waiter = Environment::GetDefaultAsyncWaiter()) {
+  // eventual client of the service. Does not take ownership of |ptr|.
+  void Bind(InterfacePtr<Interface>* ptr) {
     MessagePipe pipe;
     ptr->Bind(InterfacePtrInfo<Interface>(std::move(pipe.handle0),
-                                          Interface::Version_),
-              waiter);
-    Bind(std::move(pipe.handle1), waiter);
+                                          Interface::Version_));
+    Bind(std::move(pipe.handle1));
   }
 
   // Completes a binding that was constructed with only an interface
   // implementation by removing the message pipe endpoint from |request| and
-  // binding it to the previously specified implementation. See class comment
-  // for definition of |waiter|.
-  void Bind(
-      InterfaceRequest<GenericInterface> request,
-      const MojoAsyncWaiter* waiter = Environment::GetDefaultAsyncWaiter()) {
-    Bind(request.PassMessagePipe(), waiter);
+  // binding it to the previously specified implementation.
+  void Bind(InterfaceRequest<GenericInterface> request) {
+    Bind(request.PassMessagePipe());
   }
 
   // Whether there are any associated interfaces running on the pipe currently.
diff --git a/mojo/public/cpp/bindings/interface_ptr.h b/mojo/public/cpp/bindings/interface_ptr.h
index 2929a65..1c892fa 100644
--- a/mojo/public/cpp/bindings/interface_ptr.h
+++ b/mojo/public/cpp/bindings/interface_ptr.h
@@ -13,7 +13,6 @@
 #include "mojo/public/cpp/bindings/callback.h"
 #include "mojo/public/cpp/bindings/interface_ptr_info.h"
 #include "mojo/public/cpp/bindings/lib/interface_ptr_state.h"
-#include "mojo/public/cpp/environment/environment.h"
 
 namespace mojo {
 
@@ -66,20 +65,15 @@ class InterfacePtr {
   // Closes the bound message pipe (if any) on destruction.
   ~InterfacePtr() {}
 
-  // Binds the InterfacePtr to a remote implementation of Interface. The
-  // |waiter| is used for receiving notifications when there is data to read
-  // from the message pipe. For most callers, the default |waiter| will be
-  // sufficient.
+  // Binds the InterfacePtr to a remote implementation of Interface.
   //
   // Calling with an invalid |info| (containing an invalid message pipe handle)
   // has the same effect as reset(). In this case, the InterfacePtr is not
   // considered as bound.
-  void Bind(
-      InterfacePtrInfo<GenericInterface> info,
-      const MojoAsyncWaiter* waiter = Environment::GetDefaultAsyncWaiter()) {
+  void Bind(InterfacePtrInfo<GenericInterface> info) {
     reset();
     if (info.is_valid())
-      internal_state_.Bind(std::move(info), waiter);
+      internal_state_.Bind(std::move(info));
   }
 
   // Returns whether or not this InterfacePtr is bound to a message pipe.
@@ -214,15 +208,12 @@ class InterfacePtr {
 };
 
 // If |info| is valid (containing a valid message pipe handle), returns an
-// InterfacePtr bound to it. Otherwise, returns an unbound InterfacePtr. The
-// specified |waiter| will be used as in the InterfacePtr::Bind() method.
+// InterfacePtr bound to it. Otherwise, returns an unbound InterfacePtr.
 template <typename Interface>
-InterfacePtr<Interface> MakeProxy(
-    InterfacePtrInfo<Interface> info,
-    const MojoAsyncWaiter* waiter = Environment::GetDefaultAsyncWaiter()) {
+InterfacePtr<Interface> MakeProxy(InterfacePtrInfo<Interface> info) {
   InterfacePtr<Interface> ptr;
   if (info.is_valid())
-    ptr.Bind(std::move(info), waiter);
+    ptr.Bind(std::move(info));
   return std::move(ptr);
 }
 
diff --git a/mojo/public/cpp/bindings/lib/binding_state.h b/mojo/public/cpp/bindings/lib/binding_state.h
index 6483d94..055efc7 100644
--- a/mojo/public/cpp/bindings/lib/binding_state.h
+++ b/mojo/public/cpp/bindings/lib/binding_state.h
@@ -11,7 +11,6 @@
 #include "base/macros.h"
 #include "base/memory/ref_counted.h"
 #include "base/memory/scoped_ptr.h"
-#include "mojo/public/c/environment/async_waiter.h"
 #include "mojo/public/cpp/bindings/associated_group.h"
 #include "mojo/public/cpp/bindings/callback.h"
 #include "mojo/public/cpp/bindings/interface_ptr.h"
@@ -50,14 +49,14 @@ class BindingState<Interface, false> {
       Close();
   }
 
-  void Bind(ScopedMessagePipeHandle handle, const MojoAsyncWaiter* waiter) {
+  void Bind(ScopedMessagePipeHandle handle) {
     DCHECK(!router_);
     internal::FilterChain filters;
     filters.Append<internal::MessageHeaderValidator>();
     filters.Append<typename Interface::RequestValidator_>();
 
     router_ = new internal::Router(std::move(handle), std::move(filters),
-                                   Interface::HasSyncMethods_, waiter);
+                                   Interface::HasSyncMethods_);
     router_->set_incoming_receiver(&stub_);
     router_->set_connection_error_handler(
         [this]() { connection_error_handler_.Run(); });
@@ -146,10 +145,10 @@ class BindingState<Interface, true> {
       Close();
   }
 
-  void Bind(ScopedMessagePipeHandle handle, const MojoAsyncWaiter* waiter) {
+  void Bind(ScopedMessagePipeHandle handle) {
     DCHECK(!router_);
 
-    router_ = new internal::MultiplexRouter(false, std::move(handle), waiter);
+    router_ = new internal::MultiplexRouter(false, std::move(handle));
     stub_.serialization_context()->router = router_;
 
     endpoint_client_.reset(new internal::InterfaceEndpointClient(
diff --git a/mojo/public/cpp/bindings/lib/connector.cc b/mojo/public/cpp/bindings/lib/connector.cc
index 07c1094..7f9a9c3 100644
--- a/mojo/public/cpp/bindings/lib/connector.cc
+++ b/mojo/public/cpp/bindings/lib/connector.cc
@@ -44,12 +44,9 @@ class MayAutoLock {
 // ----------------------------------------------------------------------------
 
 Connector::Connector(ScopedMessagePipeHandle message_pipe,
-                     ConnectorConfig config,
-                     const MojoAsyncWaiter* waiter)
-    : waiter_(waiter),
-      message_pipe_(std::move(message_pipe)),
+                     ConnectorConfig config)
+    : message_pipe_(std::move(message_pipe)),
       incoming_receiver_(nullptr),
-      async_wait_id_(0),
       error_(false),
       drop_writes_(false),
       enforce_errors_from_incoming_receiver_(true),
@@ -250,12 +247,8 @@ bool Connector::RunSyncHandleWatch(const bool* should_stop) {
   return SyncHandleWatcher::current()->WatchAllHandles(should_stop_array, 2);
 }
 
-// static
-void Connector::CallOnHandleReady(void* closure, MojoResult result) {
-  Connector* self = static_cast<Connector*>(closure);
-  CHECK(self->async_wait_id_ != 0);
-  self->async_wait_id_ = 0;
-  self->OnHandleReadyInternal(result);
+void Connector::OnHandleWatcherHandleReady(MojoResult result) {
+  OnHandleReadyInternal(result);
 }
 
 void Connector::OnSyncHandleWatcherHandleReady(MojoResult result) {
@@ -280,13 +273,12 @@ void Connector::OnHandleReadyInternal(MojoResult result) {
 }
 
 void Connector::WaitToReadMore() {
-  CHECK(!async_wait_id_);
+  CHECK(!handle_watcher_.is_watching());
   CHECK(!paused_);
-  async_wait_id_ = waiter_->AsyncWait(message_pipe_.get().value(),
-                                      MOJO_HANDLE_SIGNAL_READABLE,
-                                      MOJO_DEADLINE_INDEFINITE,
-                                      &Connector::CallOnHandleReady,
-                                      this);
+  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_) {
@@ -315,7 +307,7 @@ bool Connector::ReadSingleMessage(MojoResult* read_result) {
     // 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 (!async_wait_id_) {
+    if (!handle_watcher_.is_watching()) {
       // TODO: Need to evaluate the perf impact of this.
       WaitToReadMore();
     }
@@ -358,7 +350,7 @@ void Connector::ReadAllAvailableMessages() {
       // 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 (!async_wait_id_)
+      if (!handle_watcher_.is_watching())
         WaitToReadMore();
       break;
     }
@@ -366,10 +358,7 @@ void Connector::ReadAllAvailableMessages() {
 }
 
 void Connector::CancelWait() {
-  if (async_wait_id_) {
-    waiter_->CancelWait(async_wait_id_);
-    async_wait_id_ = 0;
-  }
+  handle_watcher_.Stop();
 
   if (registered_with_sync_handle_watcher_) {
     SyncHandleWatcher::current()->UnregisterHandle(message_pipe_.get());
diff --git a/mojo/public/cpp/bindings/lib/connector.h b/mojo/public/cpp/bindings/lib/connector.h
index b0af520..2195be2 100644
--- a/mojo/public/cpp/bindings/lib/connector.h
+++ b/mojo/public/cpp/bindings/lib/connector.h
@@ -9,10 +9,9 @@
 #include "base/memory/scoped_ptr.h"
 #include "base/memory/weak_ptr.h"
 #include "base/threading/thread_checker.h"
-#include "mojo/public/c/environment/async_waiter.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/environment/environment.h"
 #include "mojo/public/cpp/system/core.h"
 
 namespace base {
@@ -42,10 +41,7 @@ class Connector : public MessageReceiver {
   };
 
   // The Connector takes ownership of |message_pipe|.
-  Connector(
-      ScopedMessagePipeHandle message_pipe,
-      ConnectorConfig config,
-      const MojoAsyncWaiter* waiter = Environment::GetDefaultAsyncWaiter());
+  Connector(ScopedMessagePipeHandle message_pipe, ConnectorConfig config);
   ~Connector() override;
 
   // Sets the receiver to handle messages read from the message pipe.  The
@@ -145,7 +141,9 @@ class Connector : public MessageReceiver {
   }
 
  private:
-  static void CallOnHandleReady(void* closure, MojoResult result);
+  // Callback of mojo::common::HandleWatcher.
+  void OnHandleWatcherHandleReady(MojoResult result);
+  // Callback of SyncHandleWatcher.
   void OnSyncHandleWatcherHandleReady(MojoResult result);
   void OnHandleReadyInternal(MojoResult result);
 
@@ -167,12 +165,12 @@ class Connector : public MessageReceiver {
   void CancelWait();
 
   Closure connection_error_handler_;
-  const MojoAsyncWaiter* waiter_;
 
   ScopedMessagePipeHandle message_pipe_;
   MessageReceiver* incoming_receiver_;
 
-  MojoAsyncWaitID async_wait_id_;
+  common::HandleWatcher handle_watcher_;
+
   bool error_;
   bool drop_writes_;
   bool enforce_errors_from_incoming_receiver_;
diff --git a/mojo/public/cpp/bindings/lib/interface_ptr_state.h b/mojo/public/cpp/bindings/lib/interface_ptr_state.h
index b2093bb..4f505d1 100644
--- a/mojo/public/cpp/bindings/lib/interface_ptr_state.h
+++ b/mojo/public/cpp/bindings/lib/interface_ptr_state.h
@@ -25,8 +25,6 @@
 #include "mojo/public/cpp/bindings/lib/router.h"
 #include "mojo/public/cpp/bindings/lib/scoped_interface_endpoint_handle.h"
 
-struct MojoAsyncWaiter;
-
 namespace mojo {
 namespace internal {
 
@@ -42,8 +40,7 @@ class InterfacePtrState<Interface, false> {
  public:
   using GenericInterface = typename Interface::GenericInterface;
 
-  InterfacePtrState()
-      : proxy_(nullptr), router_(nullptr), waiter_(nullptr), version_(0u) {}
+  InterfacePtrState() : proxy_(nullptr), router_(nullptr), version_(0u) {}
 
   ~InterfacePtrState() {
     // Destruction order matters here. We delete |proxy_| first, even though
@@ -94,21 +91,17 @@ class InterfacePtrState<Interface, false> {
     swap(other->proxy_, proxy_);
     swap(other->router_, router_);
     handle_.swap(other->handle_);
-    swap(other->waiter_, waiter_);
     swap(other->version_, version_);
   }
 
-  void Bind(InterfacePtrInfo<GenericInterface> info,
-            const MojoAsyncWaiter* waiter) {
+  void Bind(InterfacePtrInfo<GenericInterface> info) {
     DCHECK(!proxy_);
     DCHECK(!router_);
     DCHECK(!handle_.is_valid());
-    DCHECK(!waiter_);
     DCHECK_EQ(0u, version_);
     DCHECK(info.is_valid());
 
     handle_ = info.PassHandle();
-    waiter_ = waiter;
     version_ = info.version();
   }
 
@@ -163,18 +156,14 @@ class InterfacePtrState<Interface, false> {
       return;
     }
     // The object hasn't been bound.
-    if (!waiter_) {
-      DCHECK(!handle_.is_valid());
+    if (!handle_.is_valid())
       return;
-    }
 
     FilterChain filters;
     filters.Append<MessageHeaderValidator>();
     filters.Append<typename Interface::ResponseValidator_>();
 
-    router_ =
-        new Router(std::move(handle_), std::move(filters), false, waiter_);
-    waiter_ = nullptr;
+    router_ = new Router(std::move(handle_), std::move(filters), false);
 
     proxy_ = new Proxy(router_);
   }
@@ -183,10 +172,9 @@ class InterfacePtrState<Interface, false> {
   Router* router_;
 
   // |proxy_| and |router_| are not initialized until read/write with the
-  // message pipe handle is needed. |handle_| and |waiter_| are valid between
-  // the Bind() call and the initialization of |proxy_| and |router_|.
+  // message pipe handle is needed. |handle_| is valid between the Bind() call
+  // and the initialization of |proxy_| and |router_|.
   ScopedMessagePipeHandle handle_;
-  const MojoAsyncWaiter* waiter_;
 
   uint32_t version_;
 
@@ -200,7 +188,7 @@ class InterfacePtrState<Interface, true> {
  public:
   using GenericInterface = typename Interface::GenericInterface;
 
-  InterfacePtrState() : waiter_(nullptr), version_(0u) {}
+  InterfacePtrState() : version_(0u) {}
 
   ~InterfacePtrState() {
     endpoint_client_.reset();
@@ -252,22 +240,18 @@ class InterfacePtrState<Interface, true> {
     swap(other->endpoint_client_, endpoint_client_);
     swap(other->proxy_, proxy_);
     handle_.swap(other->handle_);
-    swap(other->waiter_, waiter_);
     swap(other->version_, version_);
   }
 
-  void Bind(InterfacePtrInfo<GenericInterface> info,
-            const MojoAsyncWaiter* waiter) {
+  void Bind(InterfacePtrInfo<GenericInterface> info) {
     DCHECK(!router_);
     DCHECK(!endpoint_client_);
     DCHECK(!proxy_);
     DCHECK(!handle_.is_valid());
-    DCHECK(!waiter_);
     DCHECK_EQ(0u, version_);
     DCHECK(info.is_valid());
 
     handle_ = info.PassHandle();
-    waiter_ = waiter;
     version_ = info.version();
   }
 
@@ -330,19 +314,15 @@ class InterfacePtrState<Interface, true> {
       return;
     }
     // The object hasn't been bound.
-    if (!waiter_) {
-      DCHECK(!handle_.is_valid());
+    if (!handle_.is_valid())
       return;
-    }
 
-    router_ = new MultiplexRouter(true, std::move(handle_), waiter_);
+    router_ = new MultiplexRouter(true, std::move(handle_));
     endpoint_client_.reset(new InterfaceEndpointClient(
         router_->CreateLocalEndpointHandle(kMasterInterfaceId), nullptr,
         make_scoped_ptr(new typename Interface::ResponseValidator_())));
     proxy_.reset(new Proxy(endpoint_client_.get()));
     proxy_->serialization_context()->router = endpoint_client_->router();
-
-    waiter_ = nullptr;
   }
 
   scoped_refptr<MultiplexRouter> router_;
@@ -351,10 +331,9 @@ class InterfacePtrState<Interface, true> {
   scoped_ptr<Proxy> proxy_;
 
   // |router_| (as well as other members above) is not initialized until
-  // read/write with the message pipe handle is needed. |handle_| and |waiter_|
-  // are valid between the Bind() call and the initialization of |router_|.
+  // read/write with the message pipe handle is needed. |handle_| is valid
+  // between the Bind() call and the initialization of |router_|.
   ScopedMessagePipeHandle handle_;
-  const MojoAsyncWaiter* waiter_;
 
   uint32_t version_;
 
diff --git a/mojo/public/cpp/bindings/lib/multiplex_router.cc b/mojo/public/cpp/bindings/lib/multiplex_router.cc
index b756202..48d8393 100644
--- a/mojo/public/cpp/bindings/lib/multiplex_router.cc
+++ b/mojo/public/cpp/bindings/lib/multiplex_router.cc
@@ -118,15 +118,12 @@ struct MultiplexRouter::Task {
 };
 
 MultiplexRouter::MultiplexRouter(bool set_interface_id_namesapce_bit,
-                                 ScopedMessagePipeHandle message_pipe,
-                                 const MojoAsyncWaiter* waiter)
-    : RefCountedDeleteOnMessageLoop(base::MessageLoop::current()
-                                        ->task_runner()),
+                                 ScopedMessagePipeHandle message_pipe)
+    : RefCountedDeleteOnMessageLoop(
+          base::MessageLoop::current()->task_runner()),
       set_interface_id_namespace_bit_(set_interface_id_namesapce_bit),
       header_validator_(this),
-      connector_(std::move(message_pipe),
-                 Connector::MULTI_THREADED_SEND,
-                 waiter),
+      connector_(std::move(message_pipe), Connector::MULTI_THREADED_SEND),
       encountered_error_(false),
       control_message_handler_(this),
       control_message_proxy_(&connector_),
diff --git a/mojo/public/cpp/bindings/lib/multiplex_router.h b/mojo/public/cpp/bindings/lib/multiplex_router.h
index 80631ff..d2ce406 100644
--- a/mojo/public/cpp/bindings/lib/multiplex_router.h
+++ b/mojo/public/cpp/bindings/lib/multiplex_router.h
@@ -26,7 +26,6 @@
 #include "mojo/public/cpp/bindings/lib/pipe_control_message_handler_delegate.h"
 #include "mojo/public/cpp/bindings/lib/pipe_control_message_proxy.h"
 #include "mojo/public/cpp/bindings/lib/scoped_interface_endpoint_handle.h"
-#include "mojo/public/cpp/environment/environment.h"
 
 namespace mojo {
 
@@ -52,10 +51,8 @@ class MultiplexRouter
  public:
   // If |set_interface_id_namespace_bit| is true, the interface IDs generated by
   // this router will have the highest bit set.
-  MultiplexRouter(
-      bool set_interface_id_namespace_bit,
-      ScopedMessagePipeHandle message_pipe,
-      const MojoAsyncWaiter* waiter = Environment::GetDefaultAsyncWaiter());
+  MultiplexRouter(bool set_interface_id_namespace_bit,
+                  ScopedMessagePipeHandle message_pipe);
 
   // ---------------------------------------------------------------------------
   // The following public methods are safe to call from any threads.
diff --git a/mojo/public/cpp/bindings/lib/router.cc b/mojo/public/cpp/bindings/lib/router.cc
index bb83477..ddfc0cfa8 100644
--- a/mojo/public/cpp/bindings/lib/router.cc
+++ b/mojo/public/cpp/bindings/lib/router.cc
@@ -85,13 +85,10 @@ bool Router::HandleIncomingMessageThunk::Accept(Message* message) {
 
 Router::Router(ScopedMessagePipeHandle message_pipe,
                FilterChain filters,
-               bool expects_sync_requests,
-               const MojoAsyncWaiter* waiter)
+               bool expects_sync_requests)
     : thunk_(this),
       filters_(std::move(filters)),
-      connector_(std::move(message_pipe),
-                 Connector::SINGLE_THREADED_SEND,
-                 waiter),
+      connector_(std::move(message_pipe), Connector::SINGLE_THREADED_SEND),
       incoming_receiver_(nullptr),
       next_request_id_(0),
       testing_mode_(false),
diff --git a/mojo/public/cpp/bindings/lib/router.h b/mojo/public/cpp/bindings/lib/router.h
index 12aa7e72..2fec5d2 100644
--- a/mojo/public/cpp/bindings/lib/router.h
+++ b/mojo/public/cpp/bindings/lib/router.h
@@ -17,7 +17,6 @@
 #include "mojo/public/cpp/bindings/callback.h"
 #include "mojo/public/cpp/bindings/lib/connector.h"
 #include "mojo/public/cpp/bindings/lib/filter_chain.h"
-#include "mojo/public/cpp/environment/environment.h"
 
 namespace mojo {
 namespace internal {
@@ -26,8 +25,7 @@ class Router : public MessageReceiverWithResponder {
  public:
   Router(ScopedMessagePipeHandle message_pipe,
          FilterChain filters,
-         bool expects_sync_requests,
-         const MojoAsyncWaiter* waiter = Environment::GetDefaultAsyncWaiter());
+         bool expects_sync_requests);
   ~Router() override;
 
   // Sets the receiver to handle messages read from the message pipe that do
diff --git a/mojo/public/cpp/bindings/strong_binding.h b/mojo/public/cpp/bindings/strong_binding.h
index 0b2ce66..e5d6ffd 100644
--- a/mojo/public/cpp/bindings/strong_binding.h
+++ b/mojo/public/cpp/bindings/strong_binding.h
@@ -8,7 +8,6 @@
 #include <assert.h>
 #include <utility>
 
-#include "mojo/public/c/environment/async_waiter.h"
 #include "mojo/public/cpp/bindings/binding.h"
 #include "mojo/public/cpp/bindings/callback.h"
 #include "mojo/public/cpp/bindings/interface_ptr.h"
@@ -54,53 +53,38 @@ class StrongBinding {
  public:
   explicit StrongBinding(Interface* impl) : binding_(impl) {}
 
-  StrongBinding(
-      Interface* impl,
-      ScopedMessagePipeHandle handle,
-      const MojoAsyncWaiter* waiter = Environment::GetDefaultAsyncWaiter())
+  StrongBinding(Interface* impl, ScopedMessagePipeHandle handle)
       : StrongBinding(impl) {
-    Bind(std::move(handle), waiter);
+    Bind(std::move(handle));
   }
 
-  StrongBinding(
-      Interface* impl,
-      InterfacePtr<Interface>* ptr,
-      const MojoAsyncWaiter* waiter = Environment::GetDefaultAsyncWaiter())
+  StrongBinding(Interface* impl, InterfacePtr<Interface>* ptr)
       : StrongBinding(impl) {
-    Bind(ptr, waiter);
+    Bind(ptr);
   }
 
-  StrongBinding(
-      Interface* impl,
-      InterfaceRequest<Interface> request,
-      const MojoAsyncWaiter* waiter = Environment::GetDefaultAsyncWaiter())
+  StrongBinding(Interface* impl, InterfaceRequest<Interface> request)
       : StrongBinding(impl) {
-    Bind(std::move(request), waiter);
+    Bind(std::move(request));
   }
 
   ~StrongBinding() {}
 
-  void Bind(
-      ScopedMessagePipeHandle handle,
-      const MojoAsyncWaiter* waiter = Environment::GetDefaultAsyncWaiter()) {
+  void Bind(ScopedMessagePipeHandle handle) {
     assert(!binding_.is_bound());
-    binding_.Bind(std::move(handle), waiter);
+    binding_.Bind(std::move(handle));
     binding_.set_connection_error_handler([this]() { OnConnectionError(); });
   }
 
-  void Bind(
-      InterfacePtr<Interface>* ptr,
-      const MojoAsyncWaiter* waiter = Environment::GetDefaultAsyncWaiter()) {
+  void Bind(InterfacePtr<Interface>* ptr) {
     assert(!binding_.is_bound());
-    binding_.Bind(ptr, waiter);
+    binding_.Bind(ptr);
     binding_.set_connection_error_handler([this]() { OnConnectionError(); });
   }
 
-  void Bind(
-      InterfaceRequest<Interface> request,
-      const MojoAsyncWaiter* waiter = Environment::GetDefaultAsyncWaiter()) {
+  void Bind(InterfaceRequest<Interface> request) {
     assert(!binding_.is_bound());
-    binding_.Bind(std::move(request), waiter);
+    binding_.Bind(std::move(request));
     binding_.set_connection_error_handler([this]() { OnConnectionError(); });
   }
 
diff --git a/mojo/public/cpp/bindings/tests/BUILD.gn b/mojo/public/cpp/bindings/tests/BUILD.gn
index cb5029e..78853674 100644
--- a/mojo/public/cpp/bindings/tests/BUILD.gn
+++ b/mojo/public/cpp/bindings/tests/BUILD.gn
@@ -77,6 +77,7 @@ source_set("perftests") {
   ]
 
   deps = [
+    "//mojo/message_pump",
     "//mojo/public/cpp/bindings",
     "//mojo/public/cpp/bindings:callback",
     "//mojo/public/cpp/environment:standalone",
diff --git a/mojo/public/cpp/bindings/tests/bindings_perftest.cc b/mojo/public/cpp/bindings/tests/bindings_perftest.cc
index 6844b2a..87337cd 100644
--- a/mojo/public/cpp/bindings/tests/bindings_perftest.cc
+++ b/mojo/public/cpp/bindings/tests/bindings_perftest.cc
@@ -5,6 +5,8 @@
 #include <stddef.h>
 #include <utility>
 
+#include "base/message_loop/message_loop.h"
+#include "mojo/message_pump/message_pump_mojo.h"
 #include "mojo/public/cpp/bindings/binding.h"
 #include "mojo/public/cpp/test_support/test_support.h"
 #include "mojo/public/cpp/test_support/test_utils.h"
@@ -85,9 +87,11 @@ struct BoundPingService {
 };
 
 class MojoBindingsPerftest : public testing::Test {
+ public:
+  MojoBindingsPerftest() : loop_(common::MessagePumpMojo::Create()) {}
+
  protected:
-  Environment env_;
-  RunLoop run_loop_;
+  base::MessageLoop loop_;
 };
 
 TEST_F(MojoBindingsPerftest, InProcessPingPong) {
diff --git a/tools/battor_agent/battor_agent.gyp b/tools/battor_agent/battor_agent.gyp
index 86067ed..d55e1f79 100644
--- a/tools/battor_agent/battor_agent.gyp
+++ b/tools/battor_agent/battor_agent.gyp
@@ -15,7 +15,7 @@
         '../../device/serial/serial.gyp:device_serial',
         '../../device/serial/serial.gyp:device_serial_mojo',
         '../../mojo/mojo_public.gyp:mojo_environment_standalone',
-        '../../mojo/mojo_public.gyp:mojo_public',
+        '../../mojo/mojo_edk.gyp:mojo_system_impl',
       ],
       'sources': [
         'battor_agent_bin.cc',
@@ -56,8 +56,8 @@
         '../../base/base.gyp:test_support_base',
         '../../device/serial/serial.gyp:device_serial',
         '../../device/serial/serial.gyp:device_serial_test_util',
+        '../../mojo/mojo_edk.gyp:mojo_system_impl',
         '../../mojo/mojo_public.gyp:mojo_environment_standalone',
-        '../../mojo/mojo_public.gyp:mojo_public',
       	'../../testing/gmock.gyp:gmock',
         '../../testing/gtest.gyp:gtest',
       ],