Mojo: First stab at making MessagePipes work across OS pipes.

mojo/system/remote_message_pipe_unittest.cc

+// Copyright 2013 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 "mojo/system/message_pipe.h"
+
+#include <fcntl.h>
+#include <stdint.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <sys/types.h>
+#include <unistd.h>
+
+#include "base/basictypes.h"
+#include "base/bind.h"
+#include "base/callback.h"
+#include "base/location.h"
+#include "base/logging.h"
+#include "base/message_loop/message_loop.h"
+#include "base/synchronization/waitable_event.h"
+#include "base/threading/thread.h"
+#include "mojo/system/channel.h"
+#include "mojo/system/local_message_pipe_endpoint.h"
+#include "mojo/system/platform_channel_handle.h"
+#include "mojo/system/proxy_message_pipe_endpoint.h"
+#include "mojo/system/test_utils.h"
+#include "mojo/system/waiter.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace mojo {
+namespace system {
+namespace {
+
+class RemoteMessagePipeTest : public testing::Test {
+ public:
+  RemoteMessagePipeTest() : io_thread_("io_thread") {
+  }
+
+  virtual ~RemoteMessagePipeTest() {
+  }
+
+  virtual void SetUp() OVERRIDE {
+    io_thread_.StartWithOptions(
+        base::Thread::Options(base::MessageLoop::TYPE_IO, 0));
+
+    test::PostTaskAndWait(io_thread_task_runner(),
+                          FROM_HERE,
+                          base::Bind(&RemoteMessagePipeTest::SetUpOnIOThread,
+                                     base::Unretained(this)));
+  }
+
+  virtual void TearDown() OVERRIDE {
+    test::PostTaskAndWait(io_thread_task_runner(),
+                          FROM_HERE,
+                          base::Bind(&RemoteMessagePipeTest::TearDownOnIOThread,
+                                     base::Unretained(this)));
+    io_thread_.Stop();
+  }
+
+  // This connects MP 0, port 1 and MP 1, port 0 (leaving MP 0, port 0 and MP 1,
+  // port 1 as the user-visible endpoints) to channel 0 and 1, respectively. MP
+  // 0, port 1 and MP 1, port 0 must have |ProxyMessagePipeEndpoint|s.
+  void ConnectMessagePipesOnIOThread(scoped_refptr<MessagePipe> mp_0,
+                                     scoped_refptr<MessagePipe> mp_1) {
+    CHECK_EQ(base::MessageLoop::current(), io_thread_message_loop());
+
+    MessageInTransit::EndpointId local_id_0 =
+        channels_[0]->AttachMessagePipeEndpoint(mp_0, 1);
+    MessageInTransit::EndpointId local_id_1 =
+        channels_[1]->AttachMessagePipeEndpoint(mp_1, 0);
+
+    channels_[0]->RunMessagePipeEndpoint(local_id_0, local_id_1);
+    channels_[1]->RunMessagePipeEndpoint(local_id_1, local_id_0);
+  }
+
+ protected:
+  base::MessageLoop* io_thread_message_loop() {
+    return io_thread_.message_loop();
+  }
+
+  scoped_refptr<base::TaskRunner> io_thread_task_runner() {
+    return io_thread_message_loop()->message_loop_proxy();
+  }
+
+ private:
+  void SetUpOnIOThread() {
+    CHECK_EQ(base::MessageLoop::current(), io_thread_message_loop());
+
+    // Create the socket.
+    int fds[2];
+    PCHECK(socketpair(AF_UNIX, SOCK_STREAM, 0, fds) == 0);
+
+    // Set the ends to non-blocking.
+    PCHECK(fcntl(fds[0], F_SETFL, O_NONBLOCK) == 0);
+    PCHECK(fcntl(fds[1], F_SETFL, O_NONBLOCK) == 0);
+
+    // Create |Channel|s.
+    channels_[0] = new Channel(PlatformChannelHandle(fds[0]));
+    channels_[1] = new Channel(PlatformChannelHandle(fds[1]));
+  }
+
+  void TearDownOnIOThread() {
+    channels_[1]->Shutdown();
+    channels_[1] = NULL;
+    channels_[0]->Shutdown();
+    channels_[0] = NULL;
+  }
+
+  base::Thread io_thread_;
+  scoped_refptr<Channel> channels_[2];
+
+  DISALLOW_COPY_AND_ASSIGN(RemoteMessagePipeTest);
+};
+
+TEST_F(RemoteMessagePipeTest, Basic) {
+  const char hello[] = "hello";
+  const char world[] = "world!!!1!!!1!";
+  char buffer[100] = { 0 };
+  uint32_t buffer_size = static_cast<uint32_t>(sizeof(buffer));
+  Waiter waiter;
+
+  // Connect message pipes. MP 0, port 1 will be attached to channel 0 and
+  // connected to MP 1, port 0, which will be attached to channel 1. This leaves
+  // MP 0, port 0 and MP 1, port 1 as the "user-facing" endpoints.
+
+  scoped_refptr<MessagePipe> mp_0(new MessagePipe(
+      scoped_ptr<MessagePipeEndpoint>(new LocalMessagePipeEndpoint()),
+      scoped_ptr<MessagePipeEndpoint>(new ProxyMessagePipeEndpoint())));
+  scoped_refptr<MessagePipe> mp_1(new MessagePipe(
+      scoped_ptr<MessagePipeEndpoint>(new ProxyMessagePipeEndpoint()),
+      scoped_ptr<MessagePipeEndpoint>(new LocalMessagePipeEndpoint())));
+  test::PostTaskAndWait(
+      io_thread_task_runner(),
+      FROM_HERE,
+      base::Bind(&RemoteMessagePipeTest::ConnectMessagePipesOnIOThread,
+                 base::Unretained(this), mp_0, mp_1));
+
+  // Write in one direction: MP 0, port 0 -> ... -> MP 1, port 1.
+
+  // Prepare to wait on MP 1, port 1. (Add the waiter now. Otherwise, if we do
+  // it later, it might already be readable.)
+  waiter.Init();
+  EXPECT_EQ(MOJO_RESULT_OK,
+            mp_1->AddWaiter(1, &waiter, MOJO_WAIT_FLAG_READABLE, 123));
+
+  // Write to MP 0, port 0.
+  EXPECT_EQ(MOJO_RESULT_OK,
+            mp_0->WriteMessage(0,
+                               hello, sizeof(hello),
+                               NULL, 0,
+                               MOJO_WRITE_MESSAGE_FLAG_NONE));
+
+  // Wait.
+  EXPECT_EQ(123, waiter.Wait(MOJO_DEADLINE_INDEFINITE));
+  mp_1->RemoveWaiter(1, &waiter);
+
+  // Read from MP 1, port 1.
+  EXPECT_EQ(MOJO_RESULT_OK,
+            mp_1->ReadMessage(1,
+                              buffer, &buffer_size,
+                              NULL, NULL,
+                              MOJO_READ_MESSAGE_FLAG_NONE));
+  EXPECT_EQ(sizeof(hello), static_cast<size_t>(buffer_size));
+  EXPECT_EQ(0, strcmp(buffer, hello));
+
+  // Write in the other direction: MP 1, port 1 -> ... -> MP 0, port 0.
+
+  waiter.Init();
+  EXPECT_EQ(MOJO_RESULT_OK,
+            mp_0->AddWaiter(0, &waiter, MOJO_WAIT_FLAG_READABLE, 456));
+
+  EXPECT_EQ(MOJO_RESULT_OK,
+            mp_1->WriteMessage(1,
+                               world, sizeof(world),
+                               NULL, 0,
+                               MOJO_WRITE_MESSAGE_FLAG_NONE));
+
+  EXPECT_EQ(456, waiter.Wait(MOJO_DEADLINE_INDEFINITE));
+  mp_0->RemoveWaiter(0, &waiter);
+
+  buffer_size = static_cast<uint32_t>(sizeof(buffer));
+  EXPECT_EQ(MOJO_RESULT_OK,
+            mp_0->ReadMessage(0,
+                              buffer, &buffer_size,
+                              NULL, NULL,
+                              MOJO_READ_MESSAGE_FLAG_NONE));
+  EXPECT_EQ(sizeof(world), static_cast<size_t>(buffer_size));
+  EXPECT_EQ(0, strcmp(buffer, world));
+
+  // Close MP 0, port 0.
+  mp_0->Close(0);
+
+  // Try to wait for MP 1, port 1 to become readable. This will eventually fail
+  // when it realizes that MP 0, port 0 has been closed. (It may also fail
+  // immediately.)
+  waiter.Init();
+  MojoResult result = mp_1->AddWaiter(1, &waiter, MOJO_WAIT_FLAG_READABLE, 789);
+  if (result == MOJO_RESULT_OK) {
+    EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
+              waiter.Wait(MOJO_DEADLINE_INDEFINITE));
+    mp_1->RemoveWaiter(1, &waiter);
+  } else {
+    EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION, result);
+  }
+
+  // And MP 1, port 1.
+  mp_1->Close(1);
+}
+
+TEST_F(RemoteMessagePipeTest, Multiplex) {
+  const char hello[] = "hello";
+  const char world[] = "world!!!1!!!1!";
+  char buffer[100] = { 0 };
+  uint32_t buffer_size = static_cast<uint32_t>(sizeof(buffer));
+  Waiter waiter;
+
+  // Connect message pipes as in the |Basic| test.
+
+  scoped_refptr<MessagePipe> mp_0(new MessagePipe(
+      scoped_ptr<MessagePipeEndpoint>(new LocalMessagePipeEndpoint()),
+      scoped_ptr<MessagePipeEndpoint>(new ProxyMessagePipeEndpoint())));
+  scoped_refptr<MessagePipe> mp_1(new MessagePipe(
+      scoped_ptr<MessagePipeEndpoint>(new ProxyMessagePipeEndpoint()),
+      scoped_ptr<MessagePipeEndpoint>(new LocalMessagePipeEndpoint())));
+  test::PostTaskAndWait(
+      io_thread_task_runner(),
+      FROM_HERE,
+      base::Bind(&RemoteMessagePipeTest::ConnectMessagePipesOnIOThread,
+                 base::Unretained(this), mp_0, mp_1));
+
+  // Now put another message pipe on the channel.
+
+  scoped_refptr<MessagePipe> mp_2(new MessagePipe(
+      scoped_ptr<MessagePipeEndpoint>(new LocalMessagePipeEndpoint()),
+      scoped_ptr<MessagePipeEndpoint>(new ProxyMessagePipeEndpoint())));
+  scoped_refptr<MessagePipe> mp_3(new MessagePipe(
+      scoped_ptr<MessagePipeEndpoint>(new ProxyMessagePipeEndpoint()),
+      scoped_ptr<MessagePipeEndpoint>(new LocalMessagePipeEndpoint())));
+  test::PostTaskAndWait(
+      io_thread_task_runner(),
+      FROM_HERE,
+      base::Bind(&RemoteMessagePipeTest::ConnectMessagePipesOnIOThread,
+                 base::Unretained(this), mp_2, mp_3));
+
+  // Write: MP 2, port 0 -> MP 3, port 1.
+
+  waiter.Init();
+  EXPECT_EQ(MOJO_RESULT_OK,
+            mp_3->AddWaiter(1, &waiter, MOJO_WAIT_FLAG_READABLE, 789));
+
+  EXPECT_EQ(MOJO_RESULT_OK,
+            mp_2->WriteMessage(0,
+                               hello, sizeof(hello),
+                               NULL, 0,
+                               MOJO_WRITE_MESSAGE_FLAG_NONE));
+
+  EXPECT_EQ(789, waiter.Wait(MOJO_DEADLINE_INDEFINITE));
+  mp_3->RemoveWaiter(1, &waiter);
+
+  // Make sure there's nothing on MP 0, port 0 or MP 1, port 1 or MP 2, port 0.
+  buffer_size = static_cast<uint32_t>(sizeof(buffer));
+  EXPECT_EQ(MOJO_RESULT_NOT_FOUND,
+            mp_0->ReadMessage(0,
+                              buffer, &buffer_size,
+                              NULL, NULL,
+                              MOJO_READ_MESSAGE_FLAG_NONE));
+  buffer_size = static_cast<uint32_t>(sizeof(buffer));
+  EXPECT_EQ(MOJO_RESULT_NOT_FOUND,
+            mp_1->ReadMessage(1,
+                              buffer, &buffer_size,
+                              NULL, NULL,
+                              MOJO_READ_MESSAGE_FLAG_NONE));
+  buffer_size = static_cast<uint32_t>(sizeof(buffer));
+  EXPECT_EQ(MOJO_RESULT_NOT_FOUND,
+            mp_2->ReadMessage(0,
+                              buffer, &buffer_size,
+                              NULL, NULL,
+                              MOJO_READ_MESSAGE_FLAG_NONE));
+
+  // Read from MP 3, port 1.
+  buffer_size = static_cast<uint32_t>(sizeof(buffer));
+  EXPECT_EQ(MOJO_RESULT_OK,
+            mp_3->ReadMessage(1,
+                              buffer, &buffer_size,
+                              NULL, NULL,
+                              MOJO_READ_MESSAGE_FLAG_NONE));
+  EXPECT_EQ(sizeof(hello), static_cast<size_t>(buffer_size));
+  EXPECT_EQ(0, strcmp(buffer, hello));
+
+  // Write: MP 0, port 0 -> MP 1, port 1 again.
+
+  waiter.Init();
+  EXPECT_EQ(MOJO_RESULT_OK,
+            mp_1->AddWaiter(1, &waiter, MOJO_WAIT_FLAG_READABLE, 123));
+
+  EXPECT_EQ(MOJO_RESULT_OK,
+            mp_0->WriteMessage(0,
+                               world, sizeof(world),
+                               NULL, 0,
+                               MOJO_WRITE_MESSAGE_FLAG_NONE));
+
+  EXPECT_EQ(123, waiter.Wait(MOJO_DEADLINE_INDEFINITE));
+  mp_1->RemoveWaiter(1, &waiter);
+
+  // Make sure there's nothing on the other ports.
+  buffer_size = static_cast<uint32_t>(sizeof(buffer));
+  EXPECT_EQ(MOJO_RESULT_NOT_FOUND,
+            mp_0->ReadMessage(0,
+                              buffer, &buffer_size,
+                              NULL, NULL,
+                              MOJO_READ_MESSAGE_FLAG_NONE));
+  buffer_size = static_cast<uint32_t>(sizeof(buffer));
+  EXPECT_EQ(MOJO_RESULT_NOT_FOUND,
+            mp_2->ReadMessage(0,
+                              buffer, &buffer_size,
+                              NULL, NULL,
+                              MOJO_READ_MESSAGE_FLAG_NONE));
+  buffer_size = static_cast<uint32_t>(sizeof(buffer));
+  EXPECT_EQ(MOJO_RESULT_NOT_FOUND,
+            mp_3->ReadMessage(1,
+                              buffer, &buffer_size,
+                              NULL, NULL,
+                              MOJO_READ_MESSAGE_FLAG_NONE));
+
+  buffer_size = static_cast<uint32_t>(sizeof(buffer));
+  EXPECT_EQ(MOJO_RESULT_OK,
+            mp_1->ReadMessage(1,
+                              buffer, &buffer_size,
+                              NULL, NULL,
+                              MOJO_READ_MESSAGE_FLAG_NONE));
+  EXPECT_EQ(sizeof(world), static_cast<size_t>(buffer_size));
+  EXPECT_EQ(0, strcmp(buffer, world));
+}
+
+}  // namespace
+}  // namespace system
+}  // namespace mojo