Adds Channel-associated interface support on ChannelProxy's thread

ipc/ipc_channel_mojo_unittest.cc

Index: ipc/ipc_channel_mojo_unittest.cc
diff --git a/ipc/ipc_channel_mojo_unittest.cc b/ipc/ipc_channel_mojo_unittest.cc
index e846c132bedc211ddea92863e47718753bafa938..0e085162ca0cc81caa09796ed629b3487fad0bf1 100644
--- a/ipc/ipc_channel_mojo_unittest.cc
+++ b/ipc/ipc_channel_mojo_unittest.cc
@@ -11,9 +11,12 @@
 #include <utility>
 
 #include "base/base_paths.h"
+#include "base/bind.h"
 #include "base/files/file.h"
 #include "base/files/scoped_temp_dir.h"
 #include "base/location.h"
+#include "base/macros.h"
+#include "base/message_loop/message_loop.h"
 #include "base/path_service.h"
 #include "base/pickle.h"
 #include "base/run_loop.h"
@@ -103,6 +106,7 @@ class ChannelClient {
   void Init(mojo::ScopedMessagePipeHandle handle) {
     handle_ = std::move(handle);
   }
+
   void Connect(IPC::Listener* listener) {
     channel_ = IPC::ChannelMojo::Create(std::move(handle_),
                                         IPC::Channel::MODE_CLIENT, listener);
@@ -126,34 +130,40 @@ class ChannelClient {
   std::unique_ptr<IPC::ChannelMojo> channel_;
 };
 
-class IPCChannelMojoTest : public testing::Test {
+class IPCChannelMojoTestBase : public testing::Test {
  public:
-  IPCChannelMojoTest() {}
-
-  void TearDown() override { base::RunLoop().RunUntilIdle(); }
-
   void InitWithMojo(const std::string& test_client_name) {
     handle_ = helper_.StartChild(test_client_name);
   }
 
+  bool WaitForClientShutdown() { return helper_.WaitForChildTestShutdown(); }
+
+ protected:
+  mojo::ScopedMessagePipeHandle TakeHandle() { return std::move(handle_); }
+
+ private:
+  mojo::ScopedMessagePipeHandle handle_;
+  mojo::edk::test::MultiprocessTestHelper helper_;
+};
+
+class IPCChannelMojoTest : public IPCChannelMojoTestBase {
+ public:
+  void TearDown() override { base::RunLoop().RunUntilIdle(); }
+
   void CreateChannel(IPC::Listener* listener) {
-    channel_ = IPC::ChannelMojo::Create(std::move(handle_),
-                                        IPC::Channel::MODE_SERVER, listener);
+    channel_ = IPC::ChannelMojo::Create(
+        TakeHandle(), IPC::Channel::MODE_SERVER, listener);
   }
 
   bool ConnectChannel() { return channel_->Connect(); }
 
   void DestroyChannel() { channel_.reset(); }
 
-  bool WaitForClientShutdown() { return helper_.WaitForChildTestShutdown(); }
-
   IPC::Sender* sender() { return channel(); }
   IPC::Channel* channel() { return channel_.get(); }
 
  private:
   base::MessageLoop message_loop_;
-  mojo::edk::test::MultiprocessTestHelper helper_;
-  mojo::ScopedMessagePipeHandle handle_;
   std::unique_ptr<IPC::Channel> channel_;
 };
 
@@ -699,7 +709,176 @@ DEFINE_IPC_CHANNEL_MOJO_TEST_CLIENT(SimpleAssociatedInterfaceClient,
   Close();
 }
 
+class ChannelProxyRunner {
+ public:
+  ChannelProxyRunner(std::unique_ptr<IPC::ChannelFactory> channel_factory)
+      : channel_factory_(std::move(channel_factory)),
+        io_thread_("ChannelProxyRunner IO thread") {
+  }
+
+  void CreateProxy(IPC::Listener* listener) {
+    io_thread_.StartWithOptions(
+        base::Thread::Options(base::MessageLoop::TYPE_IO, 0));
+    proxy_.reset(new IPC::ChannelProxy(listener, io_thread_.task_runner()));
+  }
+  void RunProxy() { proxy_->Init(std::move(channel_factory_), true); }
+
+  IPC::ChannelProxy* proxy() { return proxy_.get(); }
+
+ private:
+  std::unique_ptr<IPC::ChannelFactory> channel_factory_;
+
+  base::Thread io_thread_;
+  std::unique_ptr<IPC::ChannelProxy> proxy_;
+
+  DISALLOW_COPY_AND_ASSIGN(ChannelProxyRunner);
+};
+
+class IPCChannelProxyMojoTest : public IPCChannelMojoTestBase {
+ public:
+  void InitWithMojo(const std::string& client_name) {
+    IPCChannelMojoTestBase::InitWithMojo(client_name);
+    runner_.reset(new ChannelProxyRunner(
+        IPC::ChannelMojo::CreateServerFactory(TakeHandle())));
+  }
+  void CreateProxy(IPC::Listener* listener) { runner_->CreateProxy(listener); }
+  void RunProxy() { runner_->RunProxy(); }
+
+  IPC::ChannelProxy* proxy() { return runner_->proxy(); }
+
+ private:
+  base::MessageLoop message_loop_;
+  std::unique_ptr<ChannelProxyRunner> runner_;
+};
+
+class ListenerWithSimpleProxyAssociatedInterface
+    : public IPC::Listener,
+      public IPC::mojom::SimpleTestDriver {
+ public:
+  static const int kNumMessages;
+
+  ListenerWithSimpleProxyAssociatedInterface() : binding_(this) {}
+
+  ~ListenerWithSimpleProxyAssociatedInterface() override {}
+
+  bool OnMessageReceived(const IPC::Message& message) override {
+    base::PickleIterator iter(message);
+    std::string should_be_expected;
+    EXPECT_TRUE(iter.ReadString(&should_be_expected));
+    EXPECT_EQ(should_be_expected, next_expected_string_);
+    num_messages_received_++;
+    return true;
+  }
+
+  void OnChannelError() override {
+    DCHECK(received_quit_);
+  }
+
+  void RegisterInterfaceFactory(IPC::ChannelProxy* proxy) {
+    proxy->AddAssociatedInterface(
+        base::Bind(&ListenerWithSimpleProxyAssociatedInterface::BindRequest,
+                   base::Unretained(this)));
+  }
+
+  bool received_all_messages() const {
+    return num_messages_received_ == kNumMessages && received_quit_;
+  }
+
+ private:
+  // IPC::mojom::SimpleTestDriver:
+  void ExpectString(const mojo::String& str) override {
+    next_expected_string_ = str;
+  }
+
+  void RequestQuit(const RequestQuitCallback& callback) override {
+    received_quit_ = true;
+    callback.Run();
+    base::MessageLoop::current()->QuitWhenIdle();
+  }
+
+  void BindRequest(IPC::mojom::SimpleTestDriverAssociatedRequest request) {
+    DCHECK(!binding_.is_bound());
+    binding_.Bind(std::move(request));
+  }
+
+  std::string next_expected_string_;
+  int num_messages_received_ = 0;
+  bool received_quit_ = false;
+
+  mojo::AssociatedBinding<IPC::mojom::SimpleTestDriver> binding_;
+};
+
+const int ListenerWithSimpleProxyAssociatedInterface::kNumMessages = 1000;
+
+TEST_F(IPCChannelProxyMojoTest, ProxyThreadAssociatedInterface) {
+  InitWithMojo("ProxyThreadAssociatedInterfaceClient");
+
+  ListenerWithSimpleProxyAssociatedInterface listener;
+  CreateProxy(&listener);
+  listener.RegisterInterfaceFactory(proxy());
+  RunProxy();
+
+  base::RunLoop().Run();
+
+  EXPECT_TRUE(WaitForClientShutdown());
+  EXPECT_TRUE(listener.received_all_messages());
+
+  base::RunLoop().RunUntilIdle();
+}
+
+class ChannelProxyClient {
+ public:
+  void Init(mojo::ScopedMessagePipeHandle handle) {
+    runner_.reset(new ChannelProxyRunner(
+        IPC::ChannelMojo::CreateClientFactory(std::move(handle))));
+  }
+  void CreateProxy(IPC::Listener* listener) { runner_->CreateProxy(listener); }
+  void RunProxy() { runner_->RunProxy(); }
+
+  IPC::ChannelProxy* proxy() { return runner_->proxy(); }
+
+ private:
+  base::MessageLoop message_loop_;
+  std::unique_ptr<ChannelProxyRunner> runner_;
+};
+
+class ListenerThatWaitsForConnect : public IPC::Listener {
+ public:
+  explicit ListenerThatWaitsForConnect(const base::Closure& connect_handler)
+      : connect_handler_(connect_handler) {}
+
+  // IPC::Listener
+  bool OnMessageReceived(const IPC::Message& message) override { return true; }
+  void OnChannelConnected(int32_t) override { connect_handler_.Run(); }
+
+ private:
+  base::Closure connect_handler_;
+};
+
+DEFINE_IPC_CHANNEL_MOJO_TEST_CLIENT(ProxyThreadAssociatedInterfaceClient,
+                                    ChannelProxyClient) {
+  base::RunLoop connect_loop;
+  ListenerThatWaitsForConnect listener(connect_loop.QuitClosure());
+  CreateProxy(&listener);
+  RunProxy();
+  connect_loop.Run();
+
+  // Send a bunch of interleaved messages, alternating between the associated
+  // interface and a legacy IPC::Message.
+  IPC::mojom::SimpleTestDriverAssociatedPtr driver;
+  proxy()->GetRemoteAssociatedInterface(&driver);
+  for (int i = 0; i < ListenerWithSimpleProxyAssociatedInterface::kNumMessages;
+       ++i) {
+    std::string str = base::StringPrintf("Hello! %d", i);
+    driver->ExpectString(str);
+    SendString(proxy(), str);
+  }
+  driver->RequestQuit(base::MessageLoop::QuitWhenIdleClosure());
+  base::RunLoop().Run();
+}
+
 #if defined(OS_POSIX)
+
 class ListenerThatExpectsFile : public IPC::Listener {
  public:
   ListenerThatExpectsFile() : sender_(NULL) {}