Mojo: Implement local passing of MessagePipes.

mojo/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/system/message_pipe.h"
 
 #include "base/logging.h"
 #include "base/stl_util.h"
 #include "mojo/system/channel.h"
+#include "mojo/system/dispatcher.h"
 #include "mojo/system/local_message_pipe_endpoint.h"
 #include "mojo/system/message_in_transit.h"
 #include "mojo/system/message_pipe_endpoint.h"
 #include "mojo/system/proxy_message_pipe_endpoint.h"
 
 namespace mojo {
 namespace system {
 
 MessagePipe::MessagePipe(scoped_ptr<MessagePipeEndpoint> endpoint_0,
                          scoped_ptr<MessagePipeEndpoint> endpoint_1) {
@@ skipping 52 matching lines @@
       MessageInTransit::Create(
           MessageInTransit::kTypeMessagePipeEndpoint,
           MessageInTransit::kSubtypeMessagePipeEndpointData,
           bytes, num_bytes),
       dispatchers);
 }
 
 MojoResult MessagePipe::ReadMessage(
     unsigned port,
     void* bytes, uint32_t* num_bytes,
-    uint32_t max_num_dispatchers,
     std::vector<scoped_refptr<Dispatcher> >* dispatchers,
+    uint32_t* num_dispatchers,
     MojoReadMessageFlags flags) {
   DCHECK(port == 0 || port == 1);
 
   base::AutoLock locker(lock_);
   DCHECK(endpoints_[port].get());
 
   return endpoints_[port]->ReadMessage(bytes, num_bytes,
-                                       max_num_dispatchers, dispatchers,
+                                       dispatchers, num_dispatchers,
                                        flags);
 }
 
 MojoResult MessagePipe::AddWaiter(unsigned port,
                                   Waiter* waiter,
                                   MojoWaitFlags flags,
                                   MojoResult wake_result) {
   DCHECK(port == 0 || port == 1);
 
   base::AutoLock locker(lock_);
@@ skipping 33 matching lines @@
     message->Destroy();
     return MOJO_RESULT_FAILED_PRECONDITION;
   }
 
   MojoResult result = endpoints_[port]->CanEnqueueMessage(message, dispatchers);
   if (result != MOJO_RESULT_OK) {
     message->Destroy();
     return result;
   }
 
-  // TODO(vtl): No endpoints currently support transferring dispatchers, so we
-  // can get away with this. What we really need to do is create equivalent
-  // dispatchers here (and close the original dispatchers).
-  DCHECK(!dispatchers);
+  if (dispatchers) {
+    DCHECK(!dispatchers->empty());
 
-  endpoints_[port]->EnqueueMessage(message, NULL);
+    std::vector<scoped_refptr<Dispatcher> > replacement_dispatchers;
+    for (size_t i = 0; i < dispatchers->size(); i++) {
+      replacement_dispatchers.push_back(
+          (*dispatchers)[i]->CreateEquivalentDispatcherAndCloseNoLock());
+    }
+
+    endpoints_[port]->EnqueueMessage(message, &replacement_dispatchers);
+  } else {
+    endpoints_[port]->EnqueueMessage(message, NULL);
+  }
+
   return MOJO_RESULT_OK;
 }
 
 void MessagePipe::Attach(unsigned port,
                          scoped_refptr<Channel> channel,
                          MessageInTransit::EndpointId local_id) {
   DCHECK(port == 0 || port == 1);
   DCHECK(channel.get());
   DCHECK_NE(local_id, MessageInTransit::kInvalidEndpointId);
 
@@ skipping 51 matching lines @@
       rv = MOJO_RESULT_UNKNOWN;
       break;
   }
 
   message->Destroy();
   return rv;
 }
 
 }  // namespace system
 }  // namespace mojo