Update mojo sdk to rev 59145288bae55b0fce4276b017df6a1117bcf00f

mojo/edk/system/message_pipe.cc

 // Copyright 2013 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/edk/system/message_pipe.h"
 
 #include "base/logging.h"
 #include "mojo/edk/system/channel.h"
 #include "mojo/edk/system/channel_endpoint.h"
 #include "mojo/edk/system/channel_endpoint_id.h"
 #include "mojo/edk/system/endpoint_relayer.h"
+#include "mojo/edk/system/incoming_endpoint.h"
 #include "mojo/edk/system/local_message_pipe_endpoint.h"
 #include "mojo/edk/system/message_in_transit.h"
 #include "mojo/edk/system/message_pipe_dispatcher.h"
 #include "mojo/edk/system/message_pipe_endpoint.h"
 #include "mojo/edk/system/proxy_message_pipe_endpoint.h"
 
 namespace mojo {
 namespace system {
 
-namespace {
-
-// TODO(vtl): Move this into |Channel| (and possible further).
-struct SerializedMessagePipe {
-  // This is the endpoint ID on the receiving side, and should be a "remote ID".
-  // (The receiving side should already have had an endpoint attached and been
-  // run via the |Channel|s. This endpoint will have both IDs assigned, so this
-  // ID is only needed to associate that endpoint with a particular dispatcher.)
-  ChannelEndpointId receiver_endpoint_id;
-};
-
-}  // namespace
-
 // static
 MessagePipe* MessagePipe::CreateLocalLocal() {
   MessagePipe* message_pipe = new MessagePipe();
   message_pipe->endpoints_[0].reset(new LocalMessagePipeEndpoint());
   message_pipe->endpoints_[1].reset(new LocalMessagePipeEndpoint());
   return message_pipe;
 }
 
 // static
 MessagePipe* MessagePipe::CreateLocalProxy(
     scoped_refptr<ChannelEndpoint>* channel_endpoint) {
   DCHECK(!*channel_endpoint);  // Not technically wrong, but unlikely.
   MessagePipe* message_pipe = new MessagePipe();
   message_pipe->endpoints_[0].reset(new LocalMessagePipeEndpoint());
   *channel_endpoint = new ChannelEndpoint(message_pipe, 1);
   message_pipe->endpoints_[1].reset(
       new ProxyMessagePipeEndpoint(channel_endpoint->get()));
   return message_pipe;
 }
 
 // static
+MessagePipe* MessagePipe::CreateLocalProxyFromExisting(
+    MessageInTransitQueue* message_queue,
+    ChannelEndpoint* channel_endpoint) {
+  DCHECK(message_queue);
+  MessagePipe* message_pipe = new MessagePipe();
+  message_pipe->endpoints_[0].reset(
+      new LocalMessagePipeEndpoint(message_queue));
+  if (channel_endpoint) {
+    bool attached_to_channel = channel_endpoint->ReplaceClient(message_pipe, 1);
+    message_pipe->endpoints_[1].reset(
+        new ProxyMessagePipeEndpoint(channel_endpoint));
+    if (!attached_to_channel)
+      message_pipe->OnDetachFromChannel(1);
+  } else {
+    // This means that the proxy side was already closed; we only need to inform
+    // the local side of this.
+    // TODO(vtl): This is safe to do without locking (but perhaps slightly
+    // dubious), since no other thread has access to |message_pipe| yet.
+    message_pipe->endpoints_[0]->OnPeerClose();
+  }
+  return message_pipe;
+}
+
+// static
 MessagePipe* MessagePipe::CreateProxyLocal(
     scoped_refptr<ChannelEndpoint>* channel_endpoint) {
   DCHECK(!*channel_endpoint);  // Not technically wrong, but unlikely.
   MessagePipe* message_pipe = new MessagePipe();
   *channel_endpoint = new ChannelEndpoint(message_pipe, 0);
   message_pipe->endpoints_[0].reset(
       new ProxyMessagePipeEndpoint(channel_endpoint->get()));
   message_pipe->endpoints_[1].reset(new LocalMessagePipeEndpoint());
   return message_pipe;
 }
 
 // static
 unsigned MessagePipe::GetPeerPort(unsigned port) {
   DCHECK(port == 0 || port == 1);
   return port ^ 1;
 }
 
 // static
 bool MessagePipe::Deserialize(Channel* channel,
                               const void* source,
                               size_t size,
                               scoped_refptr<MessagePipe>* message_pipe,
                               unsigned* port) {
   DCHECK(!*message_pipe);  // Not technically wrong, but unlikely.
 
-  if (size != sizeof(SerializedMessagePipe)) {
+  if (size != channel->GetSerializedEndpointSize()) {
     LOG(ERROR) << "Invalid serialized message pipe";
     return false;
   }
 
-  const SerializedMessagePipe* s =
-      static_cast<const SerializedMessagePipe*>(source);
-  *message_pipe = channel->PassIncomingMessagePipe(s->receiver_endpoint_id);
-  if (!*message_pipe) {
-    LOG(ERROR) << "Failed to deserialize message pipe (ID = "
-               << s->receiver_endpoint_id << ")";
+  scoped_refptr<IncomingEndpoint> incoming_endpoint =
+      channel->DeserializeEndpoint(source);
+  if (!incoming_endpoint)
     return false;
-  }
 
-  DVLOG(2) << "Deserializing message pipe dispatcher (new local ID = "
-           << s->receiver_endpoint_id << ")";
+  *message_pipe = incoming_endpoint->ConvertToMessagePipe();
+  DCHECK(*message_pipe);
   *port = 0;
   return true;
 }
 
 MessagePipeEndpoint::Type MessagePipe::GetType(unsigned port) {
   DCHECK(port == 0 || port == 1);
   base::AutoLock locker(lock_);
   DCHECK(endpoints_[port]);
 
   return endpoints_[port]->GetType();
@@ skipping 87 matching lines @@
                                  HandleSignalsState* signals_state) {
   DCHECK(port == 0 || port == 1);
 
   base::AutoLock locker(lock_);
   DCHECK(endpoints_[port]);
 
   endpoints_[port]->RemoveAwakable(awakable, signals_state);
 }
 
 void MessagePipe::StartSerialize(unsigned /*port*/,
-                                 Channel* /*channel*/,
+                                 Channel* channel,
                                  size_t* max_size,
                                  size_t* max_platform_handles) {
-  *max_size = sizeof(SerializedMessagePipe);
+  *max_size = channel->GetSerializedEndpointSize();
   *max_platform_handles = 0;
 }
 
 bool MessagePipe::EndSerialize(
     unsigned port,
     Channel* channel,
     void* destination,
     size_t* actual_size,
     embedder::PlatformHandleVector* /*platform_handles*/) {
   DCHECK(port == 0 || port == 1);
@@ skipping 25 matching lines @@
     //     port's message pipe dispatcher will continue to hold a reference to
     //     us.
     //
     //  3. The peer port is remote.
     //
     //     We also pass its |ChannelEndpoint| to the channel, which then decides
     //     what to do. We have no reason to continue to exist.
     //
     // TODO(vtl): Factor some of this out to |ChannelEndpoint| (or |Channel|).
 
+    // The replacement for |endpoints_[port]|, if any.
+    MessagePipeEndpoint* replacement_endpoint = nullptr;
+
     unsigned peer_port = GetPeerPort(port);
-    if (!endpoints_[peer_port]) {
-      // Case 1.
+    if (!endpoints_[peer_port]) {  // Case 1.
       channel_endpoint = new ChannelEndpoint(
           nullptr, 0, static_cast<LocalMessagePipeEndpoint*>(
                           endpoints_[port].get())->message_queue());
-      endpoints_[port]->Close();
-      endpoints_[port].reset();
     } else if (endpoints_[peer_port]->GetType() ==
-               MessagePipeEndpoint::kTypeLocal) {
-      // Case 2.
+               MessagePipeEndpoint::kTypeLocal) {  // Case 2.
       channel_endpoint = new ChannelEndpoint(
           this, port, static_cast<LocalMessagePipeEndpoint*>(
                           endpoints_[port].get())->message_queue());
-      endpoints_[port]->Close();
-      endpoints_[port].reset(
-          new ProxyMessagePipeEndpoint(channel_endpoint.get()));
-    } else {
-      // Case 3.
+      replacement_endpoint =
+          new ProxyMessagePipeEndpoint(channel_endpoint.get());
+    } else {  // Case 3.
       DLOG(WARNING) << "Direct message pipe passing across multiple channels "
                        "not yet implemented; will proxy";
 
       // Create an |EndpointRelayer| to replace ourselves (rather than having a
       // |MessagePipe| object that exists solely to relay messages between two
       // |ChannelEndpoint|s, owned by the |Channel| through them.
       //
       // This reduces overhead somewhat, and more importantly restores some
       // invariants, e.g., that |MessagePipe|s are owned by dispatchers.
       //
@@ skipping 14 matching lines @@
 
       scoped_refptr<EndpointRelayer> relayer(new EndpointRelayer());
       // We'll assign our peer port's endpoint to the relayer's port 1, and this
       // port's endpoint to the relayer's port 0.
       channel_endpoint = new ChannelEndpoint(
           relayer.get(), 0, static_cast<LocalMessagePipeEndpoint*>(
                                 endpoints_[port].get())->message_queue());
       relayer->Init(channel_endpoint.get(), peer_channel_endpoint.get());
       peer_channel_endpoint->ReplaceClient(relayer.get(), 1);
 
-      endpoints_[port]->Close();
-      endpoints_[port].reset();
       // No need to call |Close()| after |ReleaseChannelEndpoint()|.
       endpoints_[peer_port].reset();
     }
+
+    endpoints_[port]->Close();
+    endpoints_[port].reset(replacement_endpoint);
   }
 
-  SerializedMessagePipe* s = static_cast<SerializedMessagePipe*>(destination);
-
-  // Convert the local endpoint to a proxy endpoint (moving the message queue)
-  // and attach it to the channel.
-  s->receiver_endpoint_id =
-      channel->AttachAndRunEndpoint(channel_endpoint, false);
-  DVLOG(2) << "Serializing message pipe (remote ID = "
-           << s->receiver_endpoint_id << ")";
-  *actual_size = sizeof(SerializedMessagePipe);
+  // TODO(vtl): More/most of the above should be moved into (some variant of)
+  // |Channel::SerializeEndpoint()|.
+  channel->SerializeEndpoint(channel_endpoint, destination);
+  *actual_size = channel->GetSerializedEndpointSize();
   return true;
 }
 
 bool MessagePipe::OnReadMessage(unsigned port, MessageInTransit* message) {
   base::AutoLock locker(lock_);
 
   if (!endpoints_[port]) {
     // This will happen only on the rare occasion that the call to
     // |OnReadMessage()| is racing with us calling
     // |ChannelEndpoint::ReplaceClient()|, in which case we reject the message,
@@ skipping 92 matching lines @@
       LOG(WARNING) << "Enqueueing null dispatcher";
       dispatchers->push_back(nullptr);
     }
   }
   message->SetDispatchers(dispatchers.Pass());
   return MOJO_RESULT_OK;
 }
 
 }  // namespace system
 }  // namespace mojo