Support early associated interface binding on ChannelMojo

ipc/ipc_mojo_bootstrap.cc

 // Copyright 2014 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "ipc/ipc_mojo_bootstrap.h"
 
 #include <stdint.h>
 
 #include <map>
 #include <memory>
 #include <utility>
+#include <vector>
 
 #include "base/callback.h"
 #include "base/logging.h"
 #include "base/macros.h"
 #include "base/memory/ptr_util.h"
-#include "base/process/process_handle.h"
 #include "base/single_thread_task_runner.h"
 #include "base/stl_util.h"
 #include "base/synchronization/lock.h"
 #include "base/threading/thread_task_runner_handle.h"
 #include "build/build_config.h"
 #include "ipc/ipc_message_utils.h"
 #include "ipc/ipc_platform_file.h"
 #include "mojo/public/cpp/bindings/associated_group.h"
 #include "mojo/public/cpp/bindings/associated_group_controller.h"
 #include "mojo/public/cpp/bindings/binding.h"
 #include "mojo/public/cpp/bindings/connector.h"
 #include "mojo/public/cpp/bindings/interface_endpoint_client.h"
 #include "mojo/public/cpp/bindings/interface_endpoint_controller.h"
 #include "mojo/public/cpp/bindings/interface_id.h"
+#include "mojo/public/cpp/bindings/message.h"
 #include "mojo/public/cpp/bindings/message_header_validator.h"
 #include "mojo/public/cpp/bindings/pipe_control_message_handler.h"
 #include "mojo/public/cpp/bindings/pipe_control_message_handler_delegate.h"
 #include "mojo/public/cpp/bindings/pipe_control_message_proxy.h"
 
 namespace IPC {
 
 namespace {
 
 class ChannelAssociatedGroupController
     : public mojo::AssociatedGroupController,
       public mojo::MessageReceiver,
       public mojo::PipeControlMessageHandlerDelegate {
  public:
-  ChannelAssociatedGroupController(bool set_interface_id_namespace_bit,
-                                   mojo::ScopedMessagePipeHandle handle)
+  ChannelAssociatedGroupController(
+      bool set_interface_id_namespace_bit,
+      const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner)
       : mojo::AssociatedGroupController(base::ThreadTaskRunnerHandle::Get()),
-        task_runner_(base::ThreadTaskRunnerHandle::Get()),
-        id_namespace_mask_(set_interface_id_namespace_bit ?
-              mojo::kInterfaceIdNamespaceMask : 0),
-        associated_group_(CreateAssociatedGroup()),
-        connector_(std::move(handle), mojo::Connector::SINGLE_THREADED_SEND,
-                   base::ThreadTaskRunnerHandle::Get()),
+        task_runner_(ipc_task_runner),
+        proxy_task_runner_(base::ThreadTaskRunnerHandle::Get()),
+        set_interface_id_namespace_bit_(set_interface_id_namespace_bit),
         header_validator_(
             "IPC::mojom::Bootstrap [master] MessageHeaderValidator", this),
         control_message_handler_(this),
-        control_message_proxy_(&connector_) {
-    connector_.set_incoming_receiver(&header_validator_);
-    connector_.set_connection_error_handler(
-        base::Bind(&ChannelAssociatedGroupController::OnPipeError,
-                   base::Unretained(this)));
+        control_message_proxy_thunk_(this),
+        control_message_proxy_(&control_message_proxy_thunk_) {
+    thread_checker_.DetachFromThread();
     control_message_handler_.SetDescription(
         "IPC::mojom::Bootstrap [master] PipeControlMessageHandler");
   }
 
-  mojo::AssociatedGroup* associated_group() { return associated_group_.get(); }
+  void Bind(mojo::ScopedMessagePipeHandle handle) {
+    DCHECK(thread_checker_.CalledOnValidThread());
+    DCHECK(task_runner_->BelongsToCurrentThread());
+    connector_.reset(new mojo::Connector(
+        std::move(handle), mojo::Connector::SINGLE_THREADED_SEND,
+        task_runner_));
+    connector_->set_incoming_receiver(&header_validator_);
+    connector_->set_connection_error_handler(
+        base::Bind(&ChannelAssociatedGroupController::OnPipeError,
+                   base::Unretained(this)));
+
+    std::vector<std::unique_ptr<mojo::Message>> outgoing_messages;
+    std::swap(outgoing_messages, outgoing_messages_);
+    for (auto& message : outgoing_messages)
+      SendMessage(message.get());
+  }
+
+  void CreateChannelEndpoints(mojom::ChannelAssociatedPtr* sender,
+                              mojom::ChannelAssociatedRequest* receiver) {
+    mojo::InterfaceId sender_id, receiver_id;
+    if (set_interface_id_namespace_bit_) {
+      sender_id = 1 | mojo::kInterfaceIdNamespaceMask;
+      receiver_id = 1;
+    } else {
+      sender_id = 1;
+      receiver_id = 1 | mojo::kInterfaceIdNamespaceMask;
+    }
+
+    {
+      base::AutoLock locker(lock_);
+      Endpoint* sender_endpoint = new Endpoint(this, sender_id);
+      Endpoint* receiver_endpoint = new Endpoint(this, receiver_id);
+      endpoints_.insert({ sender_id, sender_endpoint });
+      endpoints_.insert({ receiver_id, receiver_endpoint });
+    }
+
+    mojo::ScopedInterfaceEndpointHandle sender_handle =
+        CreateScopedInterfaceEndpointHandle(sender_id, true);
+    mojo::ScopedInterfaceEndpointHandle receiver_handle =
+        CreateScopedInterfaceEndpointHandle(receiver_id, true);
+
+    sender->Bind(mojom::ChannelAssociatedPtrInfo(std::move(sender_handle), 0));
+    receiver->Bind(std::move(receiver_handle));
+  }
 
   void ShutDown() {
     DCHECK(thread_checker_.CalledOnValidThread());
-    connector_.CloseMessagePipe();
+    connector_->CloseMessagePipe();
     OnPipeError();
-    associated_group_.reset();
-  }
-
-  void SetProxyTaskRunner(
-      scoped_refptr<base::SingleThreadTaskRunner> proxy_task_runner) {
-    proxy_task_runner_ = proxy_task_runner;
+    connector_.reset();
   }
 
   // mojo::AssociatedGroupController:
   void CreateEndpointHandlePair(
       mojo::ScopedInterfaceEndpointHandle* local_endpoint,
       mojo::ScopedInterfaceEndpointHandle* remote_endpoint) override {
     base::AutoLock locker(lock_);
     uint32_t id = 0;
     do {
       if (next_interface_id_ >= mojo::kInterfaceIdNamespaceMask)
-        next_interface_id_ = 1;
-      id = (next_interface_id_++) | id_namespace_mask_;
+        next_interface_id_ = 2;
+      id = next_interface_id_++;
+      if (set_interface_id_namespace_bit_)
+        id |= mojo::kInterfaceIdNamespaceMask;
     } while (ContainsKey(endpoints_, id));
 
     Endpoint* endpoint = new Endpoint(this, id);
     if (encountered_error_)
       endpoint->set_peer_closed();
     endpoints_.insert({ id, endpoint });
 
     *local_endpoint = CreateScopedInterfaceEndpointHandle(id, true);
     *remote_endpoint = CreateScopedInterfaceEndpointHandle(id, false);
   }
@@ skipping 13 matching lines @@
   }
 
   void CloseEndpointHandle(mojo::InterfaceId id, bool is_local) override {
     if (!mojo::IsValidInterfaceId(id))
       return;
 
     base::AutoLock locker(lock_);
     if (!is_local) {
       DCHECK(ContainsKey(endpoints_, id));
       DCHECK(!mojo::IsMasterInterfaceId(id));
-      NotifyEndpointClosedBeforeSent(id);
+      control_message_proxy_.NotifyEndpointClosedBeforeSent(id);
       return;
     }
 
     DCHECK(ContainsKey(endpoints_, id));
     Endpoint* endpoint = endpoints_[id].get();
     DCHECK(!endpoint->client());
     DCHECK(!endpoint->closed());
     MarkClosedAndMaybeRemove(endpoint);
 
     if (!mojo::IsMasterInterfaceId(id))
-      NotifyPeerEndpointClosed(id);
+      control_message_proxy_.NotifyPeerEndpointClosed(id);
   }
 
   mojo::InterfaceEndpointController* AttachEndpointClient(
       const mojo::ScopedInterfaceEndpointHandle& handle,
       mojo::InterfaceEndpointClient* client,
       scoped_refptr<base::SingleThreadTaskRunner> runner) override {
     const mojo::InterfaceId id = handle.id();
 
     DCHECK(mojo::IsValidInterfaceId(id));
     DCHECK(client);
 
+    scoped_refptr<ChannelAssociatedGroupController> keepalive(this);

[yzshen1, 2016/07/20 22:54:51]

I think this is not necessary because the InterfaceEndpointClient must hold a
ref to this. Besides, it seems no operation in this method will decrease ref
count. Have I missed anything?

[Ken Rockot(use gerrit already), 2016/07/20 23:10:20]

NotifyEndpointOfError may synchronously call into the client's NotifyError,
which may invoke an arbitrary user callback that decreases ref count.

However you're right that in this case the InterfaceEndpointClient must already
be holding a ref, so I've removed this keepalive.

+
     base::AutoLock locker(lock_);
     DCHECK(ContainsKey(endpoints_, id));
 
     Endpoint* endpoint = endpoints_[id].get();
     endpoint->AttachClient(client, std::move(runner));
 
     if (endpoint->peer_closed())
       NotifyEndpointOfError(endpoint, true /* force_async */);
 
     return endpoint;
   }
 
   void DetachEndpointClient(
       const mojo::ScopedInterfaceEndpointHandle& handle) override {
     const mojo::InterfaceId id = handle.id();
 
     DCHECK(mojo::IsValidInterfaceId(id));
 
     base::AutoLock locker(lock_);
     DCHECK(ContainsKey(endpoints_, id));
 
     Endpoint* endpoint = endpoints_[id].get();
     endpoint->DetachClient();
   }
 
   void RaiseError() override {
     if (task_runner_->BelongsToCurrentThread()) {
-      connector_.RaiseError();
+      connector_->RaiseError();
     } else {
       task_runner_->PostTask(
           FROM_HERE,
           base::Bind(&ChannelAssociatedGroupController::RaiseError, this));
     }
   }
 
  private:
   class Endpoint;
+  class ControlMessageProxyThunk;
   friend class Endpoint;
+  friend class ControlMessageProxyThunk;
 
   class Endpoint : public base::RefCountedThreadSafe<Endpoint>,
                    public mojo::InterfaceEndpointController {
    public:
     Endpoint(ChannelAssociatedGroupController* controller, mojo::InterfaceId id)
         : controller_(controller), id_(id) {}
 
     mojo::InterfaceId id() const { return id_; }
 
     bool closed() const {
@@ skipping 82 matching lines @@
     const mojo::InterfaceId id_;
 
     bool closed_ = false;
     bool peer_closed_ = false;
     mojo::InterfaceEndpointClient* client_ = nullptr;
     scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
 
     DISALLOW_COPY_AND_ASSIGN(Endpoint);
   };
 
+  class ControlMessageProxyThunk : public MessageReceiver {
+   public:
+    explicit ControlMessageProxyThunk(
+        ChannelAssociatedGroupController* controller)
+        : controller_(controller) {}
+
+   private:
+    // MessageReceiver:
+    bool Accept(mojo::Message* message) override {
+      return controller_->SendMessage(message);
+    }
+
+    ChannelAssociatedGroupController* controller_;
+
+    DISALLOW_COPY_AND_ASSIGN(ControlMessageProxyThunk);
+  };
+
   ~ChannelAssociatedGroupController() override {
     base::AutoLock locker(lock_);
-
     for (auto iter = endpoints_.begin(); iter != endpoints_.end();) {
       Endpoint* endpoint = iter->second.get();
       ++iter;
 
       DCHECK(endpoint->closed());
       MarkPeerClosedAndMaybeRemove(endpoint);
     }
 
     DCHECK(endpoints_.empty());
   }
 
   bool SendMessage(mojo::Message* message) {
     if (task_runner_->BelongsToCurrentThread()) {
       DCHECK(thread_checker_.CalledOnValidThread());
-      return connector_.Accept(message);
+      if (!connector_) {
+        // Pipe may not be bound yet, so we queue the message.
+        std::unique_ptr<mojo::Message> queued_message(new mojo::Message);
+        message->MoveTo(queued_message.get());
+        outgoing_messages_.emplace_back(std::move(queued_message));
+        return true;
+      }
+      return connector_->Accept(message);
     } else {
       // We always post tasks to the master endpoint thread when called from the
       // proxy thread in order to simulate IPC::ChannelProxy::Send behavior.
       DCHECK(proxy_task_runner_ &&
              proxy_task_runner_->BelongsToCurrentThread());
       std::unique_ptr<mojo::Message> passed_message(new mojo::Message);
       message->MoveTo(passed_message.get());
       task_runner_->PostTask(
           FROM_HERE,
           base::Bind(
@@ skipping 24 matching lines @@
       Endpoint* endpoint = iter->second.get();
       ++iter;
 
       if (endpoint->client())
         endpoints_to_notify.push_back(endpoint);
 
       MarkPeerClosedAndMaybeRemove(endpoint);
     }
 
     for (auto& endpoint : endpoints_to_notify) {
-      // Because an notification may in turn detach any endpoint, we have to
+      // Because a notification may in turn detach any endpoint, we have to
       // check each client again here.
       if (endpoint->client())
         NotifyEndpointOfError(endpoint.get(), false /* force_async */);
     }
   }
 
   void NotifyEndpointOfError(Endpoint* endpoint, bool force_async) {
     lock_.AssertAcquired();
     DCHECK(endpoint->task_runner() && endpoint->client());
     if (endpoint->task_runner()->BelongsToCurrentThread() && !force_async) {
       mojo::InterfaceEndpointClient* client = endpoint->client();
 
       base::AutoUnlock unlocker(lock_);
       client->NotifyError();
     } else {
       endpoint->task_runner()->PostTask(
           FROM_HERE,
           base::Bind(&ChannelAssociatedGroupController
                 ::NotifyEndpointOfErrorOnEndpointThread, this,
                 make_scoped_refptr(endpoint)));
     }
   }
 
   void NotifyEndpointOfErrorOnEndpointThread(scoped_refptr<Endpoint> endpoint) {
+    scoped_refptr<ChannelAssociatedGroupController> keepalive(this);

[yzshen1, 2016/07/20 22:54:51]

Why do we need to keep it alive explicitly? I thought the closure will keep a
ref until it is done running?

[Ken Rockot(use gerrit already), 2016/07/20 23:10:20]

Ah, you are right :)  Fixed.

     base::AutoLock locker(lock_);
     if (!endpoint->client())
       return;
     DCHECK(endpoint->task_runner()->BelongsToCurrentThread());
     NotifyEndpointOfError(endpoint.get(), false /* force_async */);
   }
 
   void MarkClosedAndMaybeRemove(Endpoint* endpoint) {
     lock_.AssertAcquired();
     endpoint->set_closed();
     if (endpoint->closed() && endpoint->peer_closed())
       endpoints_.erase(endpoint->id());
   }
 
   void MarkPeerClosedAndMaybeRemove(Endpoint* endpoint) {
     lock_.AssertAcquired();
     endpoint->set_peer_closed();
     if (endpoint->closed() && endpoint->peer_closed())
       endpoints_.erase(endpoint->id());
   }
 
-  void NotifyPeerEndpointClosed(mojo::InterfaceId id) {
-    if (task_runner_->BelongsToCurrentThread()) {
-      if (connector_.is_valid())
-        control_message_proxy_.NotifyPeerEndpointClosed(id);
-    } else {
-      task_runner_->PostTask(
-          FROM_HERE,
-          base::Bind(&ChannelAssociatedGroupController
-              ::NotifyPeerEndpointClosed, this, id));
-    }
-  }
-
-  void NotifyEndpointClosedBeforeSent(mojo::InterfaceId id) {
-    if (task_runner_->BelongsToCurrentThread()) {
-      if (connector_.is_valid())
-        control_message_proxy_.NotifyEndpointClosedBeforeSent(id);
-    } else {
-      task_runner_->PostTask(
-          FROM_HERE,
-          base::Bind(&ChannelAssociatedGroupController
-              ::NotifyEndpointClosedBeforeSent, this, id));
-    }
-  }
-
   Endpoint* FindOrInsertEndpoint(mojo::InterfaceId id, bool* inserted) {
     lock_.AssertAcquired();
     DCHECK(!inserted || !*inserted);
 
     auto iter = endpoints_.find(id);
     if (iter != endpoints_.end())
       return iter->second.get();
 
     Endpoint* endpoint = new Endpoint(this, id);
     endpoints_.insert({ id, endpoint });
@@ skipping 74 matching lines @@
       RaiseError();
   }
 
   Endpoint* GetEndpointForDispatch(mojo::InterfaceId id) {
     lock_.AssertAcquired();
     bool inserted = false;
     Endpoint* endpoint = FindOrInsertEndpoint(id, &inserted);
     if (inserted) {
       MarkClosedAndMaybeRemove(endpoint);
       if (!mojo::IsMasterInterfaceId(id))
-        NotifyPeerEndpointClosed(id);
+        control_message_proxy_.NotifyPeerEndpointClosed(id);
       return nullptr;
     }
 
     if (endpoint->closed())
       return nullptr;
 
     return endpoint;
   }
 
   // mojo::PipeControlMessageHandlerDelegate:
   bool OnPeerAssociatedEndpointClosed(mojo::InterfaceId id) override {
     DCHECK(thread_checker_.CalledOnValidThread());
 
     if (mojo::IsMasterInterfaceId(id))
       return false;
 
+    scoped_refptr<ChannelAssociatedGroupController> keepalive(this);
     base::AutoLock locker(lock_);
     scoped_refptr<Endpoint> endpoint = FindOrInsertEndpoint(id, nullptr);
     if (!endpoint->peer_closed()) {
       if (endpoint->client())
         NotifyEndpointOfError(endpoint.get(), false /* force_async */);
       MarkPeerClosedAndMaybeRemove(endpoint.get());
     }
 
     return true;
   }
@@ skipping 10 matching lines @@
     MarkClosedAndMaybeRemove(endpoint);
     control_message_proxy_.NotifyPeerEndpointClosed(id);
     return true;
   }
 
   // Checked in places which must be run on the master endpoint's thread.
   base::ThreadChecker thread_checker_;
 
   scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
   scoped_refptr<base::SingleThreadTaskRunner> proxy_task_runner_;
-  const uint32_t id_namespace_mask_;
-  std::unique_ptr<mojo::AssociatedGroup> associated_group_;
-  mojo::Connector connector_;
+  const bool set_interface_id_namespace_bit_;
+  std::unique_ptr<mojo::Connector> connector_;
   mojo::MessageHeaderValidator header_validator_;
   mojo::PipeControlMessageHandler control_message_handler_;
+  ControlMessageProxyThunk control_message_proxy_thunk_;
+  mojo::PipeControlMessageProxy control_message_proxy_;
+
+  // Outgoing messages that were sent before this controller was bound to a
+  // real message pipe.
+  std::vector<std::unique_ptr<mojo::Message>> outgoing_messages_;
 
   // Guards the fields below for thread-safe access.
   base::Lock lock_;
 
   bool encountered_error_ = false;
-  uint32_t next_interface_id_ = 1;
+
+  // ID #1 is reserved for the mojom::Channel interface.
+  uint32_t next_interface_id_ = 2;
+
   std::map<uint32_t, scoped_refptr<Endpoint>> endpoints_;
-  mojo::PipeControlMessageProxy control_message_proxy_;
 
   DISALLOW_COPY_AND_ASSIGN(ChannelAssociatedGroupController);
 };
 
-class BootstrapMasterProxy {
+class MojoBootstrapImpl : public MojoBootstrap {
  public:
-  BootstrapMasterProxy() {}
-  ~BootstrapMasterProxy() {
-    endpoint_client_.reset();
-    proxy_.reset();
-    if (controller_)
-      controller_->ShutDown();
+  MojoBootstrapImpl(
+      mojo::ScopedMessagePipeHandle handle,
+      Delegate* delegate,
+      const scoped_refptr<ChannelAssociatedGroupController> controller)
+            : controller_(controller),
+        handle_(std::move(handle)),
+        delegate_(delegate) {
+    associated_group_ = controller_->CreateAssociatedGroup();
   }
 
-  void Bind(mojo::ScopedMessagePipeHandle handle) {
-    DCHECK(!controller_);
-    controller_ = new ChannelAssociatedGroupController(true, std::move(handle));
-    endpoint_client_.reset(new mojo::InterfaceEndpointClient(
-        controller_->CreateLocalEndpointHandle(mojo::kMasterInterfaceId),
-        nullptr,
-        base::MakeUnique<typename mojom::Bootstrap::ResponseValidator_>(),
-        false, base::ThreadTaskRunnerHandle::Get()));
-    proxy_.reset(new mojom::BootstrapProxy(endpoint_client_.get()));
-    proxy_->serialization_context()->group_controller = controller_;
-  }
-
-  void set_connection_error_handler(const base::Closure& handler) {
-    DCHECK(endpoint_client_);
-    endpoint_client_->set_connection_error_handler(handler);
-  }
-
-  mojo::AssociatedGroup* associated_group() {
-    DCHECK(controller_);
-    return controller_->associated_group();
-  }
-
-  ChannelAssociatedGroupController* controller() {
-    DCHECK(controller_);
-    return controller_.get();
-  }
-
-  mojom::Bootstrap* operator->() {
-    DCHECK(proxy_);
-    return proxy_.get();
+  ~MojoBootstrapImpl() override {
+    controller_->ShutDown();
   }
 
  private:
-  std::unique_ptr<mojom::BootstrapProxy> proxy_;
+  // MojoBootstrap:
+  void Connect() override {
+    controller_->Bind(std::move(handle_));
+
+    IPC::mojom::ChannelAssociatedPtr sender;
+    IPC::mojom::ChannelAssociatedRequest receiver;
+    controller_->CreateChannelEndpoints(&sender, &receiver);
+
+    delegate_->OnPipesAvailable(std::move(sender), std::move(receiver));
+  }
+
+  mojo::AssociatedGroup* GetAssociatedGroup() override {
+    return associated_group_.get();
+  }
+
   scoped_refptr<ChannelAssociatedGroupController> controller_;
-  std::unique_ptr<mojo::InterfaceEndpointClient> endpoint_client_;
 
-  DISALLOW_COPY_AND_ASSIGN(BootstrapMasterProxy);
+  mojo::ScopedMessagePipeHandle handle_;
+  Delegate* delegate_;
+  std::unique_ptr<mojo::AssociatedGroup> associated_group_;
+
+  DISALLOW_COPY_AND_ASSIGN(MojoBootstrapImpl);
 };
 
-class BootstrapMasterBinding {
- public:
-  explicit BootstrapMasterBinding(mojom::Bootstrap* impl) {
-    stub_.set_sink(impl);
-  }
-
-  ~BootstrapMasterBinding() {
-    endpoint_client_.reset();
-    if (controller_)
-      controller_->ShutDown();
-  }
-
-  void set_connection_error_handler(const base::Closure& handler) {
-    DCHECK(endpoint_client_);
-    endpoint_client_->set_connection_error_handler(handler);
-  }
-
-  mojo::AssociatedGroup* associated_group() {
-    DCHECK(controller_);
-    return controller_->associated_group();
-  }
-
-  ChannelAssociatedGroupController* controller() {
-    DCHECK(controller_);
-    return controller_.get();
-  }
-
-  void Bind(mojo::ScopedMessagePipeHandle handle) {
-    DCHECK(!controller_);
-    controller_ =
-        new ChannelAssociatedGroupController(false, std::move(handle));
-    stub_.serialization_context()->group_controller = controller_;
-
-    endpoint_client_.reset(new mojo::InterfaceEndpointClient(
-        controller_->CreateLocalEndpointHandle(mojo::kMasterInterfaceId),
-        &stub_,
-        base::MakeUnique<typename mojom::Bootstrap::RequestValidator_>(),
-        false, base::ThreadTaskRunnerHandle::Get()));
-  }
-
- private:
-  mojom::BootstrapStub stub_;
-  scoped_refptr<ChannelAssociatedGroupController> controller_;
-  std::unique_ptr<mojo::InterfaceEndpointClient> endpoint_client_;
-
-  DISALLOW_COPY_AND_ASSIGN(BootstrapMasterBinding);
-};
-
-// MojoBootstrap for the server process. You should create the instance
-// using MojoBootstrap::Create().
-class MojoServerBootstrap : public MojoBootstrap {
- public:
-  MojoServerBootstrap();
-
- private:
-  // MojoBootstrap implementation.
-  void Connect() override;
-
-  mojo::AssociatedGroup* GetAssociatedGroup() override {
-    return bootstrap_.associated_group();
-  }
-
-  void SetProxyTaskRunner(
-      scoped_refptr<base::SingleThreadTaskRunner> task_runner) override {
-    bootstrap_.controller()->SetProxyTaskRunner(task_runner);
-  }
-
-  void OnInitDone(int32_t peer_pid);
-
-  BootstrapMasterProxy bootstrap_;
-  IPC::mojom::ChannelAssociatedPtrInfo send_channel_;
-  IPC::mojom::ChannelAssociatedRequest receive_channel_request_;
-
-  DISALLOW_COPY_AND_ASSIGN(MojoServerBootstrap);
-};
-
-MojoServerBootstrap::MojoServerBootstrap() = default;
-
-void MojoServerBootstrap::Connect() {
-  DCHECK_EQ(state(), STATE_INITIALIZED);
-
-  bootstrap_.Bind(TakeHandle());
-  bootstrap_.set_connection_error_handler(
-      base::Bind(&MojoServerBootstrap::Fail, base::Unretained(this)));
-
-  IPC::mojom::ChannelAssociatedRequest send_channel_request;
-  IPC::mojom::ChannelAssociatedPtrInfo receive_channel;
-
-  bootstrap_.associated_group()->CreateAssociatedInterface(
-      mojo::AssociatedGroup::WILL_PASS_REQUEST, &send_channel_,
-      &send_channel_request);
-  bootstrap_.associated_group()->CreateAssociatedInterface(
-      mojo::AssociatedGroup::WILL_PASS_PTR, &receive_channel,
-      &receive_channel_request_);
-
-  bootstrap_->Init(
-      std::move(send_channel_request), std::move(receive_channel),
-      GetSelfPID(),
-      base::Bind(&MojoServerBootstrap::OnInitDone, base::Unretained(this)));
-
-  set_state(STATE_WAITING_ACK);
-}
-
-void MojoServerBootstrap::OnInitDone(int32_t peer_pid) {
-  if (state() != STATE_WAITING_ACK) {
-    set_state(STATE_ERROR);
-    LOG(ERROR) << "Got inconsistent message from client.";
-    return;
-  }
-
-  set_state(STATE_READY);
-  bootstrap_.set_connection_error_handler(base::Closure());
-  delegate()->OnPipesAvailable(std::move(send_channel_),
-                               std::move(receive_channel_request_), peer_pid);
-}
-
-// MojoBootstrap for client processes. You should create the instance
-// using MojoBootstrap::Create().
-class MojoClientBootstrap : public MojoBootstrap, public mojom::Bootstrap {
- public:
-  MojoClientBootstrap();
-
- private:
-  // MojoBootstrap implementation.
-  void Connect() override;
-
-  mojo::AssociatedGroup* GetAssociatedGroup() override {
-    return binding_.associated_group();
-  }
-
-  void SetProxyTaskRunner(
-      scoped_refptr<base::SingleThreadTaskRunner> task_runner) override {
-    binding_.controller()->SetProxyTaskRunner(task_runner);
-  }
-
-  // mojom::Bootstrap implementation.
-  void Init(mojom::ChannelAssociatedRequest receive_channel,
-            mojom::ChannelAssociatedPtrInfo send_channel,
-            int32_t peer_pid,
-            const InitCallback& callback) override;
-
-  BootstrapMasterBinding binding_;
-
-  DISALLOW_COPY_AND_ASSIGN(MojoClientBootstrap);
-};
-
-MojoClientBootstrap::MojoClientBootstrap() : binding_(this) {}
-
-void MojoClientBootstrap::Connect() {
-  binding_.Bind(TakeHandle());
-  binding_.set_connection_error_handler(
-      base::Bind(&MojoClientBootstrap::Fail, base::Unretained(this)));
-}
-
-void MojoClientBootstrap::Init(mojom::ChannelAssociatedRequest receive_channel,
-                               mojom::ChannelAssociatedPtrInfo send_channel,
-                               int32_t peer_pid,
-                               const InitCallback& callback) {
-  callback.Run(GetSelfPID());
-  set_state(STATE_READY);
-  binding_.set_connection_error_handler(base::Closure());
-  delegate()->OnPipesAvailable(std::move(send_channel),
-                               std::move(receive_channel), peer_pid);
-}
-
 }  // namespace
 
-// MojoBootstrap
-
 // static
 std::unique_ptr<MojoBootstrap> MojoBootstrap::Create(
     mojo::ScopedMessagePipeHandle handle,
     Channel::Mode mode,
-    Delegate* delegate) {
-  CHECK(mode == Channel::MODE_CLIENT || mode == Channel::MODE_SERVER);
-  std::unique_ptr<MojoBootstrap> self =
-      mode == Channel::MODE_CLIENT
-          ? std::unique_ptr<MojoBootstrap>(new MojoClientBootstrap)
-          : std::unique_ptr<MojoBootstrap>(new MojoServerBootstrap);
-
-  self->Init(std::move(handle), delegate);
-  return self;
-}
-
-MojoBootstrap::MojoBootstrap() : delegate_(NULL), state_(STATE_INITIALIZED) {
-}
-
-MojoBootstrap::~MojoBootstrap() {}
-
-void MojoBootstrap::Init(mojo::ScopedMessagePipeHandle handle,
-                         Delegate* delegate) {
-  handle_ = std::move(handle);
-  delegate_ = delegate;
-}
-
-base::ProcessId MojoBootstrap::GetSelfPID() const {
-#if defined(OS_LINUX)
-  if (int global_pid = Channel::GetGlobalPid())
-    return global_pid;
-#endif  // OS_LINUX
-#if defined(OS_NACL)
-  return -1;
-#else
-  return base::GetCurrentProcId();
-#endif  // defined(OS_NACL)
-}
-
-void MojoBootstrap::Fail() {
-  set_state(STATE_ERROR);
-  delegate()->OnBootstrapError();
-}
-
-bool MojoBootstrap::HasFailed() const {
-  return state() == STATE_ERROR;
-}
-
-mojo::ScopedMessagePipeHandle MojoBootstrap::TakeHandle() {
-  return std::move(handle_);
+    Delegate* delegate,
+    const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner) {
+  return base::MakeUnique<MojoBootstrapImpl>(
+      std::move(handle), delegate,
+      new ChannelAssociatedGroupController(mode == Channel::MODE_SERVER,
+                                           ipc_task_runner));
 }
 
 }  // namespace IPC