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

mojo/system/raw_channel_posix_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.
 
 // TODO(vtl): Factor out the POSIX-specific bits of this test (once we have a
 // non-POSIX implementation).
 
 #include "mojo/system/raw_channel.h"
 
 #include <fcntl.h>
@@ skipping 27 matching lines @@
 #include "testing/gtest/include/gtest/gtest.h"
 
 namespace mojo {
 namespace system {
 namespace {
 
 MessageInTransit* MakeTestMessage(uint32_t num_bytes) {
   std::vector<unsigned char> bytes(num_bytes, 0);
   for (size_t i = 0; i < num_bytes; i++)
     bytes[i] = static_cast<unsigned char>(i + num_bytes);
-  return MessageInTransit::Create(bytes.data(), num_bytes);
+  return MessageInTransit::Create(
+      MessageInTransit::kTypeMessagePipeEndpoint,
+      MessageInTransit::kSubtypeMessagePipeEndpointData,
+      bytes.data(), num_bytes);
 }
 
 bool CheckMessageData(const void* bytes, uint32_t num_bytes) {
   const unsigned char* b = static_cast<const unsigned char*>(bytes);
   for (uint32_t i = 0; i < num_bytes; i++) {
     if (b[i] != static_cast<unsigned char>(i + num_bytes))
       return false;
   }
   return true;
 }
 
+void InitOnIOThread(RawChannel* raw_channel) {
+  CHECK(raw_channel->Init());
+}
+
 // -----------------------------------------------------------------------------
 
 class RawChannelPosixTest : public testing::Test {
  public:
   RawChannelPosixTest() : io_thread_("io_thread") {
     fds_[0] = -1;
     fds_[1] = -1;
   }
 
   virtual ~RawChannelPosixTest() {
@@ skipping 133 matching lines @@
   scoped_ptr<RawChannel> rc(RawChannel::Create(PlatformChannelHandle(fd(0)),
                                                &delegate,
                                                io_thread_message_loop()));
   // |RawChannel::Create()| takes ownership of the FD.
   clear_fd(0);
 
   TestMessageReaderAndChecker checker(fd(1));
 
   test::PostTaskAndWait(io_thread_task_runner(),
                         FROM_HERE,
-                        base::Bind(&RawChannel::Init,
-                                   base::Unretained(rc.get())));
+                        base::Bind(&InitOnIOThread, rc.get()));
 
   // Write and read, for a variety of sizes.
   for (uint32_t size = 1; size < 5 * 1000 * 1000; size += size / 2 + 1) {
     EXPECT_TRUE(rc->WriteMessage(MakeTestMessage(size)));
     EXPECT_TRUE(checker.ReadAndCheckNextMessage(size)) << size;
   }
 
   // Write/queue and read afterwards, for a variety of sizes.
   for (uint32_t size = 1; size < 5 * 1000 * 1000; size += size / 2 + 1)
     EXPECT_TRUE(rc->WriteMessage(MakeTestMessage(size)));
@@ skipping 72 matching lines @@
 
   ReadCheckerRawChannelDelegate delegate;
   scoped_ptr<RawChannel> rc(RawChannel::Create(PlatformChannelHandle(fd(0)),
                                                &delegate,
                                                io_thread_message_loop()));
   // |RawChannel::Create()| takes ownership of the FD.
   clear_fd(0);
 
   test::PostTaskAndWait(io_thread_task_runner(),
                         FROM_HERE,
-                        base::Bind(&RawChannel::Init,
-                                   base::Unretained(rc.get())));
+                        base::Bind(&InitOnIOThread, rc.get()));
 
   // Write and read, for a variety of sizes.
   for (uint32_t size = 1; size < 5 * 1000 * 1000; size += size / 2 + 1) {
     delegate.SetExpectedSizes(std::vector<uint32_t>(1, size));
     MessageInTransit* message = MakeTestMessage(size);
     EXPECT_EQ(static_cast<ssize_t>(message->size_with_header_and_padding()),
               write(fd(1), message, message->size_with_header_and_padding()));
     message->Destroy();
     delegate.Wait();
   }
@@ skipping 90 matching lines @@
   WriteOnlyRawChannelDelegate writer_delegate;
   scoped_ptr<RawChannel> writer_rc(
       RawChannel::Create(PlatformChannelHandle(fd(0)),
                                                &writer_delegate,
                                                io_thread_message_loop()));
   // |RawChannel::Create()| takes ownership of the FD.
   clear_fd(0);
 
   test::PostTaskAndWait(io_thread_task_runner(),
                         FROM_HERE,
-                        base::Bind(&RawChannel::Init,
-                                   base::Unretained(writer_rc.get())));
+                        base::Bind(&InitOnIOThread, writer_rc.get()));
 
   ReadCountdownRawChannelDelegate reader_delegate(
       kNumWriterThreads * kNumWriteMessagesPerThread);
   scoped_ptr<RawChannel> reader_rc(
       RawChannel::Create(PlatformChannelHandle(fd(1)),
                                                &reader_delegate,
                                                io_thread_message_loop()));
   // |RawChannel::Create()| takes ownership of the FD.
   clear_fd(1);
 
   test::PostTaskAndWait(io_thread_task_runner(),
                         FROM_HERE,
-                        base::Bind(&RawChannel::Init,
-                                   base::Unretained(reader_rc.get())));
+                        base::Bind(&InitOnIOThread, reader_rc.get()));
 
   {
     ScopedVector<RawChannelWriterThread> writer_threads;
     for (size_t i = 0; i < kNumWriterThreads; i++) {
       writer_threads.push_back(new RawChannelWriterThread(
           writer_rc.get(), kNumWriteMessagesPerThread));
     }
     for (size_t i = 0; i < writer_threads.size(); i++)
       writer_threads[i]->Start();
   }  // Joins all the writer threads.
@@ skipping 58 matching lines @@
 TEST_F(RawChannelPosixTest, OnFatalError) {
   FatalErrorRecordingRawChannelDelegate delegate;
   scoped_ptr<RawChannel> rc(RawChannel::Create(PlatformChannelHandle(fd(0)),
                                                &delegate,
                                                io_thread_message_loop()));
   // |RawChannel::Create()| takes ownership of the FD.
   clear_fd(0);
 
   test::PostTaskAndWait(io_thread_task_runner(),
                         FROM_HERE,
-                        base::Bind(&RawChannel::Init,
-                                   base::Unretained(rc.get())));
+                        base::Bind(&InitOnIOThread, rc.get()));
 
   // Close the other end, which should make writing fail.
   CHECK_EQ(close(fd(1)), 0);
   clear_fd(1);
 
   EXPECT_FALSE(rc->WriteMessage(MakeTestMessage(1)));
 
   // TODO(vtl): In theory, it's conceivable that closing the other end might
   // lead to read failing. In practice, it doesn't seem to.
   EXPECT_EQ(RawChannel::Delegate::FATAL_ERROR_FAILED_WRITE,
@@ skipping 13 matching lines @@
 TEST_F(RawChannelPosixTest, WriteMessageAfterShutdown) {
   WriteOnlyRawChannelDelegate delegate;
   scoped_ptr<RawChannel> rc(RawChannel::Create(PlatformChannelHandle(fd(0)),
                                                &delegate,
                                                io_thread_message_loop()));
   // |RawChannel::Create()| takes ownership of the FD.
   clear_fd(0);
 
   test::PostTaskAndWait(io_thread_task_runner(),
                         FROM_HERE,
-                        base::Bind(&RawChannel::Init,
-                                   base::Unretained(rc.get())));
+                        base::Bind(&InitOnIOThread, rc.get()));
   test::PostTaskAndWait(io_thread_task_runner(),
                         FROM_HERE,
                         base::Bind(&RawChannel::Shutdown,
                                    base::Unretained(rc.get())));
 
   EXPECT_FALSE(rc->WriteMessage(MakeTestMessage(1)));
 }
 
 }  // namespace
 }  // namespace system
 }  // namespace mojo