Mojo: Have |ProxyMessagePipeEndpoint|s constructed with a |ChannelEndpoint|.

mojo/system/remote_message_pipe_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 <stdint.h>
 #include <stdio.h>
 #include <string.h>
 
 #include <vector>
 
@@ skipping 42 matching lines @@
   }
 
   virtual void TearDown() OVERRIDE {
     io_thread_.PostTaskAndWait(
         FROM_HERE,
         base::Bind(&RemoteMessagePipeTest::TearDownOnIOThread,
                    base::Unretained(this)));
   }
 
  protected:
-  // This connects MP 0, port 1 and MP 1, port 0 (leaving MP 0, port 0 and MP 1,
-  // port 1 as the user-visible endpoints) to channel 0 and 1, respectively. MP
-  // 0, port 1 and MP 1, port 0 must have |ProxyMessagePipeEndpoint|s.
-  void ConnectMessagePipes(scoped_refptr<MessagePipe> mp0,
-                           scoped_refptr<MessagePipe> mp1) {
+  // This connects the two given |ChannelEndpoint|s.
+  void ConnectChannelEndpoints(scoped_refptr<ChannelEndpoint> ep0,
+                               scoped_refptr<ChannelEndpoint> ep1) {
     io_thread_.PostTaskAndWait(
         FROM_HERE,
-        base::Bind(&RemoteMessagePipeTest::ConnectMessagePipesOnIOThread,
+        base::Bind(&RemoteMessagePipeTest::ConnectChannelEndpointsOnIOThread,
                    base::Unretained(this),
-                   mp0,
-                   mp1));
+                   ep0,
+                   ep1));
   }
 
-  // This connects |mp|'s port |channel_index ^ 1| to channel |channel_index|.
-  // It assumes/requires that this is the bootstrap case, i.e., that the
-  // endpoint IDs are both/will both be |Channel::kBootstrapEndpointId|. This
-  // returns *without* waiting for it to finish connecting.
-  void BootstrapMessagePipeNoWait(unsigned channel_index,
-                                  scoped_refptr<MessagePipe> mp) {
+  // This bootstraps |ep| on |channels_[channel_index]|. It assumes/requires
+  // that this is the bootstrap case, i.e., that the endpoint IDs are both/will
+  // both be |Channel::kBootstrapEndpointId|. This returns *without* waiting for
+  // it to finish connecting.
+  void BootstrapChannelEndpointNoWait(unsigned channel_index,
+                                      scoped_refptr<ChannelEndpoint> ep) {
     io_thread_.PostTask(
         FROM_HERE,
-        base::Bind(&RemoteMessagePipeTest::BootstrapMessagePipeOnIOThread,
+        base::Bind(&RemoteMessagePipeTest::BootstrapChannelEndpointOnIOThread,
                    base::Unretained(this),
                    channel_index,
-                   mp));
+                   ep));
   }
 
   void RestoreInitialState() {
     io_thread_.PostTaskAndWait(
         FROM_HERE,
         base::Bind(&RemoteMessagePipeTest::RestoreInitialStateOnIOThread,
                    base::Unretained(this)));
   }
 
   embedder::PlatformSupport* platform_support() { return &platform_support_; }
@@ skipping 24 matching lines @@
   void CreateAndInitChannel(unsigned channel_index) {
     CHECK_EQ(base::MessageLoop::current(), io_thread()->message_loop());
     CHECK(channel_index == 0 || channel_index == 1);
     CHECK(!channels_[channel_index].get());
 
     channels_[channel_index] = new Channel(&platform_support_);
     CHECK(channels_[channel_index]->Init(
         RawChannel::Create(platform_handles_[channel_index].Pass())));
   }
 
-  void ConnectMessagePipesOnIOThread(scoped_refptr<MessagePipe> mp0,
-                                     scoped_refptr<MessagePipe> mp1) {
+  void ConnectChannelEndpointsOnIOThread(scoped_refptr<ChannelEndpoint> ep0,
+                                         scoped_refptr<ChannelEndpoint> ep1) {
     CHECK_EQ(base::MessageLoop::current(), io_thread()->message_loop());
 
     if (!channels_[0].get())
       CreateAndInitChannel(0);
     if (!channels_[1].get())
       CreateAndInitChannel(1);
 
-    MessageInTransit::EndpointId local_id0 = channels_[0]->AttachEndpoint(
-        make_scoped_refptr(new ChannelEndpoint(mp0.get(), 1)));
-    MessageInTransit::EndpointId local_id1 = channels_[1]->AttachEndpoint(
-        make_scoped_refptr(new ChannelEndpoint(mp1.get(), 0)));
+    MessageInTransit::EndpointId local_id0 = channels_[0]->AttachEndpoint(ep0);
+    MessageInTransit::EndpointId local_id1 = channels_[1]->AttachEndpoint(ep1);
 
     CHECK(channels_[0]->RunMessagePipeEndpoint(local_id0, local_id1));
     CHECK(channels_[1]->RunMessagePipeEndpoint(local_id1, local_id0));
   }
 
-  void BootstrapMessagePipeOnIOThread(unsigned channel_index,
-                                      scoped_refptr<MessagePipe> mp) {
+  void BootstrapChannelEndpointOnIOThread(unsigned channel_index,
+                                          scoped_refptr<ChannelEndpoint> ep) {
     CHECK_EQ(base::MessageLoop::current(), io_thread()->message_loop());
     CHECK(channel_index == 0 || channel_index == 1);
 
-    unsigned port = channel_index ^ 1u;
-
     CreateAndInitChannel(channel_index);
     MessageInTransit::EndpointId endpoint_id =
-        channels_[channel_index]->AttachEndpoint(
-            make_scoped_refptr(new ChannelEndpoint(mp.get(), port)));
+        channels_[channel_index]->AttachEndpoint(ep);
     if (endpoint_id == MessageInTransit::kInvalidEndpointId)
       return;
 
     CHECK_EQ(endpoint_id, Channel::kBootstrapEndpointId);
     CHECK(channels_[channel_index]->RunMessagePipeEndpoint(
         Channel::kBootstrapEndpointId, Channel::kBootstrapEndpointId));
   }
 
   void RestoreInitialStateOnIOThread() {
     CHECK_EQ(base::MessageLoop::current(), io_thread()->message_loop());
@@ skipping 16 matching lines @@
   char buffer[100] = {0};
   uint32_t buffer_size = static_cast<uint32_t>(sizeof(buffer));
   Waiter waiter;
   HandleSignalsState hss;
   uint32_t context = 0;
 
   // Connect message pipes. MP 0, port 1 will be attached to channel 0 and
   // connected to MP 1, port 0, which will be attached to channel 1. This leaves
   // MP 0, port 0 and MP 1, port 1 as the "user-facing" endpoints.
 
-  scoped_refptr<MessagePipe> mp0(MessagePipe::CreateLocalProxy());
-  scoped_refptr<MessagePipe> mp1(MessagePipe::CreateProxyLocal());
-  ConnectMessagePipes(mp0, mp1);
+  scoped_refptr<ChannelEndpoint> ep0;
+  scoped_refptr<MessagePipe> mp0(MessagePipe::CreateLocalProxy(&ep0));
+  scoped_refptr<ChannelEndpoint> ep1;
+  scoped_refptr<MessagePipe> mp1(MessagePipe::CreateProxyLocal(&ep1));
+  ConnectChannelEndpoints(ep0, ep1);
 
   // Write in one direction: MP 0, port 0 -> ... -> MP 1, port 1.
 
   // Prepare to wait on MP 1, port 1. (Add the waiter now. Otherwise, if we do
   // it later, it might already be readable.)
   waiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
             mp1->AddWaiter(1, &waiter, MOJO_HANDLE_SIGNAL_READABLE, 123, NULL));
 
   // Write to MP 0, port 0.
@@ skipping 86 matching lines @@
   static const char kHello[] = "hello";
   static const char kWorld[] = "world!!!1!!!1!";
   char buffer[100] = {0};
   uint32_t buffer_size = static_cast<uint32_t>(sizeof(buffer));
   Waiter waiter;
   HandleSignalsState hss;
   uint32_t context = 0;
 
   // Connect message pipes as in the |Basic| test.
 
-  scoped_refptr<MessagePipe> mp0(MessagePipe::CreateLocalProxy());
-  scoped_refptr<MessagePipe> mp1(MessagePipe::CreateProxyLocal());
-  ConnectMessagePipes(mp0, mp1);
+  scoped_refptr<ChannelEndpoint> ep0;
+  scoped_refptr<MessagePipe> mp0(MessagePipe::CreateLocalProxy(&ep0));
+  scoped_refptr<ChannelEndpoint> ep1;
+  scoped_refptr<MessagePipe> mp1(MessagePipe::CreateProxyLocal(&ep1));
+  ConnectChannelEndpoints(ep0, ep1);
 
   // Now put another message pipe on the channel.
 
-  scoped_refptr<MessagePipe> mp2(MessagePipe::CreateLocalProxy());
-  scoped_refptr<MessagePipe> mp3(MessagePipe::CreateProxyLocal());
-  ConnectMessagePipes(mp2, mp3);
+  scoped_refptr<ChannelEndpoint> ep2;
+  scoped_refptr<MessagePipe> mp2(MessagePipe::CreateLocalProxy(&ep2));
+  scoped_refptr<ChannelEndpoint> ep3;
+  scoped_refptr<MessagePipe> mp3(MessagePipe::CreateProxyLocal(&ep3));
+  ConnectChannelEndpoints(ep2, ep3);
 
   // Write: MP 2, port 0 -> MP 3, port 1.
 
   waiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
             mp3->AddWaiter(1, &waiter, MOJO_HANDLE_SIGNAL_READABLE, 789, NULL));
 
   EXPECT_EQ(MOJO_RESULT_OK,
             mp2->WriteMessage(0,
                               UserPointer<const void>(kHello),
@@ skipping 118 matching lines @@
   char buffer[100] = {0};
   uint32_t buffer_size = static_cast<uint32_t>(sizeof(buffer));
   Waiter waiter;
   HandleSignalsState hss;
   uint32_t context = 0;
 
   // Connect message pipes. MP 0, port 1 will be attached to channel 0 and
   // connected to MP 1, port 0, which will be attached to channel 1. This leaves
   // MP 0, port 0 and MP 1, port 1 as the "user-facing" endpoints.
 
-  scoped_refptr<MessagePipe> mp0(MessagePipe::CreateLocalProxy());
+  scoped_refptr<ChannelEndpoint> ep0;
+  scoped_refptr<MessagePipe> mp0(MessagePipe::CreateLocalProxy(&ep0));
 
   // Write to MP 0, port 0.
   EXPECT_EQ(MOJO_RESULT_OK,
             mp0->WriteMessage(0,
                               UserPointer<const void>(kHello),
                               sizeof(kHello),
                               NULL,
                               MOJO_WRITE_MESSAGE_FLAG_NONE));
 
-  BootstrapMessagePipeNoWait(0, mp0);
+  BootstrapChannelEndpointNoWait(0, ep0);
 
   // Close MP 0, port 0 before channel 1 is even connected.
   mp0->Close(0);
 
-  scoped_refptr<MessagePipe> mp1(MessagePipe::CreateProxyLocal());
+  scoped_refptr<ChannelEndpoint> ep1;
+  scoped_refptr<MessagePipe> mp1(MessagePipe::CreateProxyLocal(&ep1));
 
   // Prepare to wait on MP 1, port 1. (Add the waiter now. Otherwise, if we do
   // it later, it might already be readable.)
   waiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
             mp1->AddWaiter(1, &waiter, MOJO_HANDLE_SIGNAL_READABLE, 123, NULL));
 
-  BootstrapMessagePipeNoWait(1, mp1);
+  BootstrapChannelEndpointNoWait(1, ep1);
 
   // Wait.
   EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(MOJO_DEADLINE_INDEFINITE, &context));
   EXPECT_EQ(123u, context);
   hss = HandleSignalsState();
   // Note: MP 1, port 1 should definitely should be readable, but it may or may
   // not appear as writable (there's a race, and it may not have noticed that
   // the other side was closed yet -- e.g., inserting a sleep here would make it
   // much more likely to notice that it's no longer writable).
   mp1->RemoveWaiter(1, &waiter, &hss);
@@ skipping 14 matching lines @@
   // And MP 1, port 1.
   mp1->Close(1);
 }
 
 TEST_F(RemoteMessagePipeTest, HandlePassing) {
   static const char kHello[] = "hello";
   Waiter waiter;
   HandleSignalsState hss;
   uint32_t context = 0;
 
-  scoped_refptr<MessagePipe> mp0(MessagePipe::CreateLocalProxy());
-  scoped_refptr<MessagePipe> mp1(MessagePipe::CreateProxyLocal());
-  ConnectMessagePipes(mp0, mp1);
+  scoped_refptr<ChannelEndpoint> ep0;
+  scoped_refptr<MessagePipe> mp0(MessagePipe::CreateLocalProxy(&ep0));
+  scoped_refptr<ChannelEndpoint> ep1;
+  scoped_refptr<MessagePipe> mp1(MessagePipe::CreateProxyLocal(&ep1));
+  ConnectChannelEndpoints(ep0, ep1);
 
   // We'll try to pass this dispatcher.
   scoped_refptr<MessagePipeDispatcher> dispatcher(
       new MessagePipeDispatcher(MessagePipeDispatcher::kDefaultCreateOptions));
   scoped_refptr<MessagePipe> local_mp(MessagePipe::CreateLocalLocal());
   dispatcher->Init(local_mp, 0);
 
   // Prepare to wait on MP 1, port 1. (Add the waiter now. Otherwise, if we do
   // it later, it might already be readable.)
   waiter.Init();
@@ skipping 145 matching lines @@
 #else
 // Not yet implemented (on Windows).
 #define MAYBE_SharedBufferPassing DISABLED_SharedBufferPassing
 #endif
 TEST_F(RemoteMessagePipeTest, MAYBE_SharedBufferPassing) {
   static const char kHello[] = "hello";
   Waiter waiter;
   HandleSignalsState hss;
   uint32_t context = 0;
 
-  scoped_refptr<MessagePipe> mp0(MessagePipe::CreateLocalProxy());
-  scoped_refptr<MessagePipe> mp1(MessagePipe::CreateProxyLocal());
-  ConnectMessagePipes(mp0, mp1);
+  scoped_refptr<ChannelEndpoint> ep0;
+  scoped_refptr<MessagePipe> mp0(MessagePipe::CreateLocalProxy(&ep0));
+  scoped_refptr<ChannelEndpoint> ep1;
+  scoped_refptr<MessagePipe> mp1(MessagePipe::CreateProxyLocal(&ep1));
+  ConnectChannelEndpoints(ep0, ep1);
 
   // We'll try to pass this dispatcher.
   scoped_refptr<SharedBufferDispatcher> dispatcher;
   EXPECT_EQ(MOJO_RESULT_OK,
             SharedBufferDispatcher::Create(
                 platform_support(),
                 SharedBufferDispatcher::kDefaultCreateOptions,
                 100,
                 &dispatcher));
   ASSERT_TRUE(dispatcher.get());
@@ skipping 111 matching lines @@
 TEST_F(RemoteMessagePipeTest, MAYBE_PlatformHandlePassing) {
   base::ScopedTempDir temp_dir;
   ASSERT_TRUE(temp_dir.CreateUniqueTempDir());
 
   static const char kHello[] = "hello";
   static const char kWorld[] = "world";
   Waiter waiter;
   uint32_t context = 0;
   HandleSignalsState hss;
 
-  scoped_refptr<MessagePipe> mp0(MessagePipe::CreateLocalProxy());
-  scoped_refptr<MessagePipe> mp1(MessagePipe::CreateProxyLocal());
-  ConnectMessagePipes(mp0, mp1);
+  scoped_refptr<ChannelEndpoint> ep0;
+  scoped_refptr<MessagePipe> mp0(MessagePipe::CreateLocalProxy(&ep0));
+  scoped_refptr<ChannelEndpoint> ep1;
+  scoped_refptr<MessagePipe> mp1(MessagePipe::CreateProxyLocal(&ep1));
+  ConnectChannelEndpoints(ep0, ep1);
 
   base::FilePath unused;
   base::ScopedFILE fp(
       CreateAndOpenTemporaryFileInDir(temp_dir.path(), &unused));
   EXPECT_EQ(sizeof(kHello), fwrite(kHello, 1, sizeof(kHello), fp.get()));
   // We'll try to pass this dispatcher, which will cause a |PlatformHandle| to
   // be passed.
   scoped_refptr<PlatformHandleDispatcher> dispatcher(
       new PlatformHandleDispatcher(
           mojo::test::PlatformHandleFromFILE(fp.Pass())));
@@ skipping 80 matching lines @@
 
 // Test racing closes (on each end).
 // Note: A flaky failure would almost certainly indicate a problem in the code
 // itself (not in the test). Also, any logged warnings/errors would also
 // probably be indicative of bugs.
 TEST_F(RemoteMessagePipeTest, RacingClosesStress) {
   base::TimeDelta delay = base::TimeDelta::FromMilliseconds(5);
 
   for (unsigned i = 0; i < 256; i++) {
     DVLOG(2) << "---------------------------------------- " << i;
-    scoped_refptr<MessagePipe> mp0(MessagePipe::CreateLocalProxy());
-    BootstrapMessagePipeNoWait(0, mp0);
+    scoped_refptr<ChannelEndpoint> ep0;
+    scoped_refptr<MessagePipe> mp0(MessagePipe::CreateLocalProxy(&ep0));
+    BootstrapChannelEndpointNoWait(0, ep0);
 
-    scoped_refptr<MessagePipe> mp1(MessagePipe::CreateProxyLocal());
-    BootstrapMessagePipeNoWait(1, mp1);
+    scoped_refptr<ChannelEndpoint> ep1;
+    scoped_refptr<MessagePipe> mp1(MessagePipe::CreateProxyLocal(&ep1));
+    BootstrapChannelEndpointNoWait(1, ep1);
 
     if (i & 1u) {
       io_thread()->task_runner()->PostTask(
           FROM_HERE, base::Bind(&base::PlatformThread::Sleep, delay));
     }
     if (i & 2u)
       base::PlatformThread::Sleep(delay);
 
     mp0->Close(0);
 
@@ skipping 13 matching lines @@
 // Tests passing an end of a message pipe over a remote message pipe, and then
 // passing that end back.
 // TODO(vtl): Also test passing a message pipe across two remote message pipes.
 TEST_F(RemoteMessagePipeTest, PassMessagePipeHandleAcrossAndBack) {
   static const char kHello[] = "hello";
   static const char kWorld[] = "world";
   Waiter waiter;
   HandleSignalsState hss;
   uint32_t context = 0;
 
-  scoped_refptr<MessagePipe> mp0(MessagePipe::CreateLocalProxy());
-  scoped_refptr<MessagePipe> mp1(MessagePipe::CreateProxyLocal());
-  ConnectMessagePipes(mp0, mp1);
+  scoped_refptr<ChannelEndpoint> ep0;
+  scoped_refptr<MessagePipe> mp0(MessagePipe::CreateLocalProxy(&ep0));
+  scoped_refptr<ChannelEndpoint> ep1;
+  scoped_refptr<MessagePipe> mp1(MessagePipe::CreateProxyLocal(&ep1));
+  ConnectChannelEndpoints(ep0, ep1);
 
   // We'll try to pass this dispatcher.
   scoped_refptr<MessagePipeDispatcher> dispatcher(
       new MessagePipeDispatcher(MessagePipeDispatcher::kDefaultCreateOptions));
   scoped_refptr<MessagePipe> local_mp(MessagePipe::CreateLocalLocal());
   dispatcher->Init(local_mp, 0);
 
   // Prepare to wait on MP 1, port 1. (Add the waiter now. Otherwise, if we do
   // it later, it might already be readable.)
   waiter.Init();
@@ skipping 195 matching lines @@
   mp0->Close(0);
   mp1->Close(1);
   EXPECT_EQ(MOJO_RESULT_OK, dispatcher->Close());
   // Note that |local_mp|'s port 0 belong to |dispatcher|, which was closed.
   local_mp->Close(1);
 }
 
 }  // namespace
 }  // namespace system
 }  // namespace mojo