Mojo: Make MessagePipe::ConvertLocalToProxy() create the ChannelEndpoint.

This will be needed to move the "paused message queue" into the
ChannelEndpoint, since transferring messages from the
LocalMessagePipeEndpoint needs to happen "atomically" with its
conversion to a ProxyMessagePipeEndpoint. (Before the MessagePipe lock
is released, the ProxyMessagePipeEndpoint needs to have a place to queue
messages, and any already-queued messages must be at the front of that
queue.)

R=darin@chromium.org

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

Cr-Commit-Position: refs/heads/master@{#295822}

mojo/system/message_pipe.cc

@@ -5,6 +5,7 @@
 #include "mojo/system/message_pipe.h"
 
 #include "base/logging.h"
+#include "mojo/system/channel_endpoint.h"
 #include "mojo/system/local_message_pipe_endpoint.h"
 #include "mojo/system/message_in_transit.h"
 #include "mojo/system/message_pipe_dispatcher.h"
@@ -145,7 +146,7 @@ void MessagePipe::RemoveWaiter(unsigned port,
   endpoints_[port]->RemoveWaiter(waiter, signals_state);
 }
 
-void MessagePipe::ConvertLocalToProxy(unsigned port) {
+scoped_refptr<ChannelEndpoint> MessagePipe::ConvertLocalToProxy(unsigned port) {
   DCHECK(port == 0 || port == 1);
 
   base::AutoLock locker(lock_);
@@ -169,6 +170,8 @@ void MessagePipe::ConvertLocalToProxy(unsigned port) {
           static_cast<LocalMessagePipeEndpoint*>(endpoints_[port].get()),
           is_peer_open));
   endpoints_[port].swap(replacement_endpoint);
+
+  return make_scoped_refptr(new ChannelEndpoint(this, port));
 }
 
 MojoResult MessagePipe::EnqueueMessage(unsigned port,

mojo/system/message_pipe.h

@@ -86,7 +86,7 @@ class MOJO_SYSTEM_IMPL_EXPORT MessagePipe
 
   // This is called by the dispatcher to convert a local endpoint to a proxy
   // endpoint.
-  void ConvertLocalToProxy(unsigned port);
+  scoped_refptr<ChannelEndpoint> ConvertLocalToProxy(unsigned port);
 
   // This is used by |Channel| to enqueue messages (typically to a
   // |LocalMessagePipeEndpoint|). Unlike |WriteMessage()|, |port| is the

mojo/system/message_pipe_dispatcher.cc

@@ -247,12 +247,10 @@ bool MessagePipeDispatcher::EndSerializeAndCloseImplNoLock(
     embedder::PlatformHandleVector* /*platform_handles*/) {
   DCHECK(HasOneRef());  // Only one ref => no need to take the lock.
 
-  // Convert the local endpoint to a proxy endpoint (moving the message queue).
-  message_pipe_->ConvertLocalToProxy(port_);
-
-  // Attach the new proxy endpoint to the channel.
-  MessageInTransit::EndpointId endpoint_id = channel->AttachEndpoint(
-      make_scoped_refptr(new ChannelEndpoint(message_pipe_.get(), port_)));
+  // Convert the local endpoint to a proxy endpoint (moving the message queue)
+  // and attach it to the channel.
+  MessageInTransit::EndpointId endpoint_id =
+      channel->AttachEndpoint(message_pipe_->ConvertLocalToProxy(port_));
   // Note: It's okay to get an endpoint ID of |kInvalidEndpointId|. (It's
   // possible that the other endpoint -- the one that we're not sending -- was
   // closed in the intervening time.) In that case, we need to deserialize a