Add multiplexing of message pipes in the new EDK.

mojo/edk/system/raw_channel_unittest.cc

Index: mojo/edk/system/raw_channel_unittest.cc
diff --git a/mojo/edk/system/raw_channel_unittest.cc b/mojo/edk/system/raw_channel_unittest.cc
index fdc77d8c7c0d29b5f641cb4abe7bd3d203e271d8..c2298d61568a40ea2116ce6b2ae72135816ccd22 100644
--- a/mojo/edk/system/raw_channel_unittest.cc
+++ b/mojo/edk/system/raw_channel_unittest.cc
@@ -23,7 +23,9 @@
 #include "base/synchronization/waitable_event.h"
 #include "base/test/test_io_thread.h"
 #include "base/threading/simple_thread.h"
-#include "build/build_config.h"  // TODO(vtl): Remove this.
+#include "build/build_config.h"
+#include "mojo/edk/embedder/embedder.h"
+#include "mojo/edk/embedder/embedder_internal.h"
 #include "mojo/edk/embedder/platform_channel_pair.h"
 #include "mojo/edk/embedder/platform_handle.h"
 #include "mojo/edk/embedder/scoped_platform_handle.h"
@@ -73,7 +75,7 @@ bool WriteTestMessageToHandle(const PlatformHandle& handle,
 
 // -----------------------------------------------------------------------------
 
-class RawChannelTest : public test::MojoSystemTest {
+class RawChannelTest : public testing::Test {
  public:
   RawChannelTest() {}
   ~RawChannelTest() override {}
@@ -89,6 +91,14 @@ class RawChannelTest : public test::MojoSystemTest {
     handles[1].reset();
   }
 
+  void FlushIOThread() {
+    base::WaitableEvent event(false, false);
+    internal::g_io_thread_task_runner->PostTask(
+        FROM_HERE,
+        base::Bind(&base::WaitableEvent::Signal, base::Unretained(&event)));
+    event.Wait();
+  }
+
  protected:
   ScopedPlatformHandle handles[2];
 
@@ -189,7 +199,7 @@ TEST_F(RawChannelTest, WriteMessage) {
   WriteOnlyRawChannelDelegate delegate;
   RawChannel* rc = RawChannel::Create(handles[0].Pass());
   TestMessageReaderAndChecker checker(handles[1].get());
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE,
       base::Bind(&InitOnIOThread, rc, base::Unretained(&delegate)));
 
@@ -205,7 +215,7 @@ TEST_F(RawChannelTest, WriteMessage) {
   for (uint32_t size = 1; size < 5 * 1000 * 1000; size += size / 2 + 1)
     EXPECT_TRUE(checker.ReadAndCheckNextMessage(size)) << size;
 
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE, base::Bind(&RawChannel::Shutdown, base::Unretained(rc)));
 }
 
@@ -274,7 +284,7 @@ class ReadCheckerRawChannelDelegate : public RawChannel::Delegate {
 TEST_F(RawChannelTest, OnReadMessage) {
   ReadCheckerRawChannelDelegate delegate;
   RawChannel* rc = RawChannel::Create(handles[0].Pass());
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE,
       base::Bind(&InitOnIOThread, rc, base::Unretained(&delegate)));
 
@@ -297,7 +307,7 @@ TEST_F(RawChannelTest, OnReadMessage) {
     EXPECT_TRUE(WriteTestMessageToHandle(handles[1].get(), size));
   delegate.Wait();
 
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE, base::Bind(&RawChannel::Shutdown, base::Unretained(rc)));
 }
 
@@ -371,7 +381,7 @@ TEST_F(RawChannelTest, WriteMessageAndOnReadMessage) {
 
   WriteOnlyRawChannelDelegate writer_delegate;
   RawChannel* writer_rc = RawChannel::Create(handles[0].Pass());
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
     FROM_HERE,
     base::Bind(&InitOnIOThread, writer_rc,
                base::Unretained(&writer_delegate)));
@@ -379,7 +389,7 @@ TEST_F(RawChannelTest, WriteMessageAndOnReadMessage) {
   ReadCountdownRawChannelDelegate reader_delegate(kNumWriterThreads *
                                                   kNumWriteMessagesPerThread);
   RawChannel* reader_rc = RawChannel::Create(handles[1].Pass());
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
     FROM_HERE,
     base::Bind(&InitOnIOThread, reader_rc,
                base::Unretained(&reader_delegate)));
@@ -401,11 +411,11 @@ TEST_F(RawChannelTest, WriteMessageAndOnReadMessage) {
   // Wait for reading to finish.
   reader_delegate.Wait();
 
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE,
       base::Bind(&RawChannel::Shutdown, base::Unretained(reader_rc)));
 
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE,
       base::Bind(&RawChannel::Shutdown, base::Unretained(writer_rc)));
 }
@@ -470,9 +480,10 @@ class ErrorRecordingRawChannelDelegate
 TEST_F(RawChannelTest, OnError) {
   ErrorRecordingRawChannelDelegate delegate(0, true, true);
   RawChannel* rc = RawChannel::Create(handles[0].Pass());
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE,
       base::Bind(&InitOnIOThread, rc, base::Unretained(&delegate)));
+  FlushIOThread();
 
   // Close the handle of the other end, which should make writing fail.
   handles[1].reset();
@@ -491,7 +502,7 @@ TEST_F(RawChannelTest, OnError) {
   // notification. (If we actually get another one, |OnError()| crashes.)
   test::Sleep(test::DeadlineFromMilliseconds(20));
 
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE, base::Bind(&RawChannel::Shutdown, base::Unretained(rc)));
 }
 
@@ -513,9 +524,10 @@ TEST_F(RawChannelTest, ReadUnaffectedByWriteError) {
   // messages that were written.
   ErrorRecordingRawChannelDelegate delegate(kMessageCount, true, true);
   RawChannel* rc = RawChannel::Create(handles[0].Pass());
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE,
       base::Bind(&InitOnIOThread, rc, base::Unretained(&delegate)));
+  FlushIOThread();
 
   EXPECT_FALSE(rc->WriteMessage(MakeTestMessage(1)));
 
@@ -528,7 +540,7 @@ TEST_F(RawChannelTest, ReadUnaffectedByWriteError) {
   // And then we should get a read error.
   delegate.WaitForReadError();
 
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE, base::Bind(&RawChannel::Shutdown, base::Unretained(rc)));
 }
 
@@ -597,13 +609,13 @@ TEST_F(RawChannelTest, ReadWritePlatformHandles) {
 
   WriteOnlyRawChannelDelegate write_delegate;
   RawChannel* rc_write = RawChannel::Create(handles[0].Pass());
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE,
       base::Bind(&InitOnIOThread, rc_write, base::Unretained(&write_delegate)));
 
   ReadPlatformHandlesCheckerRawChannelDelegate read_delegate;
   RawChannel* rc_read = RawChannel::Create(handles[1].Pass());
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE,
       base::Bind(&InitOnIOThread, rc_read, base::Unretained(&read_delegate)));
 
@@ -633,10 +645,10 @@ TEST_F(RawChannelTest, ReadWritePlatformHandles) {
 
   read_delegate.Wait();
 
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE,
       base::Bind(&RawChannel::Shutdown, base::Unretained(rc_read)));
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE,
       base::Bind(&RawChannel::Shutdown, base::Unretained(rc_write)));
 }