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}

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",
   ]

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_;

ipc/mojo/ipc_channel_mojo_readers.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_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

ipc/mojo/ipc_channel_mojo_readers.h

+// 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_

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',
       ],