IPC::ChannelMojo: Introduce IPC::MojoBootstrap for Windows

ipc/mojo/ipc_channel_mojo.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/mojo/ipc_channel_mojo.h"
 
 #include "base/bind.h"
 #include "base/bind_helpers.h"
 #include "base/lazy_instance.h"
 #include "ipc/ipc_listener.h"
+#include "ipc/mojo/ipc_mojo_bootstrap.h"
 #include "mojo/embedder/embedder.h"
 
 #if defined(OS_POSIX) && !defined(OS_NACL)
 #include "ipc/file_descriptor_set_posix.h"
 #endif
 
 namespace IPC {
 
 namespace {
 
@@ skipping 12 matching lines @@
 
   virtual void OnChannelError() OVERRIDE {
     NOTREACHED();
   }
 
   virtual void OnBadMessageReceived(const Message& message) OVERRIDE {
     NOTREACHED();
   }
 };
 
-base::LazyInstance<NullListener> g_null_listener = LAZY_INSTANCE_INITIALIZER;
-
 class MojoChannelFactory : public ChannelFactory {
  public:
   MojoChannelFactory(
       ChannelHandle channel_handle,
       Channel::Mode mode,
       scoped_refptr<base::TaskRunner> io_thread_task_runner)
       : channel_handle_(channel_handle),
         mode_(mode),
         io_thread_task_runner_(io_thread_task_runner) {
   }
 
   virtual std::string GetName() const OVERRIDE {
     return channel_handle_.name;
   }
 
   virtual scoped_ptr<Channel> BuildChannel(Listener* listener) OVERRIDE {
     return ChannelMojo::Create(
         channel_handle_,
         mode_,
         listener,
         io_thread_task_runner_).PassAs<Channel>();
   }
 
  private:
   ChannelHandle channel_handle_;
   Channel::Mode mode_;
   scoped_refptr<base::TaskRunner> io_thread_task_runner_;
 };
 
-mojo::embedder::PlatformHandle ToPlatformHandle(
-    const ChannelHandle& handle) {
-#if defined(OS_POSIX) && !defined(OS_NACL)
-  return mojo::embedder::PlatformHandle(handle.socket.fd);
-#elif defined(OS_WIN)
-  return mojo::embedder::PlatformHandle(handle.pipe.handle);
-#else
-#error "Unsupported Platform!"
-#endif
-}
-
 //------------------------------------------------------------------------------
 
 // TODO(morrita): This should be built using higher-level Mojo construct
 // for clarity and extensibility.
 class HelloMessage {
  public:
   static Pickle CreateRequest(int32 pid) {
     Pickle request;
     request.WriteString(kHelloRequestMagic);
     request.WriteInt(pid);
@@ skipping 356 matching lines @@
     mojo::embedder::ChannelInfo* ptr) const {
   mojo::embedder::DestroyChannelOnIOThread(ptr);
 }
 
 //------------------------------------------------------------------------------
 
 // static
 scoped_ptr<ChannelMojo> ChannelMojo::Create(
     const ChannelHandle &channel_handle, Mode mode, Listener* listener,
     scoped_refptr<base::TaskRunner> io_thread_task_runner) {
-  return make_scoped_ptr(new ChannelMojo(
-      Channel::Create(channel_handle, mode, g_null_listener.Pointer()),
-      mode, listener, io_thread_task_runner));
+  return make_scoped_ptr(
+      new ChannelMojo(channel_handle, mode, listener, io_thread_task_runner));
 }
 
 // static
 scoped_ptr<ChannelFactory> ChannelMojo::CreateFactory(
     const ChannelHandle &channel_handle, Mode mode,
     scoped_refptr<base::TaskRunner> io_thread_task_runner) {
   return make_scoped_ptr(
       new MojoChannelFactory(
           channel_handle, mode,
           io_thread_task_runner)).PassAs<ChannelFactory>();
 }
 
-ChannelMojo::ChannelMojo(
-    scoped_ptr<Channel> bootstrap, Mode mode, Listener* listener,
-    scoped_refptr<base::TaskRunner> io_thread_task_runner)
-    : bootstrap_(bootstrap.Pass()),
-      mode_(mode), listener_(listener),
+ChannelMojo::ChannelMojo(ChannelHandle handle,
+                         Mode mode,
+                         Listener* listener,
+                         scoped_refptr<base::TaskRunner> io_thread_task_runner)
+    : mode_(mode),
+      listener_(listener),
       peer_pid_(base::kNullProcessId),
       weak_factory_(this) {
   if (base::MessageLoopProxy::current() == io_thread_task_runner.get()) {
-    InitOnIOThread();
+    InitBootstrap(handle);
   } else {
-    io_thread_task_runner->PostTask(FROM_HERE,
-                                    base::Bind(&ChannelMojo::InitOnIOThread,
-                                               weak_factory_.GetWeakPtr()));
+    io_thread_task_runner->PostTask(
+        FROM_HERE,
+        base::Bind(
+            &ChannelMojo::InitBootstrap, weak_factory_.GetWeakPtr(), handle));
   }
 }
 
 ChannelMojo::~ChannelMojo() {
   Close();
 }
 
-void ChannelMojo::InitOnIOThread() {
+void ChannelMojo::InitBootstrap(ChannelHandle handle) {
+  DCHECK(base::MessageLoopForIO::IsCurrent());
+  bootstrap_ = MojoBootstrap::Create(handle, mode_, this);
+}
+
+void ChannelMojo::InitControlReader(
+    mojo::embedder::ScopedPlatformHandle handle) {
+  DCHECK(base::MessageLoopForIO::IsCurrent());
   mojo::embedder::ChannelInfo* channel_info;
   mojo::ScopedMessagePipeHandle control_pipe =
-      mojo::embedder::CreateChannelOnIOThread(
-          mojo::embedder::ScopedPlatformHandle(
-              ToPlatformHandle(bootstrap_->TakePipeHandle())),
-          &channel_info);
+      mojo::embedder::CreateChannelOnIOThread(handle.Pass(), &channel_info);
   channel_info_.reset(channel_info);
 
   switch (mode_) {
     case MODE_SERVER:
       control_reader_.reset(new ServerControlReader(control_pipe.Pass(), this));
       break;
     case MODE_CLIENT:
       control_reader_.reset(new ClientControlReader(control_pipe.Pass(), this));
       break;
     default:
       NOTREACHED();
       break;
   }
 }
 
 bool ChannelMojo::Connect() {
   DCHECK(!message_reader_);
-  return control_reader_->Connect();
+  DCHECK(!control_reader_);
+  return bootstrap_->Connect();
 }
 
 void ChannelMojo::Close() {
   control_reader_.reset();
   message_reader_.reset();
   channel_info_.reset();
 }
 
+void ChannelMojo::OnPipeAvailable(mojo::embedder::ScopedPlatformHandle handle) {
+  InitControlReader(handle.Pass());
+  control_reader_->Connect();
+}
+
+void ChannelMojo::OnBootstrapError() {
+  listener_->OnChannelError();
+}
+
 void ChannelMojo::OnConnected(mojo::ScopedMessagePipeHandle pipe) {
   message_reader_ = make_scoped_ptr(new MessageReader(pipe.Pass(), this));
 
   for (size_t i = 0; i < pending_messages_.size(); ++i) {
     message_reader_->Send(make_scoped_ptr(pending_messages_[i]));
     pending_messages_[i] = NULL;
   }
 
   pending_messages_.clear();
 
@@ skipping 16 matching lines @@
   }
 
   return message_reader_->Send(make_scoped_ptr(message));
 }
 
 base::ProcessId ChannelMojo::GetPeerPID() const {
   return peer_pid_;
 }
 
 base::ProcessId ChannelMojo::GetSelfPID() const {
-  return bootstrap_->GetSelfPID();
+  return base::GetCurrentProcId();
 }
 
 ChannelHandle ChannelMojo::TakePipeHandle() {
-  return bootstrap_->TakePipeHandle();
+  NOTREACHED();
+  return ChannelHandle();
+}
+
+void ChannelMojo::OnClientLaunched(base::ProcessHandle handle) {
+  bootstrap_->OnClientLaunched(handle);
 }
 
 void ChannelMojo::OnMessageReceived(Message& message) {
   listener_->OnMessageReceived(message);
   if (message.dispatch_error())
     listener_->OnBadMessageReceived(message);
 }
 
 #if defined(OS_POSIX) && !defined(OS_NACL)
 int ChannelMojo::GetClientFileDescriptor() const {
   return bootstrap_->GetClientFileDescriptor();
 }
 
 int ChannelMojo::TakeClientFileDescriptor() {
   return bootstrap_->TakeClientFileDescriptor();
 }
 #endif  // defined(OS_POSIX) && !defined(OS_NACL)
 
 }  // namespace IPC