mojo/system/message_pipe_test_utils.cc

   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/message_pipe_test_utils.h"
   6
   7 #include "base/bind.h"
   8 #include "base/threading/platform_thread.h"  // For |Sleep()|.
   9 #include "mojo/system/waiter.h"
  10
  11 namespace mojo {
  12 namespace system {
  13 namespace test {
  14
  15 MojoResult WaitIfNecessary(scoped_refptr<MessagePipe> mp,
  16                            MojoHandleSignals signals,
  17                            HandleSignalsState* signals_state) {
  18   Waiter waiter;
  19   waiter.Init();
  20
  21   MojoResult add_result = mp->AddWaiter(0, &waiter, signals, 0, signals_state);
  22   if (add_result != MOJO_RESULT_OK) {
  23     return (add_result == MOJO_RESULT_ALREADY_EXISTS) ? MOJO_RESULT_OK
  24                                                       : add_result;
  25   }
  26
  27   MojoResult wait_result = waiter.Wait(MOJO_DEADLINE_INDEFINITE, NULL);
  28   mp->RemoveWaiter(0, &waiter, signals_state);
  29   return wait_result;
  30 }
  31
  32 ChannelThread::ChannelThread(embedder::PlatformSupport* platform_support)
  33     : platform_support_(platform_support),
  34       test_io_thread_(test::TestIOThread::kManualStart) {
  35 }
  36
  37 ChannelThread::~ChannelThread() {
  38   Stop();
  39 }
  40
  41 void ChannelThread::Start(embedder::ScopedPlatformHandle platform_handle,
  42                           scoped_refptr<MessagePipe> message_pipe) {
  43   test_io_thread_.Start();
  44   test_io_thread_.PostTaskAndWait(
  45       FROM_HERE,
  46       base::Bind(&ChannelThread::InitChannelOnIOThread,
  47                  base::Unretained(this),
  48                  base::Passed(&platform_handle),
  49                  message_pipe));
  50 }
  51
  52 void ChannelThread::Stop() {
  53   if (channel_.get()) {
  54     // Hack to flush write buffers before quitting.
  55     // TODO(vtl): Remove this once |Channel| has a
  56     // |FlushWriteBufferAndShutdown()| (or whatever).
  57     while (!channel_->IsWriteBufferEmpty())
  58       base::PlatformThread::Sleep(base::TimeDelta::FromMilliseconds(20));
  59
  60     test_io_thread_.PostTaskAndWait(
  61         FROM_HERE,
  62         base::Bind(&ChannelThread::ShutdownChannelOnIOThread,
  63                    base::Unretained(this)));
  64   }
  65   test_io_thread_.Stop();
  66 }
  67
  68 void ChannelThread::InitChannelOnIOThread(
  69     embedder::ScopedPlatformHandle platform_handle,
  70     scoped_refptr<MessagePipe> message_pipe) {
  71   CHECK_EQ(base::MessageLoop::current(), test_io_thread_.message_loop());
  72   CHECK(platform_handle.is_valid());
  73
  74   // Create and initialize |Channel|.
  75   channel_ = new Channel(platform_support_);
  76   CHECK(channel_->Init(RawChannel::Create(platform_handle.Pass())));
  77
  78   // Attach the message pipe endpoint.
  79   // Note: On the "server" (parent process) side, we need not attach the
  80   // message pipe endpoint immediately. However, on the "client" (child
  81   // process) side, this *must* be done here -- otherwise, the |Channel| may
  82   // receive/process messages (which it can do as soon as it's hooked up to
  83   // the IO thread message loop, and that message loop runs) before the
  84   // message pipe endpoint is attached.
  85   CHECK_EQ(channel_->AttachMessagePipeEndpoint(message_pipe, 1),
  86            Channel::kBootstrapEndpointId);
  87   CHECK(channel_->RunMessagePipeEndpoint(Channel::kBootstrapEndpointId,
  88                                          Channel::kBootstrapEndpointId));
  89 }
  90
  91 void ChannelThread::ShutdownChannelOnIOThread() {
  92   CHECK(channel_.get());
  93   channel_->Shutdown();
  94   channel_ = NULL;
  95 }
  96
  97 MultiprocessMessagePipeTestBase::MultiprocessMessagePipeTestBase()
  98     : channel_thread_(&platform_support_) {
  99 }
 100
 101 MultiprocessMessagePipeTestBase::~MultiprocessMessagePipeTestBase() {
 102 }
 103
 104 void MultiprocessMessagePipeTestBase::Init(scoped_refptr<MessagePipe> mp) {
 105   channel_thread_.Start(helper_.server_platform_handle.Pass(), mp);
 106 }
 107
 108 }  // namespace test
 109 }  // namespace system
 110 }  // namespace mojo