EDK: Add TaskRunner and PlatformHandleWatcher to RawChannel.

mojo/edk/system/raw_channel_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 "mojo/edk/system/raw_channel.h"
 
 #include <stdint.h>
 #include <stdio.h>
 
 #include <memory>
@@ skipping 177 matching lines @@
   std::vector<unsigned char> bytes_;
 
   MOJO_DISALLOW_COPY_AND_ASSIGN(TestMessageReaderAndChecker);
 };
 
 // Tests writing (and verifies reading using our own custom reader).
 TEST_F(RawChannelTest, WriteMessage) {
   WriteOnlyRawChannelDelegate delegate;
   std::unique_ptr<RawChannel> rc(RawChannel::Create(handles[0].Pass()));
   TestMessageReaderAndChecker checker(handles[1].get());
-  io_thread()->PostTaskAndWait([&rc, &delegate]() { rc->Init(&delegate); });
+  io_thread()->PostTaskAndWait([this, &rc, &delegate]() {
+    rc->Init(io_thread()->task_runner().Clone(),
+             io_thread()->platform_handle_watcher(), &delegate);
+  });
 
   // 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 61 matching lines @@
   std::vector<uint32_t> expected_sizes_ MOJO_GUARDED_BY(mutex_);
   size_t position_ MOJO_GUARDED_BY(mutex_);
 
   MOJO_DISALLOW_COPY_AND_ASSIGN(ReadCheckerRawChannelDelegate);
 };
 
 // Tests reading (writing using our own custom writer).
 TEST_F(RawChannelTest, OnReadMessage) {
   ReadCheckerRawChannelDelegate delegate;
   std::unique_ptr<RawChannel> rc(RawChannel::Create(handles[0].Pass()));
-  io_thread()->PostTaskAndWait([&rc, &delegate]() { rc->Init(&delegate); });
+  io_thread()->PostTaskAndWait([this, &rc, &delegate]() {
+    rc->Init(io_thread()->task_runner().Clone(),
+             io_thread()->platform_handle_watcher(), &delegate);
+  });
 
   // 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));
 
     EXPECT_TRUE(WriteTestMessageToHandle(handles[1].get(), size));
 
     delegate.Wait();
   }
 
@@ skipping 71 matching lines @@
 
   MOJO_DISALLOW_COPY_AND_ASSIGN(ReadCountdownRawChannelDelegate);
 };
 
 TEST_F(RawChannelTest, WriteMessageAndOnReadMessage) {
   static const size_t kNumWriterThreads = 10;
   static const size_t kNumWriteMessagesPerThread = 4000;
 
   WriteOnlyRawChannelDelegate writer_delegate;
   std::unique_ptr<RawChannel> writer_rc(RawChannel::Create(handles[0].Pass()));
-  io_thread()->PostTaskAndWait(
-      [&writer_rc, &writer_delegate]() { writer_rc->Init(&writer_delegate); });
+  io_thread()->PostTaskAndWait([this, &writer_rc, &writer_delegate]() {
+    writer_rc->Init(io_thread()->task_runner().Clone(),
+                    io_thread()->platform_handle_watcher(), &writer_delegate);
+  });
 
   ReadCountdownRawChannelDelegate reader_delegate(kNumWriterThreads *
                                                   kNumWriteMessagesPerThread);
   std::unique_ptr<RawChannel> reader_rc(RawChannel::Create(handles[1].Pass()));
-  io_thread()->PostTaskAndWait(
-      [&reader_rc, &reader_delegate]() { reader_rc->Init(&reader_delegate); });
+  io_thread()->PostTaskAndWait([this, &reader_rc, &reader_delegate]() {
+    reader_rc->Init(io_thread()->task_runner().Clone(),
+                    io_thread()->platform_handle_watcher(), &reader_delegate);
+  });
 
   {
     std::vector<std::unique_ptr<RawChannelWriterThread>> writer_threads;
     for (size_t i = 0; i < kNumWriterThreads; i++) {
       writer_threads.push_back(MakeUnique<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 61 matching lines @@
   bool expecting_read_error_;
   bool expecting_write_error_;
 
   MOJO_DISALLOW_COPY_AND_ASSIGN(ErrorRecordingRawChannelDelegate);
 };
 
 // Tests (fatal) errors.
 TEST_F(RawChannelTest, OnError) {
   ErrorRecordingRawChannelDelegate delegate(0, true, true);
   std::unique_ptr<RawChannel> rc(RawChannel::Create(handles[0].Pass()));
-  io_thread()->PostTaskAndWait([&rc, &delegate]() { rc->Init(&delegate); });
+  io_thread()->PostTaskAndWait([this, &rc, &delegate]() {
+    rc->Init(io_thread()->task_runner().Clone(),
+             io_thread()->platform_handle_watcher(), &delegate);
+  });
 
   // Close the handle of the other end, which should make writing fail.
   handles[1].reset();
 
   EXPECT_FALSE(rc->WriteMessage(MakeTestMessage(1)));
 
   // We should get a write error.
   delegate.WaitForWriteError();
 
   // We should also get a read error.
@@ skipping 19 matching lines @@
        i++, message_size += message_size / 2 + 1)
     EXPECT_TRUE(WriteTestMessageToHandle(handles[1].get(), message_size));
 
   // Close the other end, which should make writing fail.
   handles[1].reset();
 
   // Only start up reading here. The system buffer should still contain the
   // messages that were written.
   ErrorRecordingRawChannelDelegate delegate(kMessageCount, true, true);
   std::unique_ptr<RawChannel> rc(RawChannel::Create(handles[0].Pass()));
-  io_thread()->PostTaskAndWait([&rc, &delegate]() { rc->Init(&delegate); });
+  io_thread()->PostTaskAndWait([this, &rc, &delegate]() {
+    rc->Init(io_thread()->task_runner().Clone(),
+             io_thread()->platform_handle_watcher(), &delegate);
+  });
 
   EXPECT_FALSE(rc->WriteMessage(MakeTestMessage(1)));
 
   // We should definitely get a write error.
   delegate.WaitForWriteError();
 
   // Wait for reading to finish. A writing failure shouldn't affect reading.
   delegate.Wait();
 
   // And then we should get a read error.
   delegate.WaitForReadError();
 
   io_thread()->PostTaskAndWait([&rc]() { rc->Shutdown(); });
 }
 
 // RawChannelTest.WriteMessageAfterShutdown ------------------------------------
 
 // Makes sure that calling |WriteMessage()| after |Shutdown()| behaves
 // correctly.
 TEST_F(RawChannelTest, WriteMessageAfterShutdown) {
   WriteOnlyRawChannelDelegate delegate;
   std::unique_ptr<RawChannel> rc(RawChannel::Create(handles[0].Pass()));
-  io_thread()->PostTaskAndWait([&rc, &delegate]() { rc->Init(&delegate); });
+  io_thread()->PostTaskAndWait([this, &rc, &delegate]() {
+    rc->Init(io_thread()->task_runner().Clone(),
+             io_thread()->platform_handle_watcher(), &delegate);
+  });
   io_thread()->PostTaskAndWait([&rc]() { rc->Shutdown(); });
 
   EXPECT_FALSE(rc->WriteMessage(MakeTestMessage(1)));
 }
 
 // RawChannelTest.{Shutdown, ShutdownAndDestroy}OnReadMessage ------------------
 
 class ShutdownOnReadMessageRawChannelDelegate : public RawChannel::Delegate {
  public:
   explicit ShutdownOnReadMessageRawChannelDelegate(RawChannel* raw_channel,
@@ skipping 36 matching lines @@
   MOJO_DISALLOW_COPY_AND_ASSIGN(ShutdownOnReadMessageRawChannelDelegate);
 };
 
 TEST_F(RawChannelTest, ShutdownOnReadMessage) {
   // Write a few messages into the other end.
   for (size_t count = 0; count < 5; count++)
     EXPECT_TRUE(WriteTestMessageToHandle(handles[1].get(), 10));
 
   std::unique_ptr<RawChannel> rc(RawChannel::Create(handles[0].Pass()));
   ShutdownOnReadMessageRawChannelDelegate delegate(rc.get(), false);
-  io_thread()->PostTaskAndWait([&rc, &delegate]() { rc->Init(&delegate); });
+  io_thread()->PostTaskAndWait([this, &rc, &delegate]() {
+    rc->Init(io_thread()->task_runner().Clone(),
+             io_thread()->platform_handle_watcher(), &delegate);
+  });
 
   // Wait for the delegate, which will shut the |RawChannel| down.
   delegate.Wait();
 }
 
 TEST_F(RawChannelTest, ShutdownAndDestroyOnReadMessage) {
   // Write a message into the other end.
   EXPECT_TRUE(WriteTestMessageToHandle(handles[1].get(), 10));
 
   // The delegate will destroy |rc|.
   RawChannel* rc = RawChannel::Create(handles[0].Pass()).release();
   ShutdownOnReadMessageRawChannelDelegate delegate(rc, true);
-  io_thread()->PostTaskAndWait([&rc, &delegate]() { rc->Init(&delegate); });
+  io_thread()->PostTaskAndWait([this, &rc, &delegate]() {
+    rc->Init(io_thread()->task_runner().Clone(),
+             io_thread()->platform_handle_watcher(), &delegate);
+  });
 
   // Wait for the delegate, which will shut the |RawChannel| down.
   delegate.Wait();
 }
 
 // RawChannelTest.{Shutdown, ShutdownAndDestroy}OnError{Read, Write} -----------
 
 class ShutdownOnErrorRawChannelDelegate : public RawChannel::Delegate {
  public:
   ShutdownOnErrorRawChannelDelegate(RawChannel* raw_channel,
@@ skipping 36 matching lines @@
   AutoResetWaitableEvent done_event_;
   bool did_shutdown_;
 
   MOJO_DISALLOW_COPY_AND_ASSIGN(ShutdownOnErrorRawChannelDelegate);
 };
 
 TEST_F(RawChannelTest, ShutdownOnErrorRead) {
   std::unique_ptr<RawChannel> rc(RawChannel::Create(handles[0].Pass()));
   ShutdownOnErrorRawChannelDelegate delegate(
       rc.get(), false, RawChannel::Delegate::ERROR_READ_SHUTDOWN);
-  io_thread()->PostTaskAndWait([&rc, &delegate]() { rc->Init(&delegate); });
+  io_thread()->PostTaskAndWait([this, &rc, &delegate]() {
+    rc->Init(io_thread()->task_runner().Clone(),
+             io_thread()->platform_handle_watcher(), &delegate);
+  });
 
   // Close the handle of the other end, which should stuff fail.
   handles[1].reset();
 
   // Wait for the delegate, which will shut the |RawChannel| down.
   delegate.Wait();
 }
 
 TEST_F(RawChannelTest, ShutdownAndDestroyOnErrorRead) {
   RawChannel* rc = RawChannel::Create(handles[0].Pass()).release();
   ShutdownOnErrorRawChannelDelegate delegate(
       rc, true, RawChannel::Delegate::ERROR_READ_SHUTDOWN);
-  io_thread()->PostTaskAndWait([rc, &delegate]() { rc->Init(&delegate); });
+  io_thread()->PostTaskAndWait([this, &rc, &delegate]() {
+    rc->Init(io_thread()->task_runner().Clone(),
+             io_thread()->platform_handle_watcher(), &delegate);
+  });
 
   // Close the handle of the other end, which should stuff fail.
   handles[1].reset();
 
   // Wait for the delegate, which will shut the |RawChannel| down.
   delegate.Wait();
 }
 
 TEST_F(RawChannelTest, ShutdownOnErrorWrite) {
   std::unique_ptr<RawChannel> rc(RawChannel::Create(handles[0].Pass()));
   ShutdownOnErrorRawChannelDelegate delegate(rc.get(), false,
                                              RawChannel::Delegate::ERROR_WRITE);
-  io_thread()->PostTaskAndWait([&rc, &delegate]() { rc->Init(&delegate); });
+  io_thread()->PostTaskAndWait([this, &rc, &delegate]() {
+    rc->Init(io_thread()->task_runner().Clone(),
+             io_thread()->platform_handle_watcher(), &delegate);
+  });
 
   // Close the handle of the other end, which should stuff fail.
   handles[1].reset();
 
   EXPECT_FALSE(rc->WriteMessage(MakeTestMessage(1)));
 
   // Wait for the delegate, which will shut the |RawChannel| down.
   delegate.Wait();
 }
 
 TEST_F(RawChannelTest, ShutdownAndDestroyOnErrorWrite) {
   RawChannel* rc = RawChannel::Create(handles[0].Pass()).release();
   ShutdownOnErrorRawChannelDelegate delegate(rc, true,
                                              RawChannel::Delegate::ERROR_WRITE);
-  io_thread()->PostTaskAndWait([rc, &delegate]() { rc->Init(&delegate); });
+  io_thread()->PostTaskAndWait([this, &rc, &delegate]() {
+    rc->Init(io_thread()->task_runner().Clone(),
+             io_thread()->platform_handle_watcher(), &delegate);
+  });
 
   // Close the handle of the other end, which should stuff fail.
   handles[1].reset();
 
   EXPECT_FALSE(rc->WriteMessage(MakeTestMessage(1)));
 
   // Wait for the delegate, which will shut the |RawChannel| down.
   delegate.Wait();
 }
 
@@ skipping 54 matching lines @@
   AutoResetWaitableEvent done_event_;
 
   MOJO_DISALLOW_COPY_AND_ASSIGN(ReadPlatformHandlesCheckerRawChannelDelegate);
 };
 
 TEST_F(RawChannelTest, ReadWritePlatformHandles) {
   test::ScopedTestDir test_dir;
 
   WriteOnlyRawChannelDelegate write_delegate;
   std::unique_ptr<RawChannel> rc_write(RawChannel::Create(handles[0].Pass()));
-  io_thread()->PostTaskAndWait(
-      [&rc_write, &write_delegate]() { rc_write->Init(&write_delegate); });
+  io_thread()->PostTaskAndWait([this, &rc_write, &write_delegate]() {
+    rc_write->Init(io_thread()->task_runner().Clone(),
+                   io_thread()->platform_handle_watcher(), &write_delegate);
+  });
 
   ReadPlatformHandlesCheckerRawChannelDelegate read_delegate;
   std::unique_ptr<RawChannel> rc_read(RawChannel::Create(handles[1].Pass()));
-  io_thread()->PostTaskAndWait(
-      [&rc_read, &read_delegate]() { rc_read->Init(&read_delegate); });
+  io_thread()->PostTaskAndWait([this, &rc_read, &read_delegate]() {
+    rc_read->Init(io_thread()->task_runner().Clone(),
+                  io_thread()->platform_handle_watcher(), &read_delegate);
+  });
 
   util::ScopedFILE fp1(test_dir.CreateFile());
   EXPECT_EQ(1u, fwrite("1", 1, 1, fp1.get()));
   util::ScopedFILE fp2(test_dir.CreateFile());
   EXPECT_EQ(1u, fwrite("2", 1, 1, fp2.get()));
 
   {
     const char kHello[] = "hello";
     auto platform_handles = MakeUnique<std::vector<ScopedPlatformHandle>>();
     platform_handles->push_back(
@@ skipping 13 matching lines @@
 
   read_delegate.Wait();
 
   io_thread()->PostTaskAndWait([&rc_read]() { rc_read->Shutdown(); });
   io_thread()->PostTaskAndWait([&rc_write]() { rc_write->Shutdown(); });
 }
 
 }  // namespace
 }  // namespace system
 }  // namespace mojo