Update mojo sdk to rev e01f9a49449381a5eb430c1fd88bf2cae73ec35a

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/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(
@@ skipping 18 matching lines @@
   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->get());  // Not technically wrong, but unlikely.
+
+  if (size != sizeof(SerializedMessagePipe)) {
+    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->get()) {
+    LOG(ERROR) << "Failed to deserialize message pipe (ID = "
+               << s->receiver_endpoint_id << ")";
+    return false;
+  }
+
+  DVLOG(2) << "Deserializing message pipe dispatcher (new local ID = "
+           << s->receiver_endpoint_id << ")";
+  *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();
 }
 
 void MessagePipe::CancelAllWaiters(unsigned port) {
   DCHECK(port == 0 || port == 1);
@@ skipping 27 matching lines @@
     unsigned port,
     UserPointer<const void> bytes,
     uint32_t num_bytes,
     std::vector<DispatcherTransport>* transports,
     MojoWriteMessageFlags flags) {
   DCHECK(port == 0 || port == 1);
   return EnqueueMessageInternal(
       GetPeerPort(port),
       make_scoped_ptr(new MessageInTransit(
           MessageInTransit::kTypeMessagePipeEndpoint,
-          MessageInTransit::kSubtypeMessagePipeEndpointData,
-          num_bytes,
-          bytes)),
+          MessageInTransit::kSubtypeMessagePipeEndpointData, num_bytes, bytes)),
       transports);
 }
 
 MojoResult MessagePipe::ReadMessage(unsigned port,
                                     UserPointer<void> bytes,
                                     UserPointer<uint32_t> num_bytes,
                                     DispatcherVector* dispatchers,
                                     uint32_t* num_dispatchers,
                                     MojoReadMessageFlags flags) {
   DCHECK(port == 0 || port == 1);
 
   base::AutoLock locker(lock_);
   DCHECK(endpoints_[port]);
 
-  return endpoints_[port]->ReadMessage(
-      bytes, num_bytes, dispatchers, num_dispatchers, flags);
+  return endpoints_[port]->ReadMessage(bytes, num_bytes, dispatchers,
+                                       num_dispatchers, flags);
 }
 
 HandleSignalsState MessagePipe::GetHandleSignalsState(unsigned port) const {
   DCHECK(port == 0 || port == 1);
 
   base::AutoLock locker(const_cast<base::Lock&>(lock_));
   DCHECK(endpoints_[port]);
 
   return endpoints_[port]->GetHandleSignalsState();
 }
@@ skipping 15 matching lines @@
                                Waiter* waiter,
                                HandleSignalsState* signals_state) {
   DCHECK(port == 0 || port == 1);
 
   base::AutoLock locker(lock_);
   DCHECK(endpoints_[port]);
 
   endpoints_[port]->RemoveWaiter(waiter, signals_state);
 }
 
+void MessagePipe::StartSerialize(unsigned /*port*/,
+                                 Channel* /*channel*/,
+                                 size_t* max_size,
+                                 size_t* max_platform_handles) {
+  *max_size = sizeof(SerializedMessagePipe);
+  *max_platform_handles = 0;
+}
+
+bool MessagePipe::EndSerialize(
+    unsigned port,
+    Channel* channel,
+    void* destination,
+    size_t* actual_size,
+    embedder::PlatformHandleVector* /*platform_handles*/) {
+  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(ConvertLocalToProxy(port), false);
+  DVLOG(2) << "Serializing message pipe (remote ID = "
+           << s->receiver_endpoint_id << ")";
+  *actual_size = sizeof(SerializedMessagePipe);
+  return true;
+}
+
 scoped_refptr<ChannelEndpoint> MessagePipe::ConvertLocalToProxy(unsigned port) {
   DCHECK(port == 0 || port == 1);
 
   base::AutoLock locker(lock_);
   DCHECK(endpoints_[port]);
   DCHECK_EQ(endpoints_[port]->GetType(), MessagePipeEndpoint::kTypeLocal);
 
   // The local peer is already closed, so just make a |ChannelEndpoint| that'll
   // send the already-queued messages.
   if (!endpoints_[GetPeerPort(port)]) {
     scoped_refptr<ChannelEndpoint> channel_endpoint(new ChannelEndpoint(
-        nullptr,
-        0,
-        static_cast<LocalMessagePipeEndpoint*>(endpoints_[port].get())
-            ->message_queue()));
+        nullptr, 0, static_cast<LocalMessagePipeEndpoint*>(
+                        endpoints_[port].get())->message_queue()));
     endpoints_[port]->Close();
     endpoints_[port].reset();
     return channel_endpoint;
   }
 
   // TODO(vtl): Allowing this case is a temporary hack. It'll set up a
   // |MessagePipe| with two proxy endpoints, which will then act as a proxy
   // (rather than trying to connect the two ends directly).
-  DLOG_IF(WARNING,
-          endpoints_[GetPeerPort(port)]->GetType() !=
-              MessagePipeEndpoint::kTypeLocal)
+  DLOG_IF(WARNING, endpoints_[GetPeerPort(port)]->GetType() !=
+                       MessagePipeEndpoint::kTypeLocal)
       << "Direct message pipe passing across multiple channels not yet "
          "implemented; will proxy";
 
   scoped_ptr<MessagePipeEndpoint> old_endpoint(endpoints_[port].Pass());
   scoped_refptr<ChannelEndpoint> channel_endpoint(new ChannelEndpoint(
-      this,
-      port,
-      static_cast<LocalMessagePipeEndpoint*>(old_endpoint.get())
-          ->message_queue()));
+      this, port, static_cast<LocalMessagePipeEndpoint*>(old_endpoint.get())
+                      ->message_queue()));
   endpoints_[port].reset(new ProxyMessagePipeEndpoint(channel_endpoint.get()));
   old_endpoint->Close();
 
   return channel_endpoint;
 }
 
 MojoResult MessagePipe::EnqueueMessage(unsigned port,
                                        scoped_ptr<MessageInTransit> message) {
   return EnqueueMessageInternal(port, message.Pass(), nullptr);
 }
@@ skipping 71 matching lines @@
   // Clone the dispatchers and attach them to the message. (This must be done as
   // a separate loop, since we want to leave the dispatchers alone on failure.)
   scoped_ptr<DispatcherVector> dispatchers(new DispatcherVector());
   dispatchers->reserve(transports->size());
   for (size_t i = 0; i < transports->size(); i++) {
     if ((*transports)[i].is_valid()) {
       dispatchers->push_back(
           (*transports)[i].CreateEquivalentDispatcherAndClose());
     } else {
       LOG(WARNING) << "Enqueueing null dispatcher";
-      dispatchers->push_back(scoped_refptr<Dispatcher>());
+      dispatchers->push_back(nullptr);
     }
   }
   message->SetDispatchers(dispatchers.Pass());
   return MOJO_RESULT_OK;
 }
 
 MojoResult MessagePipe::HandleControlMessage(
     unsigned /*port*/,
     scoped_ptr<MessageInTransit> message) {
   LOG(WARNING) << "Unrecognized MessagePipe control message subtype "
                << message->subtype();
   return MOJO_RESULT_UNKNOWN;
 }
 
 }  // namespace system
 }  // namespace mojo