ChannelMojo Refactoring: Merge MessageReader into MessagePipeReader

MessageReader was the sole subclass  of MessageReader.
There is no need to have two classes.
This CL also decouple ChannelMojo from MessagePipeReader
through a delegate interface.

R=viettrungluu@chromium.org
BUG=377980

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

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

9 files changed, 141 insertions, 256 deletions

diff --git a/ipc/mojo/BUILD.gn b/ipc/mojo/BUILD.gn
index 324cc9a..45ffb89 100644
--- a/ipc/mojo/BUILD.gn
+++ b/ipc/mojo/BUILD.gn
@@ -17,8 +17,6 @@ component("mojo") {
     "ipc_channel_mojo.h",
     "ipc_channel_mojo_host.cc",
     "ipc_channel_mojo_host.h",
-    "ipc_channel_mojo_readers.cc",
-    "ipc_channel_mojo_readers.h",
     "ipc_mojo_bootstrap.cc",
     "ipc_mojo_bootstrap.h",
     "ipc_message_pipe_reader.cc",
diff --git a/ipc/mojo/ipc_channel_mojo.cc b/ipc/mojo/ipc_channel_mojo.cc
index 66f232d..351b27a 100644
--- a/ipc/mojo/ipc_channel_mojo.cc
+++ b/ipc/mojo/ipc_channel_mojo.cc
@@ -9,7 +9,6 @@
 #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"
@@ -279,7 +278,7 @@ void ChannelMojo::OnBootstrapError() {
 void ChannelMojo::InitMessageReader(mojo::ScopedMessagePipeHandle pipe,
                                     int32_t peer_pid) {
   message_reader_ =
-      make_scoped_ptr(new internal::MessageReader(pipe.Pass(), this));
+      make_scoped_ptr(new internal::MessagePipeReader(pipe.Pass(), this));
 
   for (size_t i = 0; i < pending_messages_.size(); ++i) {
     bool sent = message_reader_->Send(make_scoped_ptr(pending_messages_[i]));
diff --git a/ipc/mojo/ipc_channel_mojo.h b/ipc/mojo/ipc_channel_mojo.h
index 29d5f29..1ef32c6 100644
--- a/ipc/mojo/ipc_channel_mojo.h
+++ b/ipc/mojo/ipc_channel_mojo.h
@@ -20,13 +20,6 @@
 
 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
@@ -51,8 +44,10 @@ class MessageReader;
 // TODO(morrita): Add APIs to create extra MessagePipes to let
 //                Mojo-based objects talk over this Channel.
 //
-class IPC_MOJO_EXPORT ChannelMojo : public Channel,
-                                    public MojoBootstrap::Delegate {
+class IPC_MOJO_EXPORT ChannelMojo
+    : public Channel,
+      public MojoBootstrap::Delegate,
+      public NON_EXPORTED_BASE(internal::MessagePipeReader::Delegate) {
  public:
   class Delegate {
    public:
@@ -111,10 +106,10 @@ class IPC_MOJO_EXPORT ChannelMojo : public Channel,
   // MojoBootstrapDelegate implementation
   void OnBootstrapError() override;
 
-  // Called from MessagePipeReader implementations
-  void OnMessageReceived(Message& message);
-  void OnPipeClosed(internal::MessagePipeReader* reader);
-  void OnPipeError(internal::MessagePipeReader* reader);
+  // MessagePipeReader::Delegate
+  void OnMessageReceived(Message& message) override;
+  void OnPipeClosed(internal::MessagePipeReader* reader) override;
+  void OnPipeError(internal::MessagePipeReader* reader) override;
 
  protected:
   ChannelMojo(Delegate* delegate,
@@ -149,7 +144,7 @@ class IPC_MOJO_EXPORT ChannelMojo : public Channel,
   scoped_ptr<mojo::embedder::ChannelInfo,
              ChannelInfoDeleter> channel_info_;
 
-  scoped_ptr<internal::MessageReader, ReaderDeleter> message_reader_;
+  scoped_ptr<internal::MessagePipeReader, 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
deleted file mode 100644
index e4a35c5..0000000
--- a/ipc/mojo/ipc_channel_mojo_readers.cc
+++ /dev/null
@@ -1,84 +0,0 @@
-// 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"
-
-namespace IPC {
-namespace internal {
-
-//------------------------------------------------------------------------------
-
-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.get(), &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;
-}
-
-}  // namespace internal
-}  // namespace IPC
diff --git a/ipc/mojo/ipc_channel_mojo_readers.h b/ipc/mojo/ipc_channel_mojo_readers.h
deleted file mode 100644
index ffcc08b..0000000
--- a/ipc/mojo/ipc_channel_mojo_readers.h
+++ /dev/null
@@ -1,96 +0,0 @@
-// 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 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
-  void OnMessageReceived() override;
-  void OnPipeClosed() override;
-  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
-  void OnPipeClosed() override;
-  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);
-  ~ServerControlReader() override;
-
-  // ControlReader override
-  bool Connect() override;
-
-  // MessagePipeReader implementation
-  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
-  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_channel_mojo_unittest.cc b/ipc/mojo/ipc_channel_mojo_unittest.cc
index 07218c6..2442afe 100644
--- a/ipc/mojo/ipc_channel_mojo_unittest.cc
+++ b/ipc/mojo/ipc_channel_mojo_unittest.cc
@@ -14,7 +14,6 @@
 #include "ipc/ipc_test_base.h"
 #include "ipc/ipc_test_channel_listener.h"
 #include "ipc/mojo/ipc_channel_mojo_host.h"
-#include "ipc/mojo/ipc_channel_mojo_readers.h"
 
 #if defined(OS_POSIX)
 #include "base/file_descriptor_posix.h"
diff --git a/ipc/mojo/ipc_message_pipe_reader.cc b/ipc/mojo/ipc_message_pipe_reader.cc
index b0df997..36871e7 100644
--- a/ipc/mojo/ipc_message_pipe_reader.cc
+++ b/ipc/mojo/ipc_message_pipe_reader.cc
@@ -9,14 +9,17 @@
 #include "base/location.h"
 #include "base/logging.h"
 #include "base/message_loop/message_loop_proxy.h"
+#include "ipc/mojo/ipc_channel_mojo.h"
 #include "mojo/public/cpp/environment/environment.h"
 
 namespace IPC {
 namespace internal {
 
-MessagePipeReader::MessagePipeReader(mojo::ScopedMessagePipeHandle handle)
+MessagePipeReader::MessagePipeReader(mojo::ScopedMessagePipeHandle handle,
+                                     MessagePipeReader::Delegate* delegate)
     : pipe_wait_id_(0),
-      pipe_(handle.Pass()) {
+      pipe_(handle.Pass()),
+      delegate_(delegate) {
   StartWaiting();
 }
 
@@ -35,32 +38,99 @@ void MessagePipeReader::CloseWithError(MojoResult error) {
   Close();
 }
 
+bool MessagePipeReader::Send(scoped_ptr<Message> message) {
+  DCHECK(IsValid());
+
+  message->TraceMessageBegin();
+  std::vector<MojoHandle> handles;
+  MojoResult result = MOJO_RESULT_OK;
+#if defined(OS_POSIX) && !defined(OS_NACL)
+  result = ChannelMojo::ReadFromFileDescriptorSet(message.get(), &handles);
+#endif
+  if (result == MOJO_RESULT_OK) {
+    result = MojoWriteMessage(handle(),
+                              message->data(),
+                              static_cast<uint32>(message->size()),
+                              handles.empty() ? nullptr : &handles[0],
+                              static_cast<uint32>(handles.size()),
+                              MOJO_WRITE_MESSAGE_FLAG_NONE);
+  }
+
+  if (result != MOJO_RESULT_OK) {
+    std::for_each(handles.begin(), handles.end(), &MojoClose);
+    CloseWithError(result);
+    return false;
+  }
+
+  return true;
+}
+
 // static
 void MessagePipeReader::InvokePipeIsReady(void* closure, MojoResult result) {
   reinterpret_cast<MessagePipeReader*>(closure)->PipeIsReady(result);
 }
 
-void MessagePipeReader::StartWaiting() {
-  DCHECK(pipe_.is_valid());
-  DCHECK(!pipe_wait_id_);
-  // Not using MOJO_HANDLE_SIGNAL_WRITABLE here, expecting buffer in
-  // MessagePipe.
-  //
-  // TODO(morrita): Should we re-set the signal when we get new
-  // message to send?
-  pipe_wait_id_ = mojo::Environment::GetDefaultAsyncWaiter()->AsyncWait(
-      pipe_.get().value(),
-      MOJO_HANDLE_SIGNAL_READABLE,
-      MOJO_DEADLINE_INDEFINITE,
-      &InvokePipeIsReady,
-      this);
+void MessagePipeReader::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();
+  delegate_->OnMessageReceived(message);
 }
 
-void MessagePipeReader::StopWaiting() {
-  if (!pipe_wait_id_)
+void MessagePipeReader::OnPipeClosed() {
+  if (!delegate_)
     return;
-  mojo::Environment::GetDefaultAsyncWaiter()->CancelWait(pipe_wait_id_);
-  pipe_wait_id_ = 0;
+  delegate_->OnPipeClosed(this);
+  delegate_ = nullptr;
+}
+
+void MessagePipeReader::OnPipeError(MojoResult error) {
+  if (!delegate_)
+    return;
+  delegate_->OnPipeError(this);
+}
+
+MojoResult MessagePipeReader::ReadMessageBytes() {
+  DCHECK(handle_buffer_.empty());
+
+  uint32_t num_bytes = static_cast<uint32_t>(data_buffer_.size());
+  uint32_t num_handles = 0;
+  MojoResult result = MojoReadMessage(pipe_.get().value(),
+                                      num_bytes ? &data_buffer_[0] : nullptr,
+                                      &num_bytes,
+                                      nullptr,
+                                      &num_handles,
+                                      MOJO_READ_MESSAGE_FLAG_NONE);
+  data_buffer_.resize(num_bytes);
+  handle_buffer_.resize(num_handles);
+  if (result == MOJO_RESULT_RESOURCE_EXHAUSTED) {
+    // MOJO_RESULT_RESOURCE_EXHAUSTED was asking the caller that
+    // it needs more bufer. So we re-read it with resized buffers.
+    result = MojoReadMessage(pipe_.get().value(),
+                             num_bytes ? &data_buffer_[0] : nullptr,
+                             &num_bytes,
+                             num_handles ? &handle_buffer_[0] : nullptr,
+                             &num_handles,
+                             MOJO_READ_MESSAGE_FLAG_NONE);
+  }
+
+  DCHECK(0 == num_bytes || data_buffer_.size() == num_bytes);
+  DCHECK(0 == num_handles || handle_buffer_.size() == num_handles);
+  return result;
 }
 
 void MessagePipeReader::PipeIsReady(MojoResult wait_result) {
@@ -104,33 +174,27 @@ void MessagePipeReader::PipeIsReady(MojoResult wait_result) {
     StartWaiting();
 }
 
-MojoResult MessagePipeReader::ReadMessageBytes() {
-  DCHECK(handle_buffer_.empty());
-
-  uint32_t num_bytes = static_cast<uint32_t>(data_buffer_.size());
-  uint32_t num_handles = 0;
-  MojoResult result = MojoReadMessage(pipe_.get().value(),
-                                      num_bytes ? &data_buffer_[0] : NULL,
-                                      &num_bytes,
-                                      NULL,
-                                      &num_handles,
-                                      MOJO_READ_MESSAGE_FLAG_NONE);
-  data_buffer_.resize(num_bytes);
-  handle_buffer_.resize(num_handles);
-  if (result == MOJO_RESULT_RESOURCE_EXHAUSTED) {
-    // MOJO_RESULT_RESOURCE_EXHAUSTED was asking the caller that
-    // it needs more bufer. So we re-read it with resized buffers.
-    result = MojoReadMessage(pipe_.get().value(),
-                             num_bytes ? &data_buffer_[0] : NULL,
-                             &num_bytes,
-                             num_handles ? &handle_buffer_[0] : NULL,
-                             &num_handles,
-                             MOJO_READ_MESSAGE_FLAG_NONE);
-  }
+void MessagePipeReader::StartWaiting() {
+  DCHECK(pipe_.is_valid());
+  DCHECK(!pipe_wait_id_);
+  // Not using MOJO_HANDLE_SIGNAL_WRITABLE here, expecting buffer in
+  // MessagePipe.
+  //
+  // TODO(morrita): Should we re-set the signal when we get new
+  // message to send?
+  pipe_wait_id_ = mojo::Environment::GetDefaultAsyncWaiter()->AsyncWait(
+      pipe_.get().value(),
+      MOJO_HANDLE_SIGNAL_READABLE,
+      MOJO_DEADLINE_INDEFINITE,
+      &InvokePipeIsReady,
+      this);
+}
 
-  DCHECK(0 == num_bytes || data_buffer_.size() == num_bytes);
-  DCHECK(0 == num_handles || handle_buffer_.size() == num_handles);
-  return result;
+void MessagePipeReader::StopWaiting() {
+  if (!pipe_wait_id_)
+    return;
+  mojo::Environment::GetDefaultAsyncWaiter()->CancelWait(pipe_wait_id_);
+  pipe_wait_id_ = 0;
 }
 
 void MessagePipeReader::DelayedDeleter::operator()(
diff --git a/ipc/mojo/ipc_message_pipe_reader.h b/ipc/mojo/ipc_message_pipe_reader.h
index ecfa018..ee1d3b5 100644
--- a/ipc/mojo/ipc_message_pipe_reader.h
+++ b/ipc/mojo/ipc_message_pipe_reader.h
@@ -7,7 +7,9 @@
 
 #include <vector>
 
+#include "base/compiler_specific.h"
 #include "base/memory/scoped_ptr.h"
+#include "ipc/ipc_message.h"
 #include "mojo/public/c/environment/async_waiter.h"
 #include "mojo/public/cpp/system/core.h"
 
@@ -30,6 +32,13 @@ namespace internal {
 //
 class MessagePipeReader {
  public:
+  class Delegate {
+   public:
+    virtual void OnMessageReceived(Message& message) = 0;
+    virtual void OnPipeClosed(MessagePipeReader* reader) = 0;
+    virtual void OnPipeError(MessagePipeReader* reader) = 0;
+  };
+
   // Delay the object deletion using the current message loop.
   // This is intended to used by MessagePipeReader owners.
   class DelayedDeleter {
@@ -39,13 +48,16 @@ class MessagePipeReader {
     static void DeleteNow(MessagePipeReader* ptr) { delete ptr; }
 
     DelayedDeleter() {}
-    DelayedDeleter(const DefaultType&) {}
+    explicit DelayedDeleter(const DefaultType&) {}
     DelayedDeleter& operator=(const DefaultType&) { return *this; }
 
     void operator()(MessagePipeReader* ptr) const;
   };
 
-  explicit MessagePipeReader(mojo::ScopedMessagePipeHandle handle);
+  // Both parameters must be non-null.
+  // Build a reader that reads messages from |handle| and lets |delegate| know.
+  // Note that MessagePipeReader doesn't delete |delete|.
+  MessagePipeReader(mojo::ScopedMessagePipeHandle handle, Delegate* delegate);
   virtual ~MessagePipeReader();
 
   MojoHandle handle() const { return pipe_.get().value(); }
@@ -68,17 +80,15 @@ class MessagePipeReader {
   // Return true if the MessagePipe is alive.
   bool IsValid() { return pipe_.is_valid(); }
 
-  //
-  // The client have to implment these callback to get the readiness
-  // event from the reader
-  //
-  virtual void OnMessageReceived() = 0;
-  virtual void OnPipeClosed() = 0;
-  virtual void OnPipeError(MojoResult error) = 0;
+  bool Send(scoped_ptr<Message> message);
 
  private:
   static void InvokePipeIsReady(void* closure, MojoResult result);
 
+  void OnMessageReceived();
+  void OnPipeClosed();
+  void OnPipeError(MojoResult error);
+
   MojoResult ReadMessageBytes();
   void PipeIsReady(MojoResult wait_result);
   void StartWaiting();
@@ -88,6 +98,8 @@ class MessagePipeReader {
   std::vector<MojoHandle> handle_buffer_;
   MojoAsyncWaitID pipe_wait_id_;
   mojo::ScopedMessagePipeHandle pipe_;
+  // Is null once the message pipe is closed.
+  Delegate* delegate_;
 
   DISALLOW_COPY_AND_ASSIGN(MessagePipeReader);
 };
diff --git a/ipc/mojo/ipc_mojo.gyp b/ipc/mojo/ipc_mojo.gyp
index a9a2673..1ca1775 100644
--- a/ipc/mojo/ipc_mojo.gyp
+++ b/ipc/mojo/ipc_mojo.gyp
@@ -32,8 +32,6 @@
         'ipc_channel_mojo.h',
         'ipc_channel_mojo_host.cc',
         'ipc_channel_mojo_host.h',
-        'ipc_channel_mojo_readers.cc',
-        'ipc_channel_mojo_readers.h',
         'ipc_mojo_bootstrap.cc',
         'ipc_mojo_bootstrap.h',
         'ipc_message_pipe_reader.cc',