Refactoring: Move MessagePipeReader subclasess out from ChannelMojo

There are a few MessagePipeReader sublcasses defined as inter-classes
of ChannelMojo. This makes it harder to hook them for unit testing.
This CL moves these classes their own file. Now unittest can
define subclasses of these to hook some behavor for testing.

This is a preparation for a crash fix.

BUG=410813
TEST=none
R=viettrungluu@chromium.org,yzshen@chromium.org

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

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

6 files changed, 524 insertions, 382 deletions

diff --git a/ipc/mojo/BUILD.gn b/ipc/mojo/BUILD.gn
index 58c1f03..743a320 100644
--- a/ipc/mojo/BUILD.gn
+++ b/ipc/mojo/BUILD.gn
@@ -6,6 +6,8 @@ component("mojo") {
   sources = [
     "ipc_channel_mojo.cc",
     "ipc_channel_mojo.h",
+    "ipc_channel_mojo_readers.cc",
+    "ipc_channel_mojo_readers.h",
     "ipc_message_pipe_reader.cc",
     "ipc_message_pipe_reader.h",
   ]
diff --git a/ipc/mojo/ipc_channel_mojo.cc b/ipc/mojo/ipc_channel_mojo.cc
index cdb9281..086b420 100644
--- a/ipc/mojo/ipc_channel_mojo.cc
+++ b/ipc/mojo/ipc_channel_mojo.cc
@@ -8,6 +8,7 @@
 #include "base/bind_helpers.h"
 #include "base/lazy_instance.h"
 #include "ipc/ipc_listener.h"
+#include "ipc/mojo/ipc_channel_mojo_readers.h"
 #include "mojo/embedder/embedder.h"
 
 #if defined(OS_POSIX) && !defined(OS_NACL)
@@ -82,371 +83,10 @@ mojo::embedder::PlatformHandle ToPlatformHandle(
 #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);
-    return request;
-  }
-
-  static bool ReadRequest(Pickle& pickle, int32* pid) {
-    PickleIterator iter(pickle);
-    std::string hello;
-    if (!iter.ReadString(&hello)) {
-      DLOG(WARNING) << "Failed to Read magic string.";
-      return false;
-    }
-
-    if (hello != kHelloRequestMagic) {
-      DLOG(WARNING) << "Magic mismatch:" << hello;
-      return false;
-    }
-
-    int read_pid;
-    if (!iter.ReadInt(&read_pid)) {
-      DLOG(WARNING) << "Failed to Read PID.";
-      return false;
-    }
-
-    *pid = read_pid;
-    return true;
-  }
-
-  static Pickle CreateResponse(int32 pid) {
-    Pickle request;
-    request.WriteString(kHelloResponseMagic);
-    request.WriteInt(pid);
-    return request;
-  }
-
-  static bool ReadResponse(Pickle& pickle, int32* pid) {
-    PickleIterator iter(pickle);
-    std::string hello;
-    if (!iter.ReadString(&hello)) {
-      DLOG(WARNING) << "Failed to read magic string.";
-      return false;
-    }
-
-    if (hello != kHelloResponseMagic) {
-      DLOG(WARNING) << "Magic mismatch:" << hello;
-      return false;
-    }
-
-    int read_pid;
-    if (!iter.ReadInt(&read_pid)) {
-      DLOG(WARNING) << "Failed to read PID.";
-      return false;
-    }
-
-    *pid = read_pid;
-    return true;
-  }
-
- private:
-  static const char* kHelloRequestMagic;
-  static const char* kHelloResponseMagic;
-};
-
-const char* HelloMessage::kHelloRequestMagic = "MREQ";
-const char* HelloMessage::kHelloResponseMagic = "MRES";
-
 } // namespace
 
 //------------------------------------------------------------------------------
 
-// A MessagePipeReader implemenation for IPC::Message communication.
-class ChannelMojo::MessageReader : public internal::MessagePipeReader {
- public:
-  MessageReader(mojo::ScopedMessagePipeHandle pipe, ChannelMojo* owner)
-      : internal::MessagePipeReader(pipe.Pass()),
-        owner_(owner) {}
-
-  bool Send(scoped_ptr<Message> message);
-  virtual void OnMessageReceived() OVERRIDE;
-  virtual void OnPipeClosed() OVERRIDE;
-  virtual void OnPipeError(MojoResult error) OVERRIDE;
-
- private:
-  ChannelMojo* owner_;
-};
-
-void ChannelMojo::MessageReader::OnMessageReceived() {
-  Message message(data_buffer().empty() ? "" : &data_buffer()[0],
-                  static_cast<uint32>(data_buffer().size()));
-
-  std::vector<MojoHandle> handle_buffer;
-  TakeHandleBuffer(&handle_buffer);
-#if defined(OS_POSIX) && !defined(OS_NACL)
-  for (size_t i = 0; i < handle_buffer.size(); ++i) {
-    mojo::embedder::ScopedPlatformHandle platform_handle;
-    MojoResult unwrap_result = mojo::embedder::PassWrappedPlatformHandle(
-        handle_buffer[i], &platform_handle);
-    if (unwrap_result != MOJO_RESULT_OK) {
-      DLOG(WARNING) << "Pipe failed to covert handles. Closing: "
-                    << unwrap_result;
-      CloseWithError(unwrap_result);
-      return;
-    }
-
-    bool ok = message.file_descriptor_set()->Add(platform_handle.release().fd);
-    DCHECK(ok);
-  }
-#else
-  DCHECK(handle_buffer.empty());
-#endif
-
-  message.TraceMessageEnd();
-  owner_->OnMessageReceived(message);
-}
-
-void ChannelMojo::MessageReader::OnPipeClosed() {
-  if (!owner_)
-    return;
-  owner_->OnPipeClosed(this);
-  owner_ = NULL;
-}
-
-void ChannelMojo::MessageReader::OnPipeError(MojoResult error) {
-  if (!owner_)
-    return;
-  owner_->OnPipeError(this);
-}
-
-bool ChannelMojo::MessageReader::Send(scoped_ptr<Message> message) {
-  DCHECK(IsValid());
-
-  message->TraceMessageBegin();
-  std::vector<MojoHandle> handles;
-#if defined(OS_POSIX) && !defined(OS_NACL)
-  // We dup() the handles in IPC::Message to transmit.
-  // IPC::FileDescriptorSet has intricate lifecycle semantics
-  // of FDs, so just to dup()-and-own them is the safest option.
-  if (message->HasFileDescriptors()) {
-    FileDescriptorSet* fdset = message->file_descriptor_set();
-    for (size_t i = 0; i < fdset->size(); ++i) {
-      int fd_to_send = dup(fdset->GetDescriptorAt(i));
-      if (-1 == fd_to_send) {
-        DPLOG(WARNING) << "Failed to dup FD to transmit.";
-        std::for_each(handles.begin(), handles.end(), &MojoClose);
-        CloseWithError(MOJO_RESULT_UNKNOWN);
-        return false;
-      }
-
-      MojoHandle wrapped_handle;
-      MojoResult wrap_result = CreatePlatformHandleWrapper(
-          mojo::embedder::ScopedPlatformHandle(
-              mojo::embedder::PlatformHandle(fd_to_send)),
-          &wrapped_handle);
-      if (MOJO_RESULT_OK != wrap_result) {
-        DLOG(WARNING) << "Pipe failed to wrap handles. Closing: "
-                      << wrap_result;
-        std::for_each(handles.begin(), handles.end(), &MojoClose);
-        CloseWithError(wrap_result);
-        return false;
-      }
-
-      handles.push_back(wrapped_handle);
-    }
-  }
-#endif
-  MojoResult write_result = MojoWriteMessage(
-      handle(),
-      message->data(), static_cast<uint32>(message->size()),
-      handles.empty() ? NULL : &handles[0],
-      static_cast<uint32>(handles.size()),
-      MOJO_WRITE_MESSAGE_FLAG_NONE);
-  if (MOJO_RESULT_OK != write_result) {
-    std::for_each(handles.begin(), handles.end(), &MojoClose);
-    CloseWithError(write_result);
-    return false;
-  }
-
-  return true;
-}
-
-//------------------------------------------------------------------------------
-
-// MessagePipeReader implementation for control messages.
-// Actual message handling is implemented by sublcasses.
-class ChannelMojo::ControlReader : public internal::MessagePipeReader {
- public:
-  ControlReader(mojo::ScopedMessagePipeHandle pipe, ChannelMojo* owner)
-      : internal::MessagePipeReader(pipe.Pass()),
-        owner_(owner) {}
-
-  virtual bool Connect() { return true; }
-  virtual void OnPipeClosed() OVERRIDE;
-  virtual void OnPipeError(MojoResult error) OVERRIDE;
-
- protected:
-  ChannelMojo* owner_;
-};
-
-void ChannelMojo::ControlReader::OnPipeClosed() {
-  if (!owner_)
-    return;
-  owner_->OnPipeClosed(this);
-  owner_ = NULL;
-}
-
-void ChannelMojo::ControlReader::OnPipeError(MojoResult error) {
-  if (!owner_)
-    return;
-  owner_->OnPipeError(this);
-}
-
-//------------------------------------------------------------------------------
-
-// ControlReader for server-side ChannelMojo.
-class ChannelMojo::ServerControlReader : public ChannelMojo::ControlReader {
- public:
-  ServerControlReader(mojo::ScopedMessagePipeHandle pipe, ChannelMojo* owner)
-      : ControlReader(pipe.Pass(), owner) { }
-
-  virtual bool Connect() OVERRIDE;
-  virtual void OnMessageReceived() OVERRIDE;
-
- private:
-  MojoResult SendHelloRequest();
-  MojoResult RespondHelloResponse();
-
-  mojo::ScopedMessagePipeHandle message_pipe_;
-};
-
-bool ChannelMojo::ServerControlReader::Connect() {
-  MojoResult result = SendHelloRequest();
-  if (result != MOJO_RESULT_OK) {
-    CloseWithError(result);
-    return false;
-  }
-
-  return true;
-}
-
-MojoResult ChannelMojo::ServerControlReader::SendHelloRequest() {
-  DCHECK(IsValid());
-  DCHECK(!message_pipe_.is_valid());
-
-  mojo::ScopedMessagePipeHandle self;
-  mojo::ScopedMessagePipeHandle peer;
-  MojoResult create_result = mojo::CreateMessagePipe(
-      NULL, &message_pipe_, &peer);
-  if (MOJO_RESULT_OK != create_result) {
-    DLOG(WARNING) << "mojo::CreateMessagePipe failed: " << create_result;
-    return create_result;
-  }
-
-  MojoHandle peer_to_send = peer.get().value();
-  Pickle request = HelloMessage::CreateRequest(owner_->GetSelfPID());
-  MojoResult write_result = MojoWriteMessage(
-      handle(),
-      request.data(), static_cast<uint32>(request.size()),
-      &peer_to_send, 1,
-      MOJO_WRITE_MESSAGE_FLAG_NONE);
-  if (MOJO_RESULT_OK != write_result) {
-    DLOG(WARNING) << "Writing Hello request failed: " << create_result;
-    return write_result;
-  }
-
-  // |peer| is sent and no longer owned by |this|.
-  (void)peer.release();
-  return MOJO_RESULT_OK;
-}
-
-MojoResult ChannelMojo::ServerControlReader::RespondHelloResponse() {
-  Pickle request(data_buffer().empty() ? "" : &data_buffer()[0],
-                 static_cast<uint32>(data_buffer().size()));
-
-  int32 read_pid = 0;
-  if (!HelloMessage::ReadResponse(request, &read_pid)) {
-    DLOG(ERROR) << "Failed to parse Hello response.";
-    return MOJO_RESULT_UNKNOWN;
-  }
-
-  base::ProcessId pid = static_cast<base::ProcessId>(read_pid);
-  owner_->set_peer_pid(pid);
-  owner_->OnConnected(message_pipe_.Pass());
-  return MOJO_RESULT_OK;
-}
-
-void ChannelMojo::ServerControlReader::OnMessageReceived() {
-  MojoResult result = RespondHelloResponse();
-  if (result != MOJO_RESULT_OK)
-    CloseWithError(result);
-}
-
-//------------------------------------------------------------------------------
-
-// ControlReader for client-side ChannelMojo.
-class ChannelMojo::ClientControlReader : public ChannelMojo::ControlReader {
- public:
-  ClientControlReader(mojo::ScopedMessagePipeHandle pipe, ChannelMojo* owner)
-      : ControlReader(pipe.Pass(), owner) {}
-
-  virtual void OnMessageReceived() OVERRIDE;
-
- private:
-  MojoResult RespondHelloRequest(MojoHandle message_channel);
-};
-
-MojoResult ChannelMojo::ClientControlReader::RespondHelloRequest(
-    MojoHandle message_channel) {
-  DCHECK(IsValid());
-
-  mojo::ScopedMessagePipeHandle received_pipe(
-      (mojo::MessagePipeHandle(message_channel)));
-
-  int32 read_request = 0;
-  Pickle request(data_buffer().empty() ? "" : &data_buffer()[0],
-                 static_cast<uint32>(data_buffer().size()));
-  if (!HelloMessage::ReadRequest(request, &read_request)) {
-    DLOG(ERROR) << "Hello request has wrong magic.";
-    return MOJO_RESULT_UNKNOWN;
-  }
-
-  base::ProcessId pid = read_request;
-  Pickle response = HelloMessage::CreateResponse(owner_->GetSelfPID());
-  MojoResult write_result = MojoWriteMessage(
-      handle(),
-      response.data(), static_cast<uint32>(response.size()),
-      NULL, 0,
-      MOJO_WRITE_MESSAGE_FLAG_NONE);
-  if (MOJO_RESULT_OK != write_result) {
-    DLOG(ERROR) << "Writing Hello response failed: " << write_result;
-    return write_result;
-  }
-
-  owner_->set_peer_pid(pid);
-  owner_->OnConnected(received_pipe.Pass());
-  return MOJO_RESULT_OK;
-}
-
-void ChannelMojo::ClientControlReader::OnMessageReceived() {
-  std::vector<MojoHandle> handle_buffer;
-  TakeHandleBuffer(&handle_buffer);
-  if (handle_buffer.size() != 1) {
-    DLOG(ERROR) << "Hello request doesn't contains required handle: "
-                << handle_buffer.size();
-    CloseWithError(MOJO_RESULT_UNKNOWN);
-    return;
-  }
-
-  MojoResult result = RespondHelloRequest(handle_buffer[0]);
-  if (result != MOJO_RESULT_OK) {
-    DLOG(ERROR) << "Failed to respond Hello request. Closing: "
-                << result;
-    CloseWithError(result);
-  }
-}
-
-//------------------------------------------------------------------------------
-
 void ChannelMojo::ChannelInfoDeleter::operator()(
     mojo::embedder::ChannelInfo* ptr) const {
   mojo::embedder::DestroyChannelOnIOThread(ptr);
@@ -458,9 +98,8 @@ void ChannelMojo::ChannelInfoDeleter::operator()(
 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
@@ -473,11 +112,14 @@ scoped_ptr<ChannelFactory> ChannelMojo::CreateFactory(
           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(const ChannelHandle& channel_handle,
+                         Mode mode,
+                         Listener* listener,
+                         scoped_refptr<base::TaskRunner> io_thread_task_runner)
+    : bootstrap_(
+          Channel::Create(channel_handle, mode, g_null_listener.Pointer())),
+      mode_(mode),
+      listener_(listener),
       peer_pid_(base::kNullProcessId),
       weak_factory_(this) {
   if (base::MessageLoopProxy::current() == io_thread_task_runner.get()) {
@@ -504,10 +146,12 @@ void ChannelMojo::InitOnIOThread() {
 
   switch (mode_) {
     case MODE_SERVER:
-      control_reader_.reset(new ServerControlReader(control_pipe.Pass(), this));
+      control_reader_.reset(
+          new internal::ServerControlReader(control_pipe.Pass(), this));
       break;
     case MODE_CLIENT:
-      control_reader_.reset(new ClientControlReader(control_pipe.Pass(), this));
+      control_reader_.reset(
+          new internal::ClientControlReader(control_pipe.Pass(), this));
       break;
     default:
       NOTREACHED();
@@ -527,7 +171,8 @@ void ChannelMojo::Close() {
 }
 
 void ChannelMojo::OnConnected(mojo::ScopedMessagePipeHandle pipe) {
-  message_reader_ = make_scoped_ptr(new MessageReader(pipe.Pass(), this));
+  message_reader_ =
+      make_scoped_ptr(new internal::MessageReader(pipe.Pass(), this));
 
   for (size_t i = 0; i < pending_messages_.size(); ++i) {
     message_reader_->Send(make_scoped_ptr(pending_messages_[i]));
@@ -583,6 +228,62 @@ int ChannelMojo::GetClientFileDescriptor() const {
 int ChannelMojo::TakeClientFileDescriptor() {
   return bootstrap_->TakeClientFileDescriptor();
 }
+
+// static
+MojoResult ChannelMojo::WriteToFileDescriptorSet(
+    const std::vector<MojoHandle>& handle_buffer,
+    Message* message) {
+  for (size_t i = 0; i < handle_buffer.size(); ++i) {
+    mojo::embedder::ScopedPlatformHandle platform_handle;
+    MojoResult unwrap_result = mojo::embedder::PassWrappedPlatformHandle(
+        handle_buffer[i], &platform_handle);
+    if (unwrap_result != MOJO_RESULT_OK) {
+      DLOG(WARNING) << "Pipe failed to covert handles. Closing: "
+                    << unwrap_result;
+      return unwrap_result;
+    }
+
+    bool ok = message->file_descriptor_set()->Add(platform_handle.release().fd);
+    DCHECK(ok);
+  }
+
+  return MOJO_RESULT_OK;
+}
+
+// static
+MojoResult ChannelMojo::ReadFromFileDescriptorSet(
+    const Message& message,
+    std::vector<MojoHandle>* handles) {
+  // We dup() the handles in IPC::Message to transmit.
+  // IPC::FileDescriptorSet has intricate lifecycle semantics
+  // of FDs, so just to dup()-and-own them is the safest option.
+  if (message.HasFileDescriptors()) {
+    const FileDescriptorSet* fdset = message.file_descriptor_set();
+    for (size_t i = 0; i < fdset->size(); ++i) {
+      int fd_to_send = dup(fdset->GetDescriptorAt(i));
+      if (-1 == fd_to_send) {
+        DPLOG(WARNING) << "Failed to dup FD to transmit.";
+        return MOJO_RESULT_UNKNOWN;
+      }
+
+      MojoHandle wrapped_handle;
+      MojoResult wrap_result = CreatePlatformHandleWrapper(
+          mojo::embedder::ScopedPlatformHandle(
+              mojo::embedder::PlatformHandle(fd_to_send)),
+          &wrapped_handle);
+      if (MOJO_RESULT_OK != wrap_result) {
+        DLOG(WARNING) << "Pipe failed to wrap handles. Closing: "
+                      << wrap_result;
+        return wrap_result;
+      }
+
+      handles->push_back(wrapped_handle);
+    }
+  }
+
+  return MOJO_RESULT_OK;
+}
+
 #endif  // defined(OS_POSIX) && !defined(OS_NACL)
 
 }  // namespace IPC
diff --git a/ipc/mojo/ipc_channel_mojo.h b/ipc/mojo/ipc_channel_mojo.h
index dd57bc8..22c56a9 100644
--- a/ipc/mojo/ipc_channel_mojo.h
+++ b/ipc/mojo/ipc_channel_mojo.h
@@ -24,6 +24,13 @@ struct ChannelInfo;
 
 namespace IPC {
 
+namespace internal {
+class ControlReader;
+class ServerControlReader;
+class ClientControlReader;
+class MessageReader;
+}
+
 // Mojo-based IPC::Channel implementation over a platform handle.
 //
 // ChannelMojo builds Mojo MessagePipe using underlying pipe given by
@@ -74,6 +81,15 @@ class IPC_MOJO_EXPORT ChannelMojo : public Channel {
 #if defined(OS_POSIX) && !defined(OS_NACL)
   virtual int GetClientFileDescriptor() const OVERRIDE;
   virtual int TakeClientFileDescriptor() OVERRIDE;
+
+  // These access protected API of IPC::Message, which has ChannelMojo
+  // as a friend class.
+  static MojoResult WriteToFileDescriptorSet(
+      const std::vector<MojoHandle>& handle_buffer,
+      Message* message);
+  static MojoResult ReadFromFileDescriptorSet(const Message& message,
+                                              std::vector<MojoHandle>* handles);
+
 #endif  // defined(OS_POSIX) && !defined(OS_NACL)
 
   // Called from MessagePipeReader implementations
@@ -83,6 +99,12 @@ class IPC_MOJO_EXPORT ChannelMojo : public Channel {
   void OnPipeError(internal::MessagePipeReader* reader);
   void set_peer_pid(base::ProcessId pid) { peer_pid_ = pid; }
 
+ protected:
+  ChannelMojo(const ChannelHandle& channel_handle,
+              Mode mode,
+              Listener* listener,
+              scoped_refptr<base::TaskRunner> io_thread_task_runner);
+
  private:
   struct ChannelInfoDeleter {
     void operator()(mojo::embedder::ChannelInfo* ptr) const;
@@ -93,14 +115,6 @@ class IPC_MOJO_EXPORT ChannelMojo : public Channel {
   // notifications invoked by them.
   typedef internal::MessagePipeReader::DelayedDeleter ReaderDeleter;
 
-  class ControlReader;
-  class ServerControlReader;
-  class ClientControlReader;
-  class MessageReader;
-
-  ChannelMojo(scoped_ptr<Channel> bootstrap, Mode mode, Listener* listener,
-              scoped_refptr<base::TaskRunner> io_thread_task_runner);
-
   void InitOnIOThread();
 
   scoped_ptr<Channel> bootstrap_;
@@ -110,8 +124,8 @@ class IPC_MOJO_EXPORT ChannelMojo : public Channel {
   scoped_ptr<mojo::embedder::ChannelInfo,
              ChannelInfoDeleter> channel_info_;
 
-  scoped_ptr<ControlReader, ReaderDeleter> control_reader_;
-  scoped_ptr<MessageReader, ReaderDeleter> message_reader_;
+  scoped_ptr<internal::ControlReader, ReaderDeleter> control_reader_;
+  scoped_ptr<internal::MessageReader, ReaderDeleter> message_reader_;
   ScopedVector<Message> pending_messages_;
 
   base::WeakPtrFactory<ChannelMojo> weak_factory_;
diff --git a/ipc/mojo/ipc_channel_mojo_readers.cc b/ipc/mojo/ipc_channel_mojo_readers.cc
new file mode 100644
index 0000000..7af96f8
--- /dev/null
+++ b/ipc/mojo/ipc_channel_mojo_readers.cc
@@ -0,0 +1,321 @@
+// 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_readers.h"
+
+#include "ipc/mojo/ipc_channel_mojo.h"
+#include "mojo/embedder/embedder.h"
+
+#if defined(OS_POSIX) && !defined(OS_NACL)
+#include "ipc/file_descriptor_set_posix.h"
+#endif
+
+namespace IPC {
+namespace internal {
+
+namespace {
+
+// 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);
+    return request;
+  }
+
+  static bool ReadRequest(Pickle& pickle, int32* pid) {
+    PickleIterator iter(pickle);
+    std::string hello;
+    if (!iter.ReadString(&hello)) {
+      DLOG(WARNING) << "Failed to Read magic string.";
+      return false;
+    }
+
+    if (hello != kHelloRequestMagic) {
+      DLOG(WARNING) << "Magic mismatch:" << hello;
+      return false;
+    }
+
+    int read_pid;
+    if (!iter.ReadInt(&read_pid)) {
+      DLOG(WARNING) << "Failed to Read PID.";
+      return false;
+    }
+
+    *pid = read_pid;
+    return true;
+  }
+
+  static Pickle CreateResponse(int32 pid) {
+    Pickle request;
+    request.WriteString(kHelloResponseMagic);
+    request.WriteInt(pid);
+    return request;
+  }
+
+  static bool ReadResponse(Pickle& pickle, int32* pid) {
+    PickleIterator iter(pickle);
+    std::string hello;
+    if (!iter.ReadString(&hello)) {
+      DLOG(WARNING) << "Failed to read magic string.";
+      return false;
+    }
+
+    if (hello != kHelloResponseMagic) {
+      DLOG(WARNING) << "Magic mismatch:" << hello;
+      return false;
+    }
+
+    int read_pid;
+    if (!iter.ReadInt(&read_pid)) {
+      DLOG(WARNING) << "Failed to read PID.";
+      return false;
+    }
+
+    *pid = read_pid;
+    return true;
+  }
+
+ private:
+  static const char* kHelloRequestMagic;
+  static const char* kHelloResponseMagic;
+};
+
+const char* HelloMessage::kHelloRequestMagic = "MREQ";
+const char* HelloMessage::kHelloResponseMagic = "MRES";
+
+}  // namespace
+
+//------------------------------------------------------------------------------
+
+MessageReader::MessageReader(mojo::ScopedMessagePipeHandle pipe,
+                             ChannelMojo* owner)
+    : internal::MessagePipeReader(pipe.Pass()), owner_(owner) {
+}
+
+void MessageReader::OnMessageReceived() {
+  Message message(data_buffer().empty() ? "" : &data_buffer()[0],
+                  static_cast<uint32>(data_buffer().size()));
+
+  std::vector<MojoHandle> handle_buffer;
+  TakeHandleBuffer(&handle_buffer);
+#if defined(OS_POSIX) && !defined(OS_NACL)
+  MojoResult write_result =
+      ChannelMojo::WriteToFileDescriptorSet(handle_buffer, &message);
+  if (write_result != MOJO_RESULT_OK) {
+    CloseWithError(write_result);
+    return;
+  }
+#else
+  DCHECK(handle_buffer.empty());
+#endif
+
+  message.TraceMessageEnd();
+  owner_->OnMessageReceived(message);
+}
+
+void MessageReader::OnPipeClosed() {
+  if (!owner_)
+    return;
+  owner_->OnPipeClosed(this);
+  owner_ = NULL;
+}
+
+void MessageReader::OnPipeError(MojoResult error) {
+  if (!owner_)
+    return;
+  owner_->OnPipeError(this);
+}
+
+bool MessageReader::Send(scoped_ptr<Message> message) {
+  DCHECK(IsValid());
+
+  message->TraceMessageBegin();
+  std::vector<MojoHandle> handles;
+#if defined(OS_POSIX) && !defined(OS_NACL)
+  MojoResult read_result =
+      ChannelMojo::ReadFromFileDescriptorSet(*message, &handles);
+  if (read_result != MOJO_RESULT_OK) {
+    std::for_each(handles.begin(), handles.end(), &MojoClose);
+    CloseWithError(read_result);
+    return false;
+  }
+#endif
+  MojoResult write_result =
+      MojoWriteMessage(handle(),
+                       message->data(),
+                       static_cast<uint32>(message->size()),
+                       handles.empty() ? NULL : &handles[0],
+                       static_cast<uint32>(handles.size()),
+                       MOJO_WRITE_MESSAGE_FLAG_NONE);
+  if (MOJO_RESULT_OK != write_result) {
+    std::for_each(handles.begin(), handles.end(), &MojoClose);
+    CloseWithError(write_result);
+    return false;
+  }
+
+  return true;
+}
+
+//------------------------------------------------------------------------------
+
+ControlReader::ControlReader(mojo::ScopedMessagePipeHandle pipe,
+                             ChannelMojo* owner)
+    : internal::MessagePipeReader(pipe.Pass()), owner_(owner) {
+}
+
+void ControlReader::OnPipeClosed() {
+  if (!owner_)
+    return;
+  owner_->OnPipeClosed(this);
+  owner_ = NULL;
+}
+
+void ControlReader::OnPipeError(MojoResult error) {
+  if (!owner_)
+    return;
+  owner_->OnPipeError(this);
+}
+
+bool ControlReader::Connect() {
+  return true;
+}
+
+//------------------------------------------------------------------------------
+
+ServerControlReader::ServerControlReader(mojo::ScopedMessagePipeHandle pipe,
+                                         ChannelMojo* owner)
+    : ControlReader(pipe.Pass(), owner) {
+}
+
+ServerControlReader::~ServerControlReader() {
+}
+
+bool ServerControlReader::Connect() {
+  MojoResult result = SendHelloRequest();
+  if (result != MOJO_RESULT_OK) {
+    CloseWithError(result);
+    return false;
+  }
+
+  return true;
+}
+
+MojoResult ServerControlReader::SendHelloRequest() {
+  DCHECK(IsValid());
+  DCHECK(!message_pipe_.is_valid());
+
+  mojo::ScopedMessagePipeHandle self;
+  mojo::ScopedMessagePipeHandle peer;
+  MojoResult create_result =
+      mojo::CreateMessagePipe(NULL, &message_pipe_, &peer);
+  if (MOJO_RESULT_OK != create_result) {
+    DLOG(WARNING) << "mojo::CreateMessagePipe failed: " << create_result;
+    return create_result;
+  }
+
+  MojoHandle peer_to_send = peer.get().value();
+  Pickle request = HelloMessage::CreateRequest(owner_->GetSelfPID());
+  MojoResult write_result =
+      MojoWriteMessage(handle(),
+                       request.data(),
+                       static_cast<uint32>(request.size()),
+                       &peer_to_send,
+                       1,
+                       MOJO_WRITE_MESSAGE_FLAG_NONE);
+  if (MOJO_RESULT_OK != write_result) {
+    DLOG(WARNING) << "Writing Hello request failed: " << create_result;
+    return write_result;
+  }
+
+  // |peer| is sent and no longer owned by |this|.
+  (void)peer.release();
+  return MOJO_RESULT_OK;
+}
+
+MojoResult ServerControlReader::RespondHelloResponse() {
+  Pickle request(data_buffer().empty() ? "" : &data_buffer()[0],
+                 static_cast<uint32>(data_buffer().size()));
+
+  int32 read_pid = 0;
+  if (!HelloMessage::ReadResponse(request, &read_pid)) {
+    DLOG(ERROR) << "Failed to parse Hello response.";
+    return MOJO_RESULT_UNKNOWN;
+  }
+
+  base::ProcessId pid = static_cast<base::ProcessId>(read_pid);
+  owner_->set_peer_pid(pid);
+  owner_->OnConnected(message_pipe_.Pass());
+  return MOJO_RESULT_OK;
+}
+
+void ServerControlReader::OnMessageReceived() {
+  MojoResult result = RespondHelloResponse();
+  if (result != MOJO_RESULT_OK)
+    CloseWithError(result);
+}
+
+//------------------------------------------------------------------------------
+
+ClientControlReader::ClientControlReader(mojo::ScopedMessagePipeHandle pipe,
+                                         ChannelMojo* owner)
+    : ControlReader(pipe.Pass(), owner) {
+}
+
+MojoResult ClientControlReader::RespondHelloRequest(
+    MojoHandle message_channel) {
+  DCHECK(IsValid());
+
+  mojo::ScopedMessagePipeHandle received_pipe(
+      (mojo::MessagePipeHandle(message_channel)));
+
+  int32 read_request = 0;
+  Pickle request(data_buffer().empty() ? "" : &data_buffer()[0],
+                 static_cast<uint32>(data_buffer().size()));
+  if (!HelloMessage::ReadRequest(request, &read_request)) {
+    DLOG(ERROR) << "Hello request has wrong magic.";
+    return MOJO_RESULT_UNKNOWN;
+  }
+
+  base::ProcessId pid = read_request;
+  Pickle response = HelloMessage::CreateResponse(owner_->GetSelfPID());
+  MojoResult write_result =
+      MojoWriteMessage(handle(),
+                       response.data(),
+                       static_cast<uint32>(response.size()),
+                       NULL,
+                       0,
+                       MOJO_WRITE_MESSAGE_FLAG_NONE);
+  if (MOJO_RESULT_OK != write_result) {
+    DLOG(ERROR) << "Writing Hello response failed: " << write_result;
+    return write_result;
+  }
+
+  owner_->set_peer_pid(pid);
+  owner_->OnConnected(received_pipe.Pass());
+  return MOJO_RESULT_OK;
+}
+
+void ClientControlReader::OnMessageReceived() {
+  std::vector<MojoHandle> handle_buffer;
+  TakeHandleBuffer(&handle_buffer);
+  if (handle_buffer.size() != 1) {
+    DLOG(ERROR) << "Hello request doesn't contains required handle: "
+                << handle_buffer.size();
+    CloseWithError(MOJO_RESULT_UNKNOWN);
+    return;
+  }
+
+  MojoResult result = RespondHelloRequest(handle_buffer[0]);
+  if (result != MOJO_RESULT_OK) {
+    DLOG(ERROR) << "Failed to respond Hello request. Closing: " << result;
+    CloseWithError(result);
+  }
+}
+
+}  // namespace internal
+}  // namespace IPC
diff --git a/ipc/mojo/ipc_channel_mojo_readers.h b/ipc/mojo/ipc_channel_mojo_readers.h
new file mode 100644
index 0000000..13ddead
--- /dev/null
+++ b/ipc/mojo/ipc_channel_mojo_readers.h
@@ -0,0 +1,102 @@
+// 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.
+
+#ifndef IPC_MOJO_IPC_CHANNEL_MOJO_READERS_H_
+#define IPC_MOJO_IPC_CHANNEL_MOJO_READERS_H_
+
+#include <vector>
+
+#include "base/macros.h"
+#include "base/memory/scoped_ptr.h"
+#include "ipc/mojo/ipc_message_pipe_reader.h"
+#include "mojo/public/cpp/system/core.h"
+
+namespace mojo {
+namespace embedder {
+struct ChannelInfo;
+}
+}
+
+namespace IPC {
+
+class ChannelMojo;
+class Message;
+
+namespace internal {
+
+// A MessagePipeReader implementation for IPC::Message communication.
+class MessageReader : public MessagePipeReader {
+ public:
+  MessageReader(mojo::ScopedMessagePipeHandle pipe, ChannelMojo* owner);
+
+  bool Send(scoped_ptr<Message> message);
+
+  // MessagePipeReader implementation
+  virtual void OnMessageReceived() OVERRIDE;
+  virtual void OnPipeClosed() OVERRIDE;
+  virtual void OnPipeError(MojoResult error) OVERRIDE;
+
+ private:
+  ChannelMojo* owner_;
+
+  DISALLOW_COPY_AND_ASSIGN(MessageReader);
+};
+
+// MessagePipeReader implementation for control messages.
+// Actual message handling is implemented by sublcasses.
+class ControlReader : public MessagePipeReader {
+ public:
+  ControlReader(mojo::ScopedMessagePipeHandle pipe, ChannelMojo* owner);
+
+  virtual bool Connect();
+
+  // MessagePipeReader implementation
+  virtual void OnPipeClosed() OVERRIDE;
+  virtual void OnPipeError(MojoResult error) OVERRIDE;
+
+ protected:
+  ChannelMojo* owner_;
+
+  DISALLOW_COPY_AND_ASSIGN(ControlReader);
+};
+
+// ControlReader for server-side ChannelMojo.
+class ServerControlReader : public ControlReader {
+ public:
+  ServerControlReader(mojo::ScopedMessagePipeHandle pipe, ChannelMojo* owner);
+  virtual ~ServerControlReader();
+
+  // ControlReader override
+  virtual bool Connect() OVERRIDE;
+
+  // MessagePipeReader implementation
+  virtual void OnMessageReceived() OVERRIDE;
+
+ private:
+  MojoResult SendHelloRequest();
+  MojoResult RespondHelloResponse();
+
+  mojo::ScopedMessagePipeHandle message_pipe_;
+
+  DISALLOW_COPY_AND_ASSIGN(ServerControlReader);
+};
+
+// ControlReader for client-side ChannelMojo.
+class ClientControlReader : public ControlReader {
+ public:
+  ClientControlReader(mojo::ScopedMessagePipeHandle pipe, ChannelMojo* owner);
+
+  // MessagePipeReader implementation
+  virtual void OnMessageReceived() OVERRIDE;
+
+ private:
+  MojoResult RespondHelloRequest(MojoHandle message_channel);
+
+  DISALLOW_COPY_AND_ASSIGN(ClientControlReader);
+};
+
+}  // namespace internal
+}  // namespace IPC
+
+#endif  // IPC_MOJO_IPC_CHANNEL_MOJO_READERS_H_
diff --git a/ipc/mojo/ipc_mojo.gyp b/ipc/mojo/ipc_mojo.gyp
index f01d48f..c4ead0a 100644
--- a/ipc/mojo/ipc_mojo.gyp
+++ b/ipc/mojo/ipc_mojo.gyp
@@ -28,6 +28,8 @@
       'sources': [
         'ipc_channel_mojo.cc',
         'ipc_channel_mojo.h',
+        'ipc_channel_mojo_readers.cc',
+        'ipc_channel_mojo_readers.h',
         'ipc_message_pipe_reader.cc',
         'ipc_message_pipe_reader.h',
       ],