ChannelMojo: Replace hand written messsages with mojo.

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/client_channel.mojom.h"
 #include "ipc/mojo/ipc_channel_mojo_readers.h"
 #include "ipc/mojo/ipc_mojo_bootstrap.h"
 #include "mojo/edk/embedder/embedder.h"
+#include "mojo/public/cpp/bindings/error_handler.h"
 
 #if defined(OS_POSIX) && !defined(OS_NACL)
 #include "ipc/file_descriptor_set_posix.h"
 #endif
 
 namespace IPC {
 
 namespace {
 
 class MojoChannelFactory : public ChannelFactory {
@@ skipping 10 matching lines @@
   scoped_ptr<Channel> BuildChannel(Listener* listener) override {
     return ChannelMojo::Create(delegate_, channel_handle_, mode_, listener);
   }
 
  private:
   ChannelMojo::Delegate* delegate_;
   ChannelHandle channel_handle_;
   Channel::Mode mode_;
 };
 
+//------------------------------------------------------------------------------
+
+class ClientChannelMojo
+    : public ChannelMojo,
+      public NON_EXPORTED_BASE(mojo::InterfaceImpl<ClientChannel>) {
+ public:
+  ClientChannelMojo(ChannelMojo::Delegate* delegate,
+                    const ChannelHandle& handle,
+                    Listener* listener);
+
+  // MojoBootstrap::Delegate implementation
+  void OnPipeAvailable(mojo::embedder::ScopedPlatformHandle handle) override;
+  // InterfaceImpl implementation
+  void OnConnectionError() override;
+  // ClientChannel implementation
+  void InitClientChannel(
+      mojo::ScopedMessagePipeHandle pipe,
+      int32_t peer_pid,
+      const mojo::Callback<void(int32_t)>& callback) override;
+};

[viettrungluu, 2014/10/24 20:17:00]

nit: DISALLOW_COPY_AND_ASSIGN

[Hajime Morrita, 2014/10/24 21:05:37]

Done.

+
+ClientChannelMojo::ClientChannelMojo(ChannelMojo::Delegate* delegate,
+                                     const ChannelHandle& handle,
+                                     Listener* listener)
+    : ChannelMojo(delegate, handle, Channel::MODE_CLIENT, listener) {
+}
+
+void ClientChannelMojo::OnPipeAvailable(
+    mojo::embedder::ScopedPlatformHandle handle) {
+  mojo::WeakBindToPipe(this, CreateMessagingPipe(handle.Pass()));
+}
+
+void ClientChannelMojo::OnConnectionError() {
+  listener()->OnChannelError();
+}
+
+void ClientChannelMojo::InitClientChannel(
+    mojo::ScopedMessagePipeHandle pipe,
+    int32_t peer_pid,
+    const mojo::Callback<void(int32_t)>& callback) {
+  InitMessageReader(pipe.Pass(), static_cast<base::ProcessId>(peer_pid));
+  callback.Run(GetSelfPID());

[viettrungluu, 2014/10/24 20:17:00]

static_cast<int32>, like you have below?

This is slightly worrying, since base::ProcessId is a DWORD on Windows, and
casting a value that's out of range from unsigned to signed is undefined
behavior. Hrm.

[Hajime Morrita, 2014/10/24 21:05:37]

Turned it to uint32.

[viettrungluu, 2014/10/24 22:48:05]

That just gets you into trouble when you do |static_cast<base::ProcessId>()| on
Linux (since base::ProcessId is pid_t which is int32_t). (Hence the "hrm".)

+}
+
+//------------------------------------------------------------------------------
+
+class ServerChannelMojo : public ChannelMojo, public mojo::ErrorHandler {
+ public:
+  ServerChannelMojo(ChannelMojo::Delegate* delegate,
+                    const ChannelHandle& handle,
+                    Listener* listener);
+  virtual ~ServerChannelMojo();

[viettrungluu, 2014/10/24 20:17:00]

Should be marked "override" instead of "virtual".

[Hajime Morrita, 2014/10/24 21:05:37]

Done.

+
+  // MojoBootstrap::Delegate implementation
+  void OnPipeAvailable(mojo::embedder::ScopedPlatformHandle handle) override;
+  // ErrorHandler implementation
+  void OnConnectionError() override;
+  // Channel override
+  void Close() override;
+
+ private:
+  // ClientChannelClient implementation
+  void ClientChannelWasInitialized(int32_t peer_pid);
+
+  mojo::InterfacePtr<ClientChannel> client_channel_;
+  mojo::ScopedMessagePipeHandle message_pipe_;
+};

[viettrungluu, 2014/10/24 20:17:00]

"

[Hajime Morrita, 2014/10/24 21:05:37]

Done.

+
+ServerChannelMojo::ServerChannelMojo(ChannelMojo::Delegate* delegate,
+                                     const ChannelHandle& handle,
+                                     Listener* listener)
+    : ChannelMojo(delegate, handle, Channel::MODE_SERVER, listener) {
+}
+
+ServerChannelMojo::~ServerChannelMojo() {
+  Close();
+}
+
+void ServerChannelMojo::OnPipeAvailable(
+    mojo::embedder::ScopedPlatformHandle handle) {
+  mojo::ScopedMessagePipeHandle peer;
+  MojoResult create_result =
+      mojo::CreateMessagePipe(nullptr, &message_pipe_, &peer);
+  if (create_result != MOJO_RESULT_OK) {
+    DLOG(WARNING) << "mojo::CreateMessagePipe failed: " << create_result;
+    listener()->OnChannelError();
+    return;
+  }
+
+  client_channel_.Bind(CreateMessagingPipe(handle.Pass()));
+  client_channel_.set_error_handler(this);
+  client_channel_->InitClientChannel(
+      peer.Pass(),
+      static_cast<int32_t>(GetSelfPID()),
+      base::Bind(&ServerChannelMojo::ClientChannelWasInitialized,
+                 base::Unretained(this)));
+}
+
+void ServerChannelMojo::ClientChannelWasInitialized(int32_t peer_pid) {
+  InitMessageReader(message_pipe_.Pass(), peer_pid);
+}
+
+void ServerChannelMojo::OnConnectionError() {
+  listener()->OnChannelError();
+}
+
+void ServerChannelMojo::Close() {
+  client_channel_.reset();
+  message_pipe_.reset();
+  ChannelMojo::Close();
+}
+
 } // namespace
 
 //------------------------------------------------------------------------------
 
 void ChannelMojo::ChannelInfoDeleter::operator()(
     mojo::embedder::ChannelInfo* ptr) const {
   mojo::embedder::DestroyChannelOnIOThread(ptr);
 }
 
 //------------------------------------------------------------------------------
 
 // static
 bool ChannelMojo::ShouldBeUsed() {
   // TODO(morrita): Turn this on for a set of platforms.
   return false;
 }
 
 // static
 scoped_ptr<ChannelMojo> ChannelMojo::Create(ChannelMojo::Delegate* delegate,
                                             const ChannelHandle& channel_handle,
                                             Mode mode,
                                             Listener* listener) {
-  return make_scoped_ptr(
-      new ChannelMojo(delegate, channel_handle, mode, listener));
+  switch (mode) {
+    case Channel::MODE_CLIENT:
+      return make_scoped_ptr(
+          new ClientChannelMojo(delegate, channel_handle, listener));
+    case Channel::MODE_SERVER:
+      return make_scoped_ptr(
+          new ServerChannelMojo(delegate, channel_handle, listener));
+    default:
+      NOTREACHED();
+      return nullptr;
+  }
 }
 
 // static
 scoped_ptr<ChannelFactory> ChannelMojo::CreateServerFactory(
     ChannelMojo::Delegate* delegate,
     const ChannelHandle& channel_handle) {
   return make_scoped_ptr(
       new MojoChannelFactory(delegate, channel_handle, Channel::MODE_SERVER));
 }
 
@@ skipping 30 matching lines @@
 
 ChannelMojo::~ChannelMojo() {
   Close();
 }
 
 void ChannelMojo::InitDelegate(ChannelMojo::Delegate* delegate) {
   delegate_ = delegate->ToWeakPtr();
   delegate_->OnChannelCreated(weak_factory_.GetWeakPtr());
 }
 
-void ChannelMojo::InitControlReader(
+mojo::ScopedMessagePipeHandle ChannelMojo::CreateMessagingPipe(
     mojo::embedder::ScopedPlatformHandle handle) {
-  DCHECK(base::MessageLoopForIO::IsCurrent());
+  DCHECK(!channel_info_.get());
   mojo::embedder::ChannelInfo* channel_info;
-  mojo::ScopedMessagePipeHandle control_pipe =
+  mojo::ScopedMessagePipeHandle pipe =
       mojo::embedder::CreateChannelOnIOThread(handle.Pass(), &channel_info);
   channel_info_.reset(channel_info);
-
-  switch (mode_) {
-    case MODE_SERVER:
-      control_reader_.reset(
-          new internal::ServerControlReader(control_pipe.Pass(), this));
-      break;
-    case MODE_CLIENT:
-      control_reader_.reset(
-          new internal::ClientControlReader(control_pipe.Pass(), this));
-      break;
-    default:
-      NOTREACHED();
-      break;
-  }
+  return pipe.Pass();
 }
 
 bool ChannelMojo::Connect() {
   DCHECK(!message_reader_);
-  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) {
+void ChannelMojo::InitMessageReader(mojo::ScopedMessagePipeHandle pipe,
+                                    int32_t peer_pid) {
   message_reader_ =
       make_scoped_ptr(new internal::MessageReader(pipe.Pass(), this));
 
   for (size_t i = 0; i < pending_messages_.size(); ++i) {
     bool sent = message_reader_->Send(make_scoped_ptr(pending_messages_[i]));
     pending_messages_[i] = NULL;
     if (!sent) {
       pending_messages_.clear();
       listener_->OnChannelError();
       return;
     }
   }
 
   pending_messages_.clear();
 
+  set_peer_pid(peer_pid);
   listener_->OnChannelConnected(GetPeerPID());
 }
 
 void ChannelMojo::OnPipeClosed(internal::MessagePipeReader* reader) {
   Close();
 }
 
 void ChannelMojo::OnPipeError(internal::MessagePipeReader* reader) {
   listener_->OnChannelError();
 }
@@ skipping 93 matching lines @@
 
     fdset->CommitAll();
   }
 
   return MOJO_RESULT_OK;
 }
 
 #endif  // defined(OS_POSIX) && !defined(OS_NACL)
 
 }  // namespace IPC