Mojo: Add the ability to hook up a channel to the embedder API.

That is, starting with an OS "channel" (e.g., socket), you can tell the
system to set up to send messages over it. It'll provide an initial
"bootstrap" message pipe, from which you'll eventually be able to bring
up more message pipes multiplexed on that channel (once you can pass
handles in messages over the channel).

Also:
- Make a base class for tests that need an IO thread.
- Add a way for test embedders to shut down the global instance.

R=darin@chromium.org

Review URL: https://codereview.chromium.org/140403002

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@245251 0039d316-1c4b-4281-b951-d872f2087c98

16 files changed, 372 insertions, 77 deletions

diff --git a/mojo/common/test/multiprocess_test_base_unittest.cc b/mojo/common/test/multiprocess_test_base_unittest.cc
index 6c8d756..114d22c 100644
--- a/mojo/common/test/multiprocess_test_base_unittest.cc
+++ b/mojo/common/test/multiprocess_test_base_unittest.cc
@@ -18,8 +18,7 @@
 namespace mojo {
 namespace {
 
-class MultiprocessTestBaseTest : public test::MultiprocessTestBase {
-};
+typedef test::MultiprocessTestBase MultiprocessTestBaseTest;
 
 TEST_F(MultiprocessTestBaseTest, RunChild) {
 // TODO(vtl): Not implemented on Windows yet.
diff --git a/mojo/mojo.gyp b/mojo/mojo.gyp
index 2f7d73a..2643d10 100644
--- a/mojo/mojo.gyp
+++ b/mojo/mojo.gyp
@@ -158,6 +158,7 @@
         'system/core_test_base.h',
         'system/data_pipe_unittest.cc',
         'system/dispatcher_unittest.cc',
+        'system/embedder_unittest.cc',
         'system/local_data_pipe_unittest.cc',
         'system/message_pipe_dispatcher_unittest.cc',
         'system/message_pipe_unittest.cc',
@@ -165,6 +166,8 @@
         'system/raw_channel_posix_unittest.cc',
         'system/remote_message_pipe_posix_unittest.cc',
         'system/simple_dispatcher_unittest.cc',
+        'system/test_embedder.cc',
+        'system/test_embedder.h',
         'system/test_utils.cc',
         'system/test_utils.h',
         'system/waiter_list_unittest.cc',
diff --git a/mojo/public/system/core_private.cc b/mojo/public/system/core_private.cc
index 3b657ee..f0bc3f1 100644
--- a/mojo/public/system/core_private.cc
+++ b/mojo/public/system/core_private.cc
@@ -128,9 +128,20 @@ namespace mojo {
 Core::~Core() {
 }
 
+// static
 void Core::Init(Core* core) {
   assert(!g_core);
   g_core = core;
 }
 
+// static
+Core* Core::Get() {
+  return g_core;
+}
+
+// static
+void Core::Reset() {
+  g_core = NULL;
+}
+
 }  // namespace mojo
diff --git a/mojo/public/system/core_private.h b/mojo/public/system/core_private.h
index 6f6ecc9..b2608ea 100644
--- a/mojo/public/system/core_private.h
+++ b/mojo/public/system/core_private.h
@@ -17,6 +17,8 @@ class MOJO_SYSTEM_EXPORT Core {
   virtual ~Core();
 
   static void Init(Core* core);
+  static Core* Get();
+  static void Reset();
 
   virtual MojoTimeTicks GetTimeTicksNow() = 0;
   virtual MojoResult Close(MojoHandle handle) = 0;
diff --git a/mojo/system/core_impl.cc b/mojo/system/core_impl.cc
index f356321..79d7ea3 100644
--- a/mojo/system/core_impl.cc
+++ b/mojo/system/core_impl.cc
@@ -85,9 +85,19 @@ CoreImpl::HandleTableEntry::~HandleTableEntry() {
   DCHECK(!busy);
 }
 
-// static
-void CoreImpl::Init() {
-  Core::Init(new CoreImpl());
+CoreImpl::CoreImpl()
+    : next_handle_(MOJO_HANDLE_INVALID + 1) {
+}
+
+CoreImpl::~CoreImpl() {
+  // This should usually not be reached (the singleton lives forever), except in
+  // tests.
+}
+
+MojoHandle CoreImpl::AddDispatcher(
+    const scoped_refptr<Dispatcher>& dispatcher) {
+  base::AutoLock locker(handle_table_lock_);
+  return AddDispatcherNoLock(dispatcher);
 }
 
 MojoTimeTicks CoreImpl::GetTimeTicksNow() {
@@ -471,15 +481,6 @@ MojoResult CoreImpl::EndReadData(MojoHandle data_pipe_consumer_handle,
   return dispatcher->EndReadData(num_bytes_read);
 }
 
-CoreImpl::CoreImpl()
-    : next_handle_(MOJO_HANDLE_INVALID + 1) {
-}
-
-CoreImpl::~CoreImpl() {
-  // This should usually not be reached (the singleton lives forever), except in
-  // tests.
-}
-
 scoped_refptr<Dispatcher> CoreImpl::GetDispatcher(MojoHandle handle) {
   if (handle == MOJO_HANDLE_INVALID)
     return NULL;
diff --git a/mojo/system/core_impl.h b/mojo/system/core_impl.h
index 8a15eb1..f7533a6 100644
--- a/mojo/system/core_impl.h
+++ b/mojo/system/core_impl.h
@@ -15,24 +15,19 @@
 
 namespace mojo {
 
-namespace embedder {
-void Init();  // So it can be friended.
-}
-
 namespace system {
 
 class CoreImpl;
 class Dispatcher;
 
-namespace test {
-class CoreTestBase;
-}
-
 // |CoreImpl| is a singleton object that implements the Mojo system calls. All
 // public methods are thread-safe.
 class MOJO_SYSTEM_IMPL_EXPORT CoreImpl : public Core {
  public:
-  static void Init();
+  // These methods are only to be used by via the embedder API.
+  CoreImpl();
+  virtual ~CoreImpl();
+  MojoHandle AddDispatcher(const scoped_refptr<Dispatcher>& dispatcher);
 
   // |CorePrivate| implementation:
   virtual MojoTimeTicks GetTimeTicksNow() OVERRIDE;
@@ -85,9 +80,6 @@ class MOJO_SYSTEM_IMPL_EXPORT CoreImpl : public Core {
                                  uint32_t num_bytes_read) OVERRIDE;
 
  private:
-  friend void embedder::Init();
-  friend class test::CoreTestBase;
-
   // The |busy| member is used only to deal with functions (in particular
   // |WriteMessage()|) that want to hold on to a dispatcher and later remove it
   // from the handle table, without holding on to the handle table lock.
@@ -123,9 +115,6 @@ class MOJO_SYSTEM_IMPL_EXPORT CoreImpl : public Core {
   };
   typedef base::hash_map<MojoHandle, HandleTableEntry> HandleTableMap;
 
-  CoreImpl();
-  virtual ~CoreImpl();
-
   // Looks up the dispatcher for the given handle. Returns null if the handle is
   // invalid.
   scoped_refptr<Dispatcher> GetDispatcher(MojoHandle handle);
diff --git a/mojo/system/core_test_base.cc b/mojo/system/core_test_base.cc
index e702321..892351a 100644
--- a/mojo/system/core_test_base.cc
+++ b/mojo/system/core_test_base.cc
@@ -176,8 +176,7 @@ void CoreTestBase::TearDown() {
 MojoHandle CoreTestBase::CreateMockHandle(CoreTestBase::MockHandleInfo* info) {
   CHECK(core_);
   scoped_refptr<MockDispatcher> dispatcher(new MockDispatcher(info));
-  base::AutoLock locker(core_->handle_table_lock_);
-  return core_->AddDispatcherNoLock(dispatcher);
+  return core_->AddDispatcher(dispatcher);
 }
 
 // CoreTestBase_MockHandleInfo -------------------------------------------------
diff --git a/mojo/system/embedder.cc b/mojo/system/embedder.cc
index b66fb2b..4136f7c 100644
--- a/mojo/system/embedder.cc
+++ b/mojo/system/embedder.cc
@@ -4,14 +4,106 @@
 
 #include "mojo/system/embedder.h"
 
+#include "base/bind.h"
+#include "base/location.h"
+#include "base/logging.h"
+#include "base/memory/scoped_ptr.h"
+#include "mojo/system/channel.h"
 #include "mojo/system/core_impl.h"
+#include "mojo/system/local_message_pipe_endpoint.h"
+#include "mojo/system/message_pipe.h"
+#include "mojo/system/message_pipe_dispatcher.h"
+#include "mojo/system/proxy_message_pipe_endpoint.h"
 
 namespace mojo {
+
+namespace embedder {
+
+struct ChannelInfo {
+  scoped_refptr<system::Channel> channel;
+};
+
+}  // namespace embedder
+
+// Have helpers in the |system| namespace, to avoid saying "system::" all over
+// the place.
+namespace system {
+namespace {
+
+void CreateChannelOnIOThread(
+    ScopedPlatformHandle platform_handle,
+    scoped_refptr<MessagePipe> message_pipe,
+    embedder::DidCreateChannelOnIOThreadCallback callback) {
+  CHECK(platform_handle.is_valid());
+
+  scoped_ptr<embedder::ChannelInfo> channel_info(new embedder::ChannelInfo);
+
+  // Create and initialize |Channel|.
+  channel_info->channel = new Channel();
+  bool success = channel_info->channel->Init(platform_handle.Pass());
+  DCHECK(success);
+
+  // Attach the message pipe endpoint.
+  MessageInTransit::EndpointId endpoint_id =
+      channel_info->channel->AttachMessagePipeEndpoint(message_pipe, 1);
+  DCHECK_EQ(endpoint_id, Channel::kBootstrapEndpointId);
+  channel_info->channel->RunMessagePipeEndpoint(Channel::kBootstrapEndpointId,
+                                                Channel::kBootstrapEndpointId);
+
+  // Hand the channel back to the embedder.
+  callback.Run(channel_info.release());
+}
+
+MojoHandle CreateChannelHelper(
+    ScopedPlatformHandle platform_handle,
+    scoped_refptr<base::TaskRunner> io_thread_task_runner,
+    embedder::DidCreateChannelOnIOThreadCallback callback) {
+  DCHECK(platform_handle.is_valid());
+
+  scoped_refptr<MessagePipe> message_pipe(new MessagePipe(
+      scoped_ptr<MessagePipeEndpoint>(new LocalMessagePipeEndpoint()),
+      scoped_ptr<MessagePipeEndpoint>(new ProxyMessagePipeEndpoint())));
+  scoped_refptr<MessagePipeDispatcher> dispatcher(new MessagePipeDispatcher());
+  dispatcher->Init(message_pipe, 0);
+
+  CoreImpl* core_impl = static_cast<CoreImpl*>(Core::Get());
+  DCHECK(core_impl);
+  MojoHandle rv = core_impl->AddDispatcher(dispatcher);
+  // TODO(vtl): Do we properly handle the failure case here?
+  if (rv != MOJO_HANDLE_INVALID) {
+    io_thread_task_runner->PostTask(FROM_HERE,
+                                    base::Bind(&CreateChannelOnIOThread,
+                                               base::Passed(&platform_handle),
+                                               message_pipe,
+                                               callback));
+  }
+  return rv;
+}
+
+}  // namespace
+}  // namespace system
+
 namespace embedder {
 
 void Init() {
   Core::Init(new system::CoreImpl());
 }
 
+MojoHandle CreateChannel(
+    system::ScopedPlatformHandle platform_handle,
+    scoped_refptr<base::TaskRunner> io_thread_task_runner,
+    DidCreateChannelOnIOThreadCallback callback) {
+  return system::CreateChannelHelper(platform_handle.Pass(),
+                                     io_thread_task_runner,
+                                     callback);
+}
+
+void DestroyChannelOnIOThread(ChannelInfo* channel_info) {
+  DCHECK(channel_info);
+  DCHECK(channel_info->channel.get());
+  channel_info->channel->Shutdown();
+  delete channel_info;
+}
+
 }  // namespace embedder
 }  // namespace mojo
diff --git a/mojo/system/embedder.h b/mojo/system/embedder.h
index 2b67e81..2431ba2 100644
--- a/mojo/system/embedder.h
+++ b/mojo/system/embedder.h
@@ -5,13 +5,51 @@
 #ifndef MOJO_SYSTEM_EMBEDDER_H_
 #define MOJO_SYSTEM_EMBEDDER_H_
 
+#include "base/callback.h"
+#include "base/memory/ref_counted.h"
+#include "base/task_runner.h"
+#include "mojo/public/system/core.h"
+#include "mojo/system/scoped_platform_handle.h"
 #include "mojo/system/system_impl_export.h"
 
 namespace mojo {
 namespace embedder {
 
+// Must be called first to initialize the (global, singleton) system.
 MOJO_SYSTEM_IMPL_EXPORT void Init();
 
+// Creates a new "channel", returning a handle to the bootstrap message pipe on
+// that channel. |platform_handle| should be an OS-dependent handle to one side
+// of a suitable bidirectional OS "pipe" (e.g., a file descriptor to a socket on
+// POSIX, a handle to a named pipe on Windows); this "pipe" should be connected
+// and ready for operation (e.g., to be written to or read from).
+// |io_thread_task_runner| should be a |TaskRunner| for the thread on which the
+// "channel" will run (read data and demultiplex).
+//
+// Returns |MOJO_HANDLE_INVALID| on error. Note that this will happen only if,
+// e.g., the handle table is full (operation of the channel begins
+// asynchronously and if, e.g., the other end of the "pipe" is closed, this will
+// report an error to the returned handle in the usual way).
+//
+// Notes: The handle returned is ready for use immediately, with messages
+// written to it queued. E.g., it would be perfectly valid for a message to be
+// immediately written to the returned handle and the handle closed, all before
+// the channel has begun operation on the IO thread. In this case, the channel
+// is expected to connect as usual, send the queued message, and report that the
+// handle was closed to the other side. (This message may well contain another
+// handle, so there may well still be message pipes "on" this channel.)
+//
+// TODO(vtl): Figure out channel teardown.
+struct ChannelInfo;
+typedef base::Callback<void(ChannelInfo*)> DidCreateChannelOnIOThreadCallback;
+MOJO_SYSTEM_IMPL_EXPORT MojoHandle CreateChannel(
+    system::ScopedPlatformHandle platform_handle,
+    scoped_refptr<base::TaskRunner> io_thread_task_runner,
+    DidCreateChannelOnIOThreadCallback callback);
+
+MOJO_SYSTEM_IMPL_EXPORT void DestroyChannelOnIOThread(
+    ChannelInfo* channel_info);
+
 }  // namespace embedder
 }  // namespace mojo
 
diff --git a/mojo/system/embedder_unittest.cc b/mojo/system/embedder_unittest.cc
new file mode 100644
index 0000000..ddd889f
--- /dev/null
+++ b/mojo/system/embedder_unittest.cc
@@ -0,0 +1,102 @@
+// 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/system/embedder.h"
+
+#include <string.h>
+
+#include "base/bind.h"
+#include "base/location.h"
+#include "base/logging.h"
+// TODO(vtl): Remove build_config.h include when fully implemented on Windows.
+#include "build/build_config.h"
+#include "mojo/public/system/core.h"
+#include "mojo/system/platform_channel_pair.h"
+#include "mojo/system/test_embedder.h"
+#include "mojo/system/test_utils.h"
+
+namespace mojo {
+namespace embedder {
+namespace {
+
+typedef system::test::TestWithIOThreadBase EmbedderTest;
+
+void StoreChannelInfo(ChannelInfo** store_channel_info_here,
+                      ChannelInfo* channel_info) {
+  CHECK(store_channel_info_here);
+  CHECK(channel_info);
+  *store_channel_info_here = channel_info;
+}
+
+TEST_F(EmbedderTest, ChannelsBasic) {
+  Init();
+
+// TODO(vtl): |PlatformChannelPair| not implemented on Windows yet.
+#if !defined(OS_WIN)
+  system::PlatformChannelPair channel_pair;
+  system::ScopedPlatformHandle server_handle = channel_pair.PassServerHandle();
+  system::ScopedPlatformHandle client_handle = channel_pair.PassClientHandle();
+
+  ChannelInfo* server_channel_info = NULL;
+  MojoHandle server_mp = CreateChannel(server_handle.Pass(),
+                                       io_thread_task_runner(),
+                                       base::Bind(&StoreChannelInfo,
+                                                  &server_channel_info));
+  EXPECT_NE(server_mp, MOJO_HANDLE_INVALID);
+
+  ChannelInfo* client_channel_info = NULL;
+  MojoHandle client_mp = CreateChannel(client_handle.Pass(),
+                                       io_thread_task_runner(),
+                                       base::Bind(&StoreChannelInfo,
+                                                  &client_channel_info));
+  EXPECT_NE(client_mp, MOJO_HANDLE_INVALID);
+
+  // We can write to a message pipe handle immediately.
+  const char kHello[] = "hello";
+  EXPECT_EQ(MOJO_RESULT_OK,
+            MojoWriteMessage(server_mp, kHello,
+                             static_cast<uint32_t>(sizeof(kHello)), NULL, 0u,
+                             MOJO_WRITE_MESSAGE_FLAG_NONE));
+
+  // Now wait for the other side to become readable.
+  EXPECT_EQ(MOJO_RESULT_OK,
+            MojoWait(client_mp, MOJO_WAIT_FLAG_READABLE,
+                     MOJO_DEADLINE_INDEFINITE));
+
+  char buffer[1000] = {};
+  uint32_t num_bytes = static_cast<uint32_t>(sizeof(kHello));
+  EXPECT_EQ(MOJO_RESULT_OK,
+            MojoReadMessage(client_mp, buffer, &num_bytes, NULL, NULL,
+                            MOJO_READ_MESSAGE_FLAG_NONE));
+  EXPECT_EQ(sizeof(kHello), num_bytes);
+  EXPECT_EQ(0, memcmp(buffer, kHello, num_bytes));
+
+  // TODO(vtl): FIXME -- This rapid-fire closing leads to a warning: "Received a
+  // message for nonexistent local destination ID 1". This is due to a race
+  // condition (in channel.cc/message_pipe.cc).
+  EXPECT_EQ(MOJO_RESULT_OK, MojoClose(server_mp));
+  EXPECT_EQ(MOJO_RESULT_OK, MojoClose(client_mp));
+
+  EXPECT_TRUE(server_channel_info != NULL);
+  system::test::PostTaskAndWait(io_thread_task_runner(),
+                                FROM_HERE,
+                                base::Bind(&DestroyChannelOnIOThread,
+                                           server_channel_info));
+
+  EXPECT_TRUE(client_channel_info != NULL);
+  system::test::PostTaskAndWait(io_thread_task_runner(),
+                                FROM_HERE,
+                                base::Bind(&DestroyChannelOnIOThread,
+                                           client_channel_info));
+#endif  // !defined(OS_WIN)
+
+  test::Shutdown();
+}
+
+// TODO(vtl): Test immediate write & close.
+// TODO(vtl): Test broken-connection cases.
+
+}  // namespace
+}  // namespace embedder
+}  // namespace mojo
diff --git a/mojo/system/raw_channel_posix_unittest.cc b/mojo/system/raw_channel_posix_unittest.cc
index 8d0adc6..2324001 100644
--- a/mojo/system/raw_channel_posix_unittest.cc
+++ b/mojo/system/raw_channel_posix_unittest.cc
@@ -15,7 +15,6 @@
 
 #include "base/basictypes.h"
 #include "base/bind.h"
-#include "base/callback.h"
 #include "base/compiler_specific.h"
 #include "base/location.h"
 #include "base/logging.h"
@@ -28,14 +27,12 @@
 #include "base/synchronization/waitable_event.h"
 #include "base/threading/platform_thread.h"  // For |Sleep()|.
 #include "base/threading/simple_thread.h"
-#include "base/threading/thread.h"
 #include "base/time/time.h"
 #include "mojo/system/message_in_transit.h"
 #include "mojo/system/platform_channel_pair.h"
 #include "mojo/system/platform_handle.h"
 #include "mojo/system/scoped_platform_handle.h"
 #include "mojo/system/test_utils.h"
-#include "testing/gtest/include/gtest/gtest.h"
 
 namespace mojo {
 namespace system {
@@ -66,17 +63,13 @@ void InitOnIOThread(RawChannel* raw_channel) {
 
 // -----------------------------------------------------------------------------
 
-class RawChannelPosixTest : public testing::Test {
+class RawChannelPosixTest : public test::TestWithIOThreadBase {
  public:
-  RawChannelPosixTest() : io_thread_("io_thread") {
-  }
-
-  virtual ~RawChannelPosixTest() {
-  }
+  RawChannelPosixTest() {}
+  virtual ~RawChannelPosixTest() {}
 
   virtual void SetUp() OVERRIDE {
-    io_thread_.StartWithOptions(
-        base::Thread::Options(base::MessageLoop::TYPE_IO, 0));
+    test::TestWithIOThreadBase::SetUp();
 
     PlatformChannelPair channel_pair;
     handles[0] = channel_pair.PassServerHandle();
@@ -87,23 +80,13 @@ class RawChannelPosixTest : public testing::Test {
     handles[0].reset();
     handles[1].reset();
 
-    io_thread_.Stop();
+    test::TestWithIOThreadBase::TearDown();
   }
 
  protected:
-  base::MessageLoop* io_thread_message_loop() {
-    return io_thread_.message_loop();
-  }
-
-  scoped_refptr<base::TaskRunner> io_thread_task_runner() {
-    return io_thread_message_loop()->message_loop_proxy();
-  }
-
   ScopedPlatformHandle handles[2];
 
  private:
-  base::Thread io_thread_;
-
   DISALLOW_COPY_AND_ASSIGN(RawChannelPosixTest);
 };
 
@@ -519,7 +502,6 @@ TEST_F(RawChannelPosixTest, OnFatalError) {
                         FROM_HERE,
                         base::Bind(&RawChannel::Shutdown,
                                    base::Unretained(rc.get())));
-
 }
 
 // RawChannelPosixTest.WriteMessageAfterShutdown -------------------------------
diff --git a/mojo/system/remote_message_pipe_posix_unittest.cc b/mojo/system/remote_message_pipe_posix_unittest.cc
index 537392f..c1d6b0c 100644
--- a/mojo/system/remote_message_pipe_posix_unittest.cc
+++ b/mojo/system/remote_message_pipe_posix_unittest.cc
@@ -10,7 +10,6 @@
 
 #include "base/basictypes.h"
 #include "base/bind.h"
-#include "base/callback.h"
 #include "base/location.h"
 #include "base/logging.h"
 #include "base/message_loop/message_loop.h"
@@ -23,24 +22,18 @@
 #include "mojo/system/scoped_platform_handle.h"
 #include "mojo/system/test_utils.h"
 #include "mojo/system/waiter.h"
-#include "testing/gtest/include/gtest/gtest.h"
 
 namespace mojo {
 namespace system {
 namespace {
 
-class RemoteMessagePipeTest : public testing::Test {
+class RemoteMessagePipeTest : public test::TestWithIOThreadBase {
  public:
-  RemoteMessagePipeTest() : io_thread_("io_thread") {
-  }
-
-  virtual ~RemoteMessagePipeTest() {
-  }
+  RemoteMessagePipeTest() {}
+  virtual ~RemoteMessagePipeTest() {}
 
   virtual void SetUp() OVERRIDE {
-    io_thread_.StartWithOptions(
-        base::Thread::Options(base::MessageLoop::TYPE_IO, 0));
-
+    test::TestWithIOThreadBase::SetUp();
     test::PostTaskAndWait(io_thread_task_runner(),
                           FROM_HERE,
                           base::Bind(&RemoteMessagePipeTest::SetUpOnIOThread,
@@ -52,7 +45,7 @@ class RemoteMessagePipeTest : public testing::Test {
                           FROM_HERE,
                           base::Bind(&RemoteMessagePipeTest::TearDownOnIOThread,
                                      base::Unretained(this)));
-    io_thread_.Stop();
+    test::TestWithIOThreadBase::TearDown();
   }
 
   // This connects MP 0, port 1 and MP 1, port 0 (leaving MP 0, port 0 and MP 1,
@@ -79,15 +72,6 @@ class RemoteMessagePipeTest : public testing::Test {
                    base::Unretained(this), channel_index, mp));
   }
 
- protected:
-  base::MessageLoop* io_thread_message_loop() {
-    return io_thread_.message_loop();
-  }
-
-  scoped_refptr<base::TaskRunner> io_thread_task_runner() {
-    return io_thread_message_loop()->message_loop_proxy();
-  }
-
  private:
   void SetUpOnIOThread() {
     CHECK_EQ(base::MessageLoop::current(), io_thread_message_loop());
@@ -151,7 +135,6 @@ class RemoteMessagePipeTest : public testing::Test {
     }
   }
 
-  base::Thread io_thread_;
   ScopedPlatformHandle platform_handles_[2];
   scoped_refptr<Channel> channels_[2];
 
diff --git a/mojo/system/test_embedder.cc b/mojo/system/test_embedder.cc
new file mode 100644
index 0000000..ca0b534
--- /dev/null
+++ b/mojo/system/test_embedder.cc
@@ -0,0 +1,26 @@
+// 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/system/test_embedder.h"
+
+#include "base/logging.h"
+#include "mojo/system/core_impl.h"
+
+namespace mojo {
+namespace embedder {
+namespace test {
+
+void Shutdown() {
+  system::CoreImpl* core_impl = static_cast<system::CoreImpl*>(Core::Get());
+  CHECK(core_impl);
+  Core::Reset();
+
+  // TODO(vtl): Check for leaks, etc.
+
+  delete core_impl;
+}
+
+}  // namespace test
+}  // namespace embedder
+}  // namespace mojo
diff --git a/mojo/system/test_embedder.h b/mojo/system/test_embedder.h
new file mode 100644
index 0000000..457165f
--- /dev/null
+++ b/mojo/system/test_embedder.h
@@ -0,0 +1,23 @@
+// 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.
+
+#ifndef MOJO_SYSTEM_TEST_EMBEDDER_H_
+#define MOJO_SYSTEM_TEST_EMBEDDER_H_
+
+namespace mojo {
+namespace embedder {
+namespace test {
+
+// This shuts down the global, singleton instance. (Note: "Real" embedders are
+// not expected to ever shut down this instance. This |Shutdown()| function will
+// do more work to ensure that tests don't leak, etc.)
+// TODO(vtl): Figure out the library/component/DLL/export situation for test
+// embedder stuff. For now, it's linked directly into the unit test binary.
+void Shutdown();
+
+}  // namespace test
+}  // namespace embedder
+}  // namespace mojo
+
+#endif  // MOJO_SYSTEM_EMBEDDER_H_
diff --git a/mojo/system/test_utils.cc b/mojo/system/test_utils.cc
index 5c120a8..673b36e 100644
--- a/mojo/system/test_utils.cc
+++ b/mojo/system/test_utils.cc
@@ -31,6 +31,23 @@ void PostTaskAndWait(scoped_refptr<base::TaskRunner> task_runner,
   event.Wait();
 }
 
+// TestWithIOThreadBase --------------------------------------------------------
+
+TestWithIOThreadBase::TestWithIOThreadBase() : io_thread_("io_thread") {
+}
+
+TestWithIOThreadBase::~TestWithIOThreadBase() {
+}
+
+void TestWithIOThreadBase::SetUp() {
+  io_thread_.StartWithOptions(
+      base::Thread::Options(base::MessageLoop::TYPE_IO, 0));
+}
+
+void TestWithIOThreadBase::TearDown() {
+  io_thread_.Stop();
+}
+
 }  // namespace test
 }  // namespace system
 }  // namespace mojo
diff --git a/mojo/system/test_utils.h b/mojo/system/test_utils.h
index d40490f..73949b9 100644
--- a/mojo/system/test_utils.h
+++ b/mojo/system/test_utils.h
@@ -7,11 +7,14 @@
 
 #include <stdint.h>
 
-#include "base/basictypes.h"
 #include "base/callback_forward.h"
+#include "base/compiler_specific.h"
+#include "base/macros.h"
 #include "base/memory/ref_counted.h"
 #include "base/task_runner.h"
+#include "base/threading/thread.h"
 #include "base/time/time.h"
+#include "testing/gtest/include/gtest/gtest.h"
 
 namespace tracked_objects {
 class Location;
@@ -47,6 +50,31 @@ void PostTaskAndWait(scoped_refptr<base::TaskRunner> task_runner,
                      const tracked_objects::Location& from_here,
                      const base::Closure& task);
 
+// TestWithIOThreadBase --------------------------------------------------------
+
+class TestWithIOThreadBase : public testing::Test {
+ public:
+  TestWithIOThreadBase();
+  virtual ~TestWithIOThreadBase();
+
+  virtual void SetUp() OVERRIDE;
+  virtual void TearDown() OVERRIDE;
+
+ protected:
+  base::MessageLoop* io_thread_message_loop() {
+    return io_thread_.message_loop();
+  }
+
+  scoped_refptr<base::TaskRunner> io_thread_task_runner() {
+    return io_thread_message_loop()->message_loop_proxy();
+  }
+
+ private:
+  base::Thread io_thread_;
+
+  DISALLOW_COPY_AND_ASSIGN(TestWithIOThreadBase);
+};
+
 }  // namespace test
 }  // namespace system
 }  // namespace mojo