Support generic synchronization against an IPC::Channel

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 229716edc0dbae08db8694e8537745961347d0bc..24750291c1ae8ea4b18b2a87ad87d5c7c5692d32 100644
--- a/ipc/ipc_channel_mojo_unittest.cc
+++ b/ipc/ipc_channel_mojo_unittest.cc
@@ -37,6 +37,7 @@
 #include "ipc/ipc_test.mojom.h"
 #include "ipc/ipc_test_base.h"
 #include "ipc/ipc_test_channel_listener.h"
+#include "mojo/public/cpp/bindings/binding.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
 #if defined(OS_POSIX)
@@ -58,6 +59,12 @@ void SendValue(IPC::Sender* sender, int32_t value) {
   ASSERT_TRUE(sender->Send(message));
 }
 
+void SendMessagePipe(IPC::Sender* sender, mojo::ScopedMessagePipeHandle pipe) {
+  IPC::Message* message = new IPC::Message(0, 2, IPC::Message::PRIORITY_NORMAL);
+  IPC::MojoMessageHelper::WriteMessagePipeTo(message, std::move(pipe));
+  ASSERT_TRUE(sender->Send(message));
+}
+
 class ListenerThatExpectsOK : public IPC::Listener {
  public:
   ListenerThatExpectsOK() : received_ok_(false) {}
@@ -690,6 +697,19 @@ class IPCChannelProxyMojoTest : public IPCChannelMojoTestBase {
     base::RunLoop().RunUntilIdle();
   }
 
+  // Launches a secondary test client.
+  std::unique_ptr<ChannelProxyRunner>
+      LaunchClient(const std::string& client_name,
+                   IPC::Listener* listener,
+                   ClientHandle* client) {
+    client->Init(client_name);
+    std::unique_ptr<ChannelProxyRunner> runner(
+        new ChannelProxyRunner(client->TakeHandle(), true));
+    runner->CreateProxy(listener);
+    runner->RunProxy();
+    return runner;
+  }
+
   IPC::ChannelProxy* proxy() { return runner_->proxy(); }
 
  private:
@@ -1256,6 +1276,218 @@ DEFINE_IPC_CHANNEL_MOJO_TEST_CLIENT_WITH_CUSTOM_FIXTURE(CreatePausedClient,
   DestroyProxy();
 }
 
+class TokenListener : public IPC::Listener,
+                      public IPC::mojom::SynchronizationTestClient {
+ public:
+  explicit TokenListener(IPC::mojom::SynchronizationTestClientRequest request)
+      : binding_(this, std::move(request)) {}
+  ~TokenListener() override {}
+
+  bool OnMessageReceived(const IPC::Message& message) override {
+    base::PickleIterator iter(message);
+    EXPECT_TRUE(iter.ReadString(&registered_token_));
+    return true;
+  }
+
+  const std::string& registered_token() const { return registered_token_; }
+
+  void set_quit_closure(const base::Closure& closure) {
+    quit_closure_ = closure;
+  }
+
+ private:
+  // IPC::mojom::SynchronizationTestClient:
+  void OnTokenReceived(const std::string& token,
+                       const OnTokenReceivedCallback& callback) override {
+    EXPECT_FALSE(registered_token_.empty());
+    EXPECT_EQ(registered_token_, token);
+    callback.Run();
+    quit_closure_.Run();
+  }
+
+  std::string registered_token_;
+  base::Closure quit_closure_;
+  mojo::Binding<IPC::mojom::SynchronizationTestClient> binding_;
+};
+
+TEST_F(IPCChannelProxyMojoTest, ChannelSynchronizationEvent) {
+  // This test creates two child processes A and B. It then establishes a direct
+  // pipe between process A and B -- specifically process A gets a
+  // SynchronizationTestReceiver proxy to process B.
+  //
+  // Process A sends a legacy IPC message to the main process to register a
+  // token there (see TokenListener above).
+  //
+  // Process A also sends a message directly to B referencing the token, and
+  // including a channel synchronization event for B to wait on.
+  //
+  // Process B waits on the event and then sends the token back to the main
+  // process over an independent interface (SynchronizationTestClient).
+  //
+  // Because a channel synchronization event was used, it should be impossible
+  // for the main process to receive process B's message before it receives
+  // process A's token registration.
+  //
+  // This test verifies that.
+  ClientHandle sender_handle;
+  IPC::mojom::SynchronizationTestClientPtr client;
+  TokenListener sender_listener(mojo::GetProxy(&client));
+  std::unique_ptr<ChannelProxyRunner> sender =
+      LaunchClient("SynchronizationTestSender", &sender_listener,
+                   &sender_handle);
+
+  ClientHandle receiver_handle;
+  DummyListener receiver_listener;
+  std::unique_ptr<ChannelProxyRunner> receiver =
+      LaunchClient("SynchronizationTestReceiver", &receiver_listener,
+                   &receiver_handle);
+
+  base::RunLoop loop;
+  sender_listener.set_quit_closure(loop.QuitClosure());
+
+  IPC::mojom::SynchronizationTestReceiverPtr receiver_proxy;
+  SendMessagePipe(receiver->proxy(),
+                  mojo::GetProxy(&receiver_proxy).PassMessagePipe());
+
+  IPC::mojom::SynchronizationTestSenderPtr sender_proxy;
+  SendMessagePipe(sender->proxy(),
+                  mojo::GetProxy(&sender_proxy).PassMessagePipe());
+
+  sender_proxy->SendToken(std::move(receiver_proxy), std::move(client));
+  loop.Run();
+
+  sender_proxy->Quit();
+
+  EXPECT_TRUE(sender_handle.WaitForShutdown());
+  EXPECT_TRUE(receiver_handle.WaitForShutdown());
+}
+
+class SynchronizationTestPipeListener : public IPC::Listener {
+ public:
+  using PipeHandler = base::Callback<void(mojo::ScopedMessagePipeHandle)>;
+
+  SynchronizationTestPipeListener() {}
+  ~SynchronizationTestPipeListener() override {}
+
+  bool OnMessageReceived(const IPC::Message& message) override {
+    base::PickleIterator iter(message);
+    mojo::ScopedMessagePipeHandle handle;
+    EXPECT_TRUE(IPC::MojoMessageHelper::ReadMessagePipeFrom(
+        &message, &iter, &handle));
+    pipe_handler_.Run(std::move(handle));
+    return true;
+  }
+
+  void set_pipe_handler(const PipeHandler& handler) { pipe_handler_ = handler; }
+
+ private:
+  PipeHandler pipe_handler_;
+  base::Closure quit_closure_;
+};
+
+class SynchronizationTestSenderImpl
+    : public IPC::mojom::SynchronizationTestSender {
+ public:
+  SynchronizationTestSenderImpl(
+      IPC::ChannelProxy* channel,
+      const base::Closure& quit_closure)
+      : channel_(channel), quit_closure_(quit_closure), binding_(this) {}
+
+  ~SynchronizationTestSenderImpl() override {}
+
+  void Bind(mojo::ScopedMessagePipeHandle handle) {
+    binding_.Bind(std::move(handle));
+  }
+
+  // IPC::mojom::SynchronizationTestSender:
+  void SendToken(IPC::mojom::SynchronizationTestReceiverPtr receiver,
+                 IPC::mojom::SynchronizationTestClientPtr client) override {
+    // Register the token with the parent process over legacy IPC, then send it
+    // to the receiver process with a synchronization event.
+    const std::string kToken = "make IPC great again";
+    SendString(channel_, kToken);
+    receiver->ReceiveToken(
+        kToken, std::move(client), channel_->CreateSynchronizationEvent());
+  }
+
+  void Quit() override { quit_closure_.Run(); }
+
+ private:
+  IPC::ChannelProxy* const channel_;
+  const base::Closure quit_closure_;
+  mojo::Binding<IPC::mojom::SynchronizationTestSender> binding_;
+
+  DISALLOW_COPY_AND_ASSIGN(SynchronizationTestSenderImpl);
+};
+
+DEFINE_IPC_CHANNEL_MOJO_TEST_CLIENT_WITH_CUSTOM_FIXTURE(
+    SynchronizationTestSender, ChannelProxyClient) {
+  SynchronizationTestPipeListener listener;
+  CreateProxy(&listener);
+  RunProxy();
+
+  base::RunLoop loop;
+  SynchronizationTestSenderImpl impl(proxy(), loop.QuitClosure());
+  listener.set_pipe_handler(base::Bind(&SynchronizationTestSenderImpl::Bind,
+                                       base::Unretained(&impl)));
+  loop.Run();
+  DestroyProxy();
+}
+
+class SynchronizationTestReceiverImpl
+    : public IPC::mojom::SynchronizationTestReceiver,
+      public mojo::common::mojom::Event {
+ public:
+  explicit SynchronizationTestReceiverImpl(
+      const base::Closure& quit_closure)
+      : quit_closure_(quit_closure), binding_(this), event_binding_(this) {}
+
+  ~SynchronizationTestReceiverImpl() override {}
+
+  void Bind(mojo::ScopedMessagePipeHandle handle) {
+    binding_.Bind(std::move(handle));
+  }
+
+  // IPC::mojom::SynchronizationTestReceiver:
+  void ReceiveToken(const std::string& token,
+                    IPC::mojom::SynchronizationTestClientPtr client,
+                    mojo::common::mojom::EventRequest event) override {
+    // Hang onto the token until |event| signals.
+    received_token_ = token;
+    client_ = std::move(client);
+    event_binding_.Bind(std::move(event));
+  }
+
+  // mojo::common::mojom::Event:
+  void Signal() override {
+    // This request should always reach the parent process after the sender's
+    // legacy IPC which registered the same token there.
+    client_->OnTokenReceived(received_token_, quit_closure_);
+  }
+
+ private:
+  const base::Closure quit_closure_;
+  std::string received_token_;
+  IPC::mojom::SynchronizationTestClientPtr client_;
+  mojo::Binding<IPC::mojom::SynchronizationTestReceiver> binding_;
+  mojo::Binding<mojo::common::mojom::Event> event_binding_;
+
+  DISALLOW_COPY_AND_ASSIGN(SynchronizationTestReceiverImpl);
+};
+
+DEFINE_IPC_CHANNEL_MOJO_TEST_CLIENT_WITH_CUSTOM_FIXTURE(
+    SynchronizationTestReceiver, ChannelProxyClient) {
+  SynchronizationTestPipeListener listener;
+  CreateProxy(&listener);
+  RunProxy();
+  base::RunLoop loop;
+  SynchronizationTestReceiverImpl impl(loop.QuitClosure());
+  listener.set_pipe_handler(base::Bind(&SynchronizationTestReceiverImpl::Bind,
+                                       base::Unretained(&impl)));
+  loop.Run();
+  DestroyProxy();
+}
+
 #if defined(OS_POSIX)
 
 class ListenerThatExpectsFile : public IPC::Listener {