Support generic synchronization against an IPC::Channel

ipc/ipc_channel_mojo_unittest.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/ipc_channel_mojo.h"
 
 #include <stddef.h>
 #include <stdint.h>
 
 #include <memory>
@@ skipping 19 matching lines @@
 #include "build/build_config.h"
 #include "ipc/ipc_message.h"
 #include "ipc/ipc_mojo_handle_attachment.h"
 #include "ipc/ipc_mojo_message_helper.h"
 #include "ipc/ipc_mojo_param_traits.h"
 #include "ipc/ipc_sync_channel.h"
 #include "ipc/ipc_sync_message.h"
 #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)
 #include "base/file_descriptor_posix.h"
 #include "ipc/ipc_platform_file_attachment_posix.h"
 #endif
 
 namespace {
 
 void SendString(IPC::Sender* sender, const std::string& str) {
   IPC::Message* message = new IPC::Message(0, 2, IPC::Message::PRIORITY_NORMAL);
   message->WriteString(str);
   ASSERT_TRUE(sender->Send(message));
 }
 
 void SendValue(IPC::Sender* sender, int32_t value) {
   IPC::Message* message = new IPC::Message(0, 2, IPC::Message::PRIORITY_NORMAL);
   message->WriteInt(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) {}
 
   ~ListenerThatExpectsOK() override {}
 
   bool OnMessageReceived(const IPC::Message& message) override {
     base::PickleIterator iter(message);
     std::string should_be_ok;
     EXPECT_TRUE(iter.ReadString(&should_be_ok));
@@ skipping 612 matching lines @@
   }
   void CreateProxy(IPC::Listener* listener) { runner_->CreateProxy(listener); }
   void RunProxy() {
     runner_->RunProxy();
   }
   void DestroyProxy() {
     runner_.reset();
     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:
   std::unique_ptr<ChannelProxyRunner> runner_;
 };
 
 class ListenerWithSimpleProxyAssociatedInterface
     : public IPC::Listener,
       public IPC::mojom::SimpleTestDriver {
  public:
@@ skipping 546 matching lines @@
   expected_values.push(4);
   expected_values.push(5);
   expected_values.push(2);
   expected_values.push(3);
   RunProxy();
   base::RunLoop().Run();
   EXPECT_TRUE(expected_values.empty());
   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 {
  public:
   ListenerThatExpectsFile() : sender_(NULL) {}
 
   ~ListenerThatExpectsFile() override {}
 
   bool OnMessageReceived(const IPC::Message& message) override {
     base::PickleIterator iter(message);
@@ skipping 140 matching lines @@
   Connect(&listener);
 
   base::RunLoop().Run();
 
   Close();
 }
 
 #endif  // OS_LINUX
 
 }  // namespace