Home | History | Annotate | Download | only in system 
      1 // Copyright 2014 The Chromium Authors. All rights reserved.
      2 // Use of this source code is governed by a BSD-style license that can be
      3 // found in the LICENSE file.
      4 
      5 #include "mojo/system/raw_channel.h"
      6 
      7 #include <stdint.h>
      8 
      9 #include <vector>
     10 
     11 #include "base/bind.h"
     12 #include "base/location.h"
     13 #include "base/logging.h"
     14 #include "base/macros.h"
     15 #include "base/memory/scoped_ptr.h"
     16 #include "base/memory/scoped_vector.h"
     17 #include "base/rand_util.h"
     18 #include "base/synchronization/lock.h"
     19 #include "base/synchronization/waitable_event.h"
     20 #include "base/threading/platform_thread.h"  // For |Sleep()|.
     21 #include "base/threading/simple_thread.h"
     22 #include "base/time/time.h"
     23 #include "build/build_config.h"
     24 #include "mojo/common/test/test_utils.h"
     25 #include "mojo/embedder/platform_channel_pair.h"
     26 #include "mojo/embedder/platform_handle.h"
     27 #include "mojo/embedder/scoped_platform_handle.h"
     28 #include "mojo/system/message_in_transit.h"
     29 #include "mojo/system/test_utils.h"
     30 #include "testing/gtest/include/gtest/gtest.h"
     31 
     32 namespace mojo {
     33 namespace system {
     34 namespace {
     35 
     36 scoped_ptr<MessageInTransit> MakeTestMessage(uint32_t num_bytes) {
     37   std::vector<unsigned char> bytes(num_bytes, 0);
     38   for (size_t i = 0; i < num_bytes; i++)
     39     bytes[i] = static_cast<unsigned char>(i + num_bytes);
     40   return make_scoped_ptr(
     41       new MessageInTransit(MessageInTransit::kTypeMessagePipeEndpoint,
     42                            MessageInTransit::kSubtypeMessagePipeEndpointData,
     43                            num_bytes, bytes.empty() ? NULL : &bytes[0]));
     44 }
     45 
     46 bool CheckMessageData(const void* bytes, uint32_t num_bytes) {
     47   const unsigned char* b = static_cast<const unsigned char*>(bytes);
     48   for (uint32_t i = 0; i < num_bytes; i++) {
     49     if (b[i] != static_cast<unsigned char>(i + num_bytes))
     50       return false;
     51   }
     52   return true;
     53 }
     54 
     55 void InitOnIOThread(RawChannel* raw_channel, RawChannel::Delegate* delegate) {
     56   CHECK(raw_channel->Init(delegate));
     57 }
     58 
     59 bool WriteTestMessageToHandle(const embedder::PlatformHandle& handle,
     60                               uint32_t num_bytes) {
     61   scoped_ptr<MessageInTransit> message(MakeTestMessage(num_bytes));
     62 
     63   size_t write_size = 0;
     64   mojo::test::BlockingWrite(
     65       handle, message->main_buffer(), message->main_buffer_size(), &write_size);
     66   return write_size == message->main_buffer_size();
     67 }
     68 
     69 // -----------------------------------------------------------------------------
     70 
     71 class RawChannelTest : public testing::Test {
     72  public:
     73   RawChannelTest() : io_thread_(test::TestIOThread::kManualStart) {}
     74   virtual ~RawChannelTest() {}
     75 
     76   virtual void SetUp() OVERRIDE {
     77     embedder::PlatformChannelPair channel_pair;
     78     handles[0] = channel_pair.PassServerHandle();
     79     handles[1] = channel_pair.PassClientHandle();
     80     io_thread_.Start();
     81   }
     82 
     83   virtual void TearDown() OVERRIDE {
     84     io_thread_.Stop();
     85     handles[0].reset();
     86     handles[1].reset();
     87   }
     88 
     89  protected:
     90   test::TestIOThread* io_thread() { return &io_thread_; }
     91 
     92   embedder::ScopedPlatformHandle handles[2];
     93 
     94  private:
     95   test::TestIOThread io_thread_;
     96 
     97   DISALLOW_COPY_AND_ASSIGN(RawChannelTest);
     98 };
     99 
    100 // RawChannelTest.WriteMessage -------------------------------------------------
    101 
    102 class WriteOnlyRawChannelDelegate : public RawChannel::Delegate {
    103  public:
    104   WriteOnlyRawChannelDelegate() {}
    105   virtual ~WriteOnlyRawChannelDelegate() {}
    106 
    107   // |RawChannel::Delegate| implementation:
    108   virtual void OnReadMessage(
    109       const MessageInTransit::View& /*message_view*/,
    110       embedder::ScopedPlatformHandleVectorPtr /*platform_handles*/) OVERRIDE {
    111     CHECK(false);  // Should not get called.
    112   }
    113   virtual void OnFatalError(FatalError fatal_error) OVERRIDE {
    114     // We'll get a read error when the connection is closed.
    115     CHECK_EQ(fatal_error, FATAL_ERROR_READ);
    116   }
    117 
    118  private:
    119   DISALLOW_COPY_AND_ASSIGN(WriteOnlyRawChannelDelegate);
    120 };
    121 
    122 static const int64_t kMessageReaderSleepMs = 1;
    123 static const size_t kMessageReaderMaxPollIterations = 3000;
    124 
    125 class TestMessageReaderAndChecker {
    126  public:
    127   explicit TestMessageReaderAndChecker(embedder::PlatformHandle handle)
    128       : handle_(handle) {}
    129   ~TestMessageReaderAndChecker() { CHECK(bytes_.empty()); }
    130 
    131   bool ReadAndCheckNextMessage(uint32_t expected_size) {
    132     unsigned char buffer[4096];
    133 
    134     for (size_t i = 0; i < kMessageReaderMaxPollIterations;) {
    135       size_t read_size = 0;
    136       CHECK(mojo::test::NonBlockingRead(handle_, buffer, sizeof(buffer),
    137                                         &read_size));
    138 
    139       // Append newly-read data to |bytes_|.
    140       bytes_.insert(bytes_.end(), buffer, buffer + read_size);
    141 
    142       // If we have the header....
    143       size_t message_size;
    144       if (MessageInTransit::GetNextMessageSize(
    145               bytes_.empty() ? NULL : &bytes_[0],
    146               bytes_.size(),
    147               &message_size)) {
    148         // If we've read the whole message....
    149         if (bytes_.size() >= message_size) {
    150           bool rv = true;
    151           MessageInTransit::View message_view(message_size, &bytes_[0]);
    152           CHECK_EQ(message_view.main_buffer_size(), message_size);
    153 
    154           if (message_view.num_bytes() != expected_size) {
    155             LOG(ERROR) << "Wrong size: " << message_size << " instead of "
    156                        << expected_size << " bytes.";
    157             rv = false;
    158           } else if (!CheckMessageData(message_view.bytes(),
    159                                        message_view.num_bytes())) {
    160             LOG(ERROR) << "Incorrect message bytes.";
    161             rv = false;
    162           }
    163 
    164           // Erase message data.
    165           bytes_.erase(bytes_.begin(),
    166                        bytes_.begin() +
    167                            message_view.main_buffer_size());
    168           return rv;
    169         }
    170       }
    171 
    172       if (static_cast<size_t>(read_size) < sizeof(buffer)) {
    173         i++;
    174         base::PlatformThread::Sleep(
    175             base::TimeDelta::FromMilliseconds(kMessageReaderSleepMs));
    176       }
    177     }
    178 
    179     LOG(ERROR) << "Too many iterations.";
    180     return false;
    181   }
    182 
    183  private:
    184   const embedder::PlatformHandle handle_;
    185 
    186   // The start of the received data should always be on a message boundary.
    187   std::vector<unsigned char> bytes_;
    188 
    189   DISALLOW_COPY_AND_ASSIGN(TestMessageReaderAndChecker);
    190 };
    191 
    192 // Tests writing (and verifies reading using our own custom reader).
    193 TEST_F(RawChannelTest, WriteMessage) {
    194   WriteOnlyRawChannelDelegate delegate;
    195   scoped_ptr<RawChannel> rc(RawChannel::Create(handles[0].Pass()));
    196   TestMessageReaderAndChecker checker(handles[1].get());
    197   io_thread()->PostTaskAndWait(FROM_HERE,
    198                                base::Bind(&InitOnIOThread, rc.get(),
    199                                           base::Unretained(&delegate)));
    200 
    201   // Write and read, for a variety of sizes.
    202   for (uint32_t size = 1; size < 5 * 1000 * 1000; size += size / 2 + 1) {
    203     EXPECT_TRUE(rc->WriteMessage(MakeTestMessage(size)));
    204     EXPECT_TRUE(checker.ReadAndCheckNextMessage(size)) << size;
    205   }
    206 
    207   // Write/queue and read afterwards, for a variety of sizes.
    208   for (uint32_t size = 1; size < 5 * 1000 * 1000; size += size / 2 + 1)
    209     EXPECT_TRUE(rc->WriteMessage(MakeTestMessage(size)));
    210   for (uint32_t size = 1; size < 5 * 1000 * 1000; size += size / 2 + 1)
    211     EXPECT_TRUE(checker.ReadAndCheckNextMessage(size)) << size;
    212 
    213   io_thread()->PostTaskAndWait(FROM_HERE,
    214                                base::Bind(&RawChannel::Shutdown,
    215                                           base::Unretained(rc.get())));
    216 }
    217 
    218 // RawChannelTest.OnReadMessage ------------------------------------------------
    219 
    220 class ReadCheckerRawChannelDelegate : public RawChannel::Delegate {
    221  public:
    222   ReadCheckerRawChannelDelegate()
    223       : done_event_(false, false),
    224         position_(0) {}
    225   virtual ~ReadCheckerRawChannelDelegate() {}
    226 
    227   // |RawChannel::Delegate| implementation (called on the I/O thread):
    228   virtual void OnReadMessage(
    229       const MessageInTransit::View& message_view,
    230       embedder::ScopedPlatformHandleVectorPtr platform_handles) OVERRIDE {
    231     EXPECT_FALSE(platform_handles);
    232 
    233     size_t position;
    234     size_t expected_size;
    235     bool should_signal = false;
    236     {
    237       base::AutoLock locker(lock_);
    238       CHECK_LT(position_, expected_sizes_.size());
    239       position = position_;
    240       expected_size = expected_sizes_[position];
    241       position_++;
    242       if (position_ >= expected_sizes_.size())
    243         should_signal = true;
    244     }
    245 
    246     EXPECT_EQ(expected_size, message_view.num_bytes()) << position;
    247     if (message_view.num_bytes() == expected_size) {
    248       EXPECT_TRUE(CheckMessageData(message_view.bytes(),
    249                   message_view.num_bytes())) << position;
    250     }
    251 
    252     if (should_signal)
    253       done_event_.Signal();
    254   }
    255   virtual void OnFatalError(FatalError fatal_error) OVERRIDE {
    256     // We'll get a read error when the connection is closed.
    257     CHECK_EQ(fatal_error, FATAL_ERROR_READ);
    258   }
    259 
    260   // Waits for all the messages (of sizes |expected_sizes_|) to be seen.
    261   void Wait() {
    262     done_event_.Wait();
    263   }
    264 
    265   void SetExpectedSizes(const std::vector<uint32_t>& expected_sizes) {
    266     base::AutoLock locker(lock_);
    267     CHECK_EQ(position_, expected_sizes_.size());
    268     expected_sizes_ = expected_sizes;
    269     position_ = 0;
    270   }
    271 
    272  private:
    273   base::WaitableEvent done_event_;
    274 
    275   base::Lock lock_;  // Protects the following members.
    276   std::vector<uint32_t> expected_sizes_;
    277   size_t position_;
    278 
    279   DISALLOW_COPY_AND_ASSIGN(ReadCheckerRawChannelDelegate);
    280 };
    281 
    282 // Tests reading (writing using our own custom writer).
    283 TEST_F(RawChannelTest, OnReadMessage) {
    284   ReadCheckerRawChannelDelegate delegate;
    285   scoped_ptr<RawChannel> rc(RawChannel::Create(handles[0].Pass()));
    286   io_thread()->PostTaskAndWait(FROM_HERE,
    287                                base::Bind(&InitOnIOThread, rc.get(),
    288                                           base::Unretained(&delegate)));
    289 
    290   // Write and read, for a variety of sizes.
    291   for (uint32_t size = 1; size < 5 * 1000 * 1000; size += size / 2 + 1) {
    292     delegate.SetExpectedSizes(std::vector<uint32_t>(1, size));
    293 
    294     EXPECT_TRUE(WriteTestMessageToHandle(handles[1].get(), size));
    295 
    296     delegate.Wait();
    297   }
    298 
    299   // Set up reader and write as fast as we can.
    300   // Write/queue and read afterwards, for a variety of sizes.
    301   std::vector<uint32_t> expected_sizes;
    302   for (uint32_t size = 1; size < 5 * 1000 * 1000; size += size / 2 + 1)
    303     expected_sizes.push_back(size);
    304   delegate.SetExpectedSizes(expected_sizes);
    305   for (uint32_t size = 1; size < 5 * 1000 * 1000; size += size / 2 + 1)
    306     EXPECT_TRUE(WriteTestMessageToHandle(handles[1].get(), size));
    307   delegate.Wait();
    308 
    309   io_thread()->PostTaskAndWait(FROM_HERE,
    310                                base::Bind(&RawChannel::Shutdown,
    311                                           base::Unretained(rc.get())));
    312 }
    313 
    314 // RawChannelTest.WriteMessageAndOnReadMessage ---------------------------------
    315 
    316 class RawChannelWriterThread : public base::SimpleThread {
    317  public:
    318   RawChannelWriterThread(RawChannel* raw_channel, size_t write_count)
    319       : base::SimpleThread("raw_channel_writer_thread"),
    320         raw_channel_(raw_channel),
    321         left_to_write_(write_count) {
    322   }
    323 
    324   virtual ~RawChannelWriterThread() {
    325     Join();
    326   }
    327 
    328  private:
    329   virtual void Run() OVERRIDE {
    330     static const int kMaxRandomMessageSize = 25000;
    331 
    332     while (left_to_write_-- > 0) {
    333       EXPECT_TRUE(raw_channel_->WriteMessage(MakeTestMessage(
    334           static_cast<uint32_t>(base::RandInt(1, kMaxRandomMessageSize)))));
    335     }
    336   }
    337 
    338   RawChannel* const raw_channel_;
    339   size_t left_to_write_;
    340 
    341   DISALLOW_COPY_AND_ASSIGN(RawChannelWriterThread);
    342 };
    343 
    344 class ReadCountdownRawChannelDelegate : public RawChannel::Delegate {
    345  public:
    346   explicit ReadCountdownRawChannelDelegate(size_t expected_count)
    347       : done_event_(false, false),
    348         expected_count_(expected_count),
    349         count_(0) {}
    350   virtual ~ReadCountdownRawChannelDelegate() {}
    351 
    352   // |RawChannel::Delegate| implementation (called on the I/O thread):
    353   virtual void OnReadMessage(
    354       const MessageInTransit::View& message_view,
    355       embedder::ScopedPlatformHandleVectorPtr platform_handles) OVERRIDE {
    356     EXPECT_FALSE(platform_handles);
    357 
    358     EXPECT_LT(count_, expected_count_);
    359     count_++;
    360 
    361     EXPECT_TRUE(CheckMessageData(message_view.bytes(),
    362                 message_view.num_bytes()));
    363 
    364     if (count_ >= expected_count_)
    365       done_event_.Signal();
    366   }
    367   virtual void OnFatalError(FatalError fatal_error) OVERRIDE {
    368     // We'll get a read error when the connection is closed.
    369     CHECK_EQ(fatal_error, FATAL_ERROR_READ);
    370   }
    371 
    372   // Waits for all the messages to have been seen.
    373   void Wait() {
    374     done_event_.Wait();
    375   }
    376 
    377  private:
    378   base::WaitableEvent done_event_;
    379   size_t expected_count_;
    380   size_t count_;
    381 
    382   DISALLOW_COPY_AND_ASSIGN(ReadCountdownRawChannelDelegate);
    383 };
    384 
    385 TEST_F(RawChannelTest, WriteMessageAndOnReadMessage) {
    386   static const size_t kNumWriterThreads = 10;
    387   static const size_t kNumWriteMessagesPerThread = 4000;
    388 
    389   WriteOnlyRawChannelDelegate writer_delegate;
    390   scoped_ptr<RawChannel> writer_rc(RawChannel::Create(handles[0].Pass()));
    391   io_thread()->PostTaskAndWait(FROM_HERE,
    392                                base::Bind(&InitOnIOThread, writer_rc.get(),
    393                                           base::Unretained(&writer_delegate)));
    394 
    395   ReadCountdownRawChannelDelegate reader_delegate(
    396       kNumWriterThreads * kNumWriteMessagesPerThread);
    397   scoped_ptr<RawChannel> reader_rc(RawChannel::Create(handles[1].Pass()));
    398   io_thread()->PostTaskAndWait(FROM_HERE,
    399                                base::Bind(&InitOnIOThread, reader_rc.get(),
    400                                           base::Unretained(&reader_delegate)));
    401 
    402   {
    403     ScopedVector<RawChannelWriterThread> writer_threads;
    404     for (size_t i = 0; i < kNumWriterThreads; i++) {
    405       writer_threads.push_back(new RawChannelWriterThread(
    406           writer_rc.get(), kNumWriteMessagesPerThread));
    407     }
    408     for (size_t i = 0; i < writer_threads.size(); i++)
    409       writer_threads[i]->Start();
    410   }  // Joins all the writer threads.
    411 
    412   // Sleep a bit, to let any extraneous reads be processed. (There shouldn't be
    413   // any, but we want to know about them.)
    414   base::PlatformThread::Sleep(base::TimeDelta::FromMilliseconds(100));
    415 
    416   // Wait for reading to finish.
    417   reader_delegate.Wait();
    418 
    419   io_thread()->PostTaskAndWait(FROM_HERE,
    420                                base::Bind(&RawChannel::Shutdown,
    421                                           base::Unretained(reader_rc.get())));
    422 
    423   io_thread()->PostTaskAndWait(FROM_HERE,
    424                                base::Bind(&RawChannel::Shutdown,
    425                                           base::Unretained(writer_rc.get())));
    426 }
    427 
    428 // RawChannelTest.OnFatalError -------------------------------------------------
    429 
    430 class FatalErrorRecordingRawChannelDelegate
    431     : public ReadCountdownRawChannelDelegate {
    432  public:
    433   FatalErrorRecordingRawChannelDelegate(size_t expected_read_count,
    434                                         bool expect_read_error,
    435                                         bool expect_write_error)
    436       : ReadCountdownRawChannelDelegate(expected_read_count),
    437         got_read_fatal_error_event_(false, false),
    438         got_write_fatal_error_event_(false, false),
    439         expecting_read_error_(expect_read_error),
    440         expecting_write_error_(expect_write_error) {
    441   }
    442 
    443   virtual ~FatalErrorRecordingRawChannelDelegate() {}
    444 
    445   virtual void OnFatalError(FatalError fatal_error) OVERRIDE {
    446     switch (fatal_error) {
    447       case FATAL_ERROR_READ:
    448         ASSERT_TRUE(expecting_read_error_);
    449         expecting_read_error_ = false;
    450         got_read_fatal_error_event_.Signal();
    451         break;
    452       case FATAL_ERROR_WRITE:
    453         ASSERT_TRUE(expecting_write_error_);
    454         expecting_write_error_ = false;
    455         got_write_fatal_error_event_.Signal();
    456         break;
    457     }
    458   }
    459 
    460   void WaitForReadFatalError() { got_read_fatal_error_event_.Wait(); }
    461   void WaitForWriteFatalError() { got_write_fatal_error_event_.Wait(); }
    462 
    463  private:
    464   base::WaitableEvent got_read_fatal_error_event_;
    465   base::WaitableEvent got_write_fatal_error_event_;
    466 
    467   bool expecting_read_error_;
    468   bool expecting_write_error_;
    469 
    470   DISALLOW_COPY_AND_ASSIGN(FatalErrorRecordingRawChannelDelegate);
    471 };
    472 
    473 // Tests fatal errors.
    474 TEST_F(RawChannelTest, OnFatalError) {
    475   FatalErrorRecordingRawChannelDelegate delegate(0, true, true);
    476   scoped_ptr<RawChannel> rc(RawChannel::Create(handles[0].Pass()));
    477   io_thread()->PostTaskAndWait(FROM_HERE,
    478                                base::Bind(&InitOnIOThread, rc.get(),
    479                                           base::Unretained(&delegate)));
    480 
    481   // Close the handle of the other end, which should make writing fail.
    482   handles[1].reset();
    483 
    484   EXPECT_FALSE(rc->WriteMessage(MakeTestMessage(1)));
    485 
    486   // We should get a write fatal error.
    487   delegate.WaitForWriteFatalError();
    488 
    489   // We should also get a read fatal error.
    490   delegate.WaitForReadFatalError();
    491 
    492   EXPECT_FALSE(rc->WriteMessage(MakeTestMessage(2)));
    493 
    494   // Sleep a bit, to make sure we don't get another |OnFatalError()|
    495   // notification. (If we actually get another one, |OnFatalError()| crashes.)
    496   base::PlatformThread::Sleep(base::TimeDelta::FromMilliseconds(20));
    497 
    498   io_thread()->PostTaskAndWait(FROM_HERE,
    499                                base::Bind(&RawChannel::Shutdown,
    500                                           base::Unretained(rc.get())));
    501 }
    502 
    503 // RawChannelTest.ReadUnaffectedByWriteFatalError ------------------------------
    504 
    505 TEST_F(RawChannelTest, ReadUnaffectedByWriteFatalError) {
    506   const size_t kMessageCount = 5;
    507 
    508   // Write a few messages into the other end.
    509   uint32_t message_size = 1;
    510   for (size_t i = 0; i < kMessageCount;
    511        i++, message_size += message_size / 2 + 1)
    512     EXPECT_TRUE(WriteTestMessageToHandle(handles[1].get(), message_size));
    513 
    514   // Close the other end, which should make writing fail.
    515   handles[1].reset();
    516 
    517   // Only start up reading here. The system buffer should still contain the
    518   // messages that were written.
    519   FatalErrorRecordingRawChannelDelegate delegate(kMessageCount, true, true);
    520   scoped_ptr<RawChannel> rc(RawChannel::Create(handles[0].Pass()));
    521   io_thread()->PostTaskAndWait(FROM_HERE,
    522                                base::Bind(&InitOnIOThread, rc.get(),
    523                                           base::Unretained(&delegate)));
    524 
    525   EXPECT_FALSE(rc->WriteMessage(MakeTestMessage(1)));
    526 
    527   // We should definitely get a write fatal error.
    528   delegate.WaitForWriteFatalError();
    529 
    530   // Wait for reading to finish. A writing failure shouldn't affect reading.
    531   delegate.Wait();
    532 
    533   // And then we should get a read fatal error.
    534   delegate.WaitForReadFatalError();
    535 
    536   io_thread()->PostTaskAndWait(FROM_HERE,
    537                                base::Bind(&RawChannel::Shutdown,
    538                                           base::Unretained(rc.get())));
    539 }
    540 
    541 // RawChannelTest.WriteMessageAfterShutdown ------------------------------------
    542 
    543 // Makes sure that calling |WriteMessage()| after |Shutdown()| behaves
    544 // correctly.
    545 TEST_F(RawChannelTest, WriteMessageAfterShutdown) {
    546   WriteOnlyRawChannelDelegate delegate;
    547   scoped_ptr<RawChannel> rc(RawChannel::Create(handles[0].Pass()));
    548   io_thread()->PostTaskAndWait(FROM_HERE,
    549                                base::Bind(&InitOnIOThread, rc.get(),
    550                                           base::Unretained(&delegate)));
    551   io_thread()->PostTaskAndWait(FROM_HERE,
    552                                base::Bind(&RawChannel::Shutdown,
    553                                           base::Unretained(rc.get())));
    554 
    555   EXPECT_FALSE(rc->WriteMessage(MakeTestMessage(1)));
    556 }
    557 
    558 // RawChannelTest.ShutdownOnReadMessage ----------------------------------------
    559 
    560 class ShutdownOnReadMessageRawChannelDelegate : public RawChannel::Delegate {
    561  public:
    562   explicit ShutdownOnReadMessageRawChannelDelegate(RawChannel* raw_channel)
    563       : raw_channel_(raw_channel),
    564         done_event_(false, false),
    565         did_shutdown_(false) {}
    566   virtual ~ShutdownOnReadMessageRawChannelDelegate() {}
    567 
    568   // |RawChannel::Delegate| implementation (called on the I/O thread):
    569   virtual void OnReadMessage(
    570       const MessageInTransit::View& message_view,
    571       embedder::ScopedPlatformHandleVectorPtr platform_handles) OVERRIDE {
    572     EXPECT_FALSE(platform_handles);
    573     EXPECT_FALSE(did_shutdown_);
    574     EXPECT_TRUE(CheckMessageData(message_view.bytes(),
    575                 message_view.num_bytes()));
    576     raw_channel_->Shutdown();
    577     did_shutdown_ = true;
    578     done_event_.Signal();
    579   }
    580   virtual void OnFatalError(FatalError /*fatal_error*/) OVERRIDE {
    581     CHECK(false);  // Should not get called.
    582   }
    583 
    584   // Waits for shutdown.
    585   void Wait() {
    586     done_event_.Wait();
    587     EXPECT_TRUE(did_shutdown_);
    588   }
    589 
    590  private:
    591   RawChannel* const raw_channel_;
    592   base::WaitableEvent done_event_;
    593   bool did_shutdown_;
    594 
    595   DISALLOW_COPY_AND_ASSIGN(ShutdownOnReadMessageRawChannelDelegate);
    596 };
    597 
    598 TEST_F(RawChannelTest, ShutdownOnReadMessage) {
    599   // Write a few messages into the other end.
    600   for (size_t count = 0; count < 5; count++)
    601     EXPECT_TRUE(WriteTestMessageToHandle(handles[1].get(), 10));
    602 
    603   scoped_ptr<RawChannel> rc(RawChannel::Create(handles[0].Pass()));
    604   ShutdownOnReadMessageRawChannelDelegate delegate(rc.get());
    605   io_thread()->PostTaskAndWait(FROM_HERE,
    606                                base::Bind(&InitOnIOThread, rc.get(),
    607                                           base::Unretained(&delegate)));
    608 
    609   // Wait for the delegate, which will shut the |RawChannel| down.
    610   delegate.Wait();
    611 }
    612 
    613 // RawChannelTest.ShutdownOnFatalError{Read, Write} ----------------------------
    614 
    615 class ShutdownOnFatalErrorRawChannelDelegate : public RawChannel::Delegate {
    616  public:
    617   ShutdownOnFatalErrorRawChannelDelegate(RawChannel* raw_channel,
    618                                          FatalError shutdown_on_error_type)
    619       : raw_channel_(raw_channel),
    620         shutdown_on_error_type_(shutdown_on_error_type),
    621         done_event_(false, false),
    622         did_shutdown_(false) {}
    623   virtual ~ShutdownOnFatalErrorRawChannelDelegate() {}
    624 
    625   // |RawChannel::Delegate| implementation (called on the I/O thread):
    626   virtual void OnReadMessage(
    627       const MessageInTransit::View& /*message_view*/,
    628       embedder::ScopedPlatformHandleVectorPtr /*platform_handles*/) OVERRIDE {
    629     CHECK(false);  // Should not get called.
    630   }
    631   virtual void OnFatalError(FatalError fatal_error) OVERRIDE {
    632     EXPECT_FALSE(did_shutdown_);
    633     if (fatal_error != shutdown_on_error_type_)
    634       return;
    635     raw_channel_->Shutdown();
    636     did_shutdown_ = true;
    637     done_event_.Signal();
    638   }
    639 
    640   // Waits for shutdown.
    641   void Wait() {
    642     done_event_.Wait();
    643     EXPECT_TRUE(did_shutdown_);
    644   }
    645 
    646  private:
    647   RawChannel* const raw_channel_;
    648   const FatalError shutdown_on_error_type_;
    649   base::WaitableEvent done_event_;
    650   bool did_shutdown_;
    651 
    652   DISALLOW_COPY_AND_ASSIGN(ShutdownOnFatalErrorRawChannelDelegate);
    653 };
    654 
    655 TEST_F(RawChannelTest, ShutdownOnFatalErrorRead) {
    656   scoped_ptr<RawChannel> rc(RawChannel::Create(handles[0].Pass()));
    657   ShutdownOnFatalErrorRawChannelDelegate delegate(
    658       rc.get(), RawChannel::Delegate::FATAL_ERROR_READ);
    659   io_thread()->PostTaskAndWait(FROM_HERE,
    660                                base::Bind(&InitOnIOThread, rc.get(),
    661                                           base::Unretained(&delegate)));
    662 
    663   // Close the handle of the other end, which should stuff fail.
    664   handles[1].reset();
    665 
    666   // Wait for the delegate, which will shut the |RawChannel| down.
    667   delegate.Wait();
    668 }
    669 
    670 TEST_F(RawChannelTest, ShutdownOnFatalErrorWrite) {
    671   scoped_ptr<RawChannel> rc(RawChannel::Create(handles[0].Pass()));
    672   ShutdownOnFatalErrorRawChannelDelegate delegate(
    673       rc.get(), RawChannel::Delegate::FATAL_ERROR_WRITE);
    674   io_thread()->PostTaskAndWait(FROM_HERE,
    675                                base::Bind(&InitOnIOThread, rc.get(),
    676                                           base::Unretained(&delegate)));
    677 
    678   // Close the handle of the other end, which should stuff fail.
    679   handles[1].reset();
    680 
    681   EXPECT_FALSE(rc->WriteMessage(MakeTestMessage(1)));
    682 
    683   // Wait for the delegate, which will shut the |RawChannel| down.
    684   delegate.Wait();
    685 }
    686 
    687 }  // namespace
    688 }  // namespace system
    689 }  // namespace mojo
    690