Mojo: Attach dispatchers to message earlier.

In particular, ProxyMessagePipeEndpoint::EnqueueMessage() gets a message with
dispatchers already attached. It will be up to the Channel to force them into a
serialized state.

Note that this is a bit less optimal (especially right now), since we means we
have to create duplicate dispatchers even for things that will be serialized
immediately. However, this structure makes things more flexible, since we can
defer serialization if we want (e.g., we don't want Channel to send everything
to RawChannel in a serialized state immediately).

R=yzshen@chromium.org

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

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@253331 0039d316-1c4b-4281-b951-d872f2087c98

6 files changed, 33 insertions, 57 deletions

diff --git a/mojo/system/local_message_pipe_endpoint.cc b/mojo/system/local_message_pipe_endpoint.cc
index 3ed73d7..6fdec6d7 100644
--- a/mojo/system/local_message_pipe_endpoint.cc
+++ b/mojo/system/local_message_pipe_endpoint.cc
@@ -49,28 +49,9 @@ void LocalMessagePipeEndpoint::OnPeerClose() {
 }
 
 void LocalMessagePipeEndpoint::EnqueueMessage(
-    scoped_ptr<MessageInTransit> message,
-    std::vector<DispatcherTransport>* transports) {
+    scoped_ptr<MessageInTransit> message) {
   DCHECK(is_open_);
   DCHECK(is_peer_open_);
-  DCHECK(!transports || !transports->empty());
-
-  // "Move" the dispatchers
-  if (transports) {
-    scoped_ptr<std::vector<scoped_refptr<Dispatcher> > >
-        dispatchers(new std::vector<scoped_refptr<Dispatcher> >());
-    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>());
-      }
-    }
-    message->SetDispatchers(dispatchers.Pass());
-  }
 
   bool was_empty = message_queue_.empty();
   message_queue_.push_back(message.release());
diff --git a/mojo/system/local_message_pipe_endpoint.h b/mojo/system/local_message_pipe_endpoint.h
index d5c7203..e9c8502 100644
--- a/mojo/system/local_message_pipe_endpoint.h
+++ b/mojo/system/local_message_pipe_endpoint.h
@@ -27,9 +27,7 @@ class MOJO_SYSTEM_IMPL_EXPORT LocalMessagePipeEndpoint
   // |MessagePipeEndpoint| implementation:
   virtual void Close() OVERRIDE;
   virtual void OnPeerClose() OVERRIDE;
-  virtual void EnqueueMessage(
-      scoped_ptr<MessageInTransit> message,
-      std::vector<DispatcherTransport>* transports) OVERRIDE;
+  virtual void EnqueueMessage(scoped_ptr<MessageInTransit> message) OVERRIDE;
 
   // There's a dispatcher for |LocalMessagePipeEndpoint|s, so we have to
   // implement/override these:
diff --git a/mojo/system/message_pipe.cc b/mojo/system/message_pipe.cc
index 76c0fae..d95df38 100644
--- a/mojo/system/message_pipe.cc
+++ b/mojo/system/message_pipe.cc
@@ -157,10 +157,27 @@ MojoResult MessagePipe::EnqueueMessage(
         }
       }
     }
+
+    // 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<std::vector<scoped_refptr<Dispatcher> > >
+        dispatchers(new std::vector<scoped_refptr<Dispatcher> >());
+    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>());
+      }
+    }
+    message->SetDispatchers(dispatchers.Pass());
   }
 
   // The endpoint's |EnqueueMessage()| may not report failure.
-  endpoints_[port]->EnqueueMessage(message.Pass(), transports);
+  endpoints_[port]->EnqueueMessage(message.Pass());
   return MOJO_RESULT_OK;
 }
 
diff --git a/mojo/system/message_pipe_endpoint.h b/mojo/system/message_pipe_endpoint.h
index d4a002d..bd9a206 100644
--- a/mojo/system/message_pipe_endpoint.h
+++ b/mojo/system/message_pipe_endpoint.h
@@ -38,12 +38,11 @@ class MOJO_SYSTEM_IMPL_EXPORT MessagePipeEndpoint {
   // All implementations must implement these.
   virtual void Close() = 0;
   virtual void OnPeerClose() = 0;
-  // Implements |MessagePipe::EnqueueMessage()| (see its description for
-  // details). A major difference is that this |EnqueueMessage()| cannot report
-  // failure (if, e.g., a channel is torn down at this point, it should silently
-  // swallow the message).
-  virtual void EnqueueMessage(scoped_ptr<MessageInTransit> message,
-                              std::vector<DispatcherTransport>* transports) = 0;
+  // Implements |MessagePipe::EnqueueMessage()|. The major differences are that:
+  //  a) Dispatchers have been vetted and cloned/attached to the message.
+  //  b) At this point, we cannot report failure (if, e.g., a channel is torn
+  //     down at this point, we should silently swallow the message).
+  virtual void EnqueueMessage(scoped_ptr<MessageInTransit> message) = 0;
 
   // Implementations must override these if they represent a local endpoint,
   // i.e., one for which there's a |MessagePipeDispatcher| (and thus a handle).
diff --git a/mojo/system/proxy_message_pipe_endpoint.cc b/mojo/system/proxy_message_pipe_endpoint.cc
index ddea878..61a2448 100644
--- a/mojo/system/proxy_message_pipe_endpoint.cc
+++ b/mojo/system/proxy_message_pipe_endpoint.cc
@@ -54,12 +54,13 @@ void ProxyMessagePipeEndpoint::OnPeerClose() {
 }
 
 void ProxyMessagePipeEndpoint::EnqueueMessage(
-    scoped_ptr<MessageInTransit> message,
-    std::vector<DispatcherTransport>* transports) {
-  DCHECK(!transports || !transports->empty());
-
-  if (transports)
-    AttachAndCloseDispatchers(message.get(), transports);
+    scoped_ptr<MessageInTransit> message) {
+  if (message->dispatchers() && !message->dispatchers()->empty()) {
+    // Since the dispatchers are attached to the message, they'll be closed on
+    // message destruction.
+    LOG(ERROR) << "Sending handles over remote message pipes not yet supported "
+                  "(sent handles will simply be closed)";
+  }
 
   EnqueueMessageInternal(message.Pass());
 }
@@ -96,19 +97,6 @@ void ProxyMessagePipeEndpoint::Run(MessageInTransit::EndpointId remote_id) {
   paused_message_queue_.clear();
 }
 
-void ProxyMessagePipeEndpoint::AttachAndCloseDispatchers(
-    MessageInTransit* message,
-    std::vector<DispatcherTransport>* transports) {
-  DCHECK(transports);
-  DCHECK(!transports->empty());
-
-  // TODO(vtl)
-  LOG(ERROR) << "Sending handles over remote message pipes not yet supported "
-                "(closing sent handles)";
-  for (size_t i = 0; i < transports->size(); i++)
-    (*transports)[i].Close();
-}
-
 // Note: We may have to enqueue messages even when our (local) peer isn't open
 // -- it may have been written to and closed immediately, before we were ready.
 // This case is handled in |Run()| (which will call us).
diff --git a/mojo/system/proxy_message_pipe_endpoint.h b/mojo/system/proxy_message_pipe_endpoint.h
index f395f03..1e1b2ea 100644
--- a/mojo/system/proxy_message_pipe_endpoint.h
+++ b/mojo/system/proxy_message_pipe_endpoint.h
@@ -47,9 +47,7 @@ class MOJO_SYSTEM_IMPL_EXPORT ProxyMessagePipeEndpoint
   // |MessagePipeEndpoint| implementation:
   virtual void Close() OVERRIDE;
   virtual void OnPeerClose() OVERRIDE;
-  virtual void EnqueueMessage(
-      scoped_ptr<MessageInTransit> message,
-      std::vector<DispatcherTransport>* transports) OVERRIDE;
+  virtual void EnqueueMessage(scoped_ptr<MessageInTransit> message) OVERRIDE;
   virtual void Attach(scoped_refptr<Channel> channel,
                       MessageInTransit::EndpointId local_id) OVERRIDE;
   virtual void Run(MessageInTransit::EndpointId remote_id) OVERRIDE;
@@ -63,11 +61,6 @@ class MOJO_SYSTEM_IMPL_EXPORT ProxyMessagePipeEndpoint
     return remote_id_ != MessageInTransit::kInvalidEndpointId;
   }
 
-  // "Attaches" |transports| (which must be non-null and nonempty) to |message|
-  // by "serializing" them in an appropriate way, and closes each dispatcher.
-  // (Note that this simply modifies |*message|, and doesn't take ownership.)
-  void AttachAndCloseDispatchers(MessageInTransit* message,
-                                 std::vector<DispatcherTransport>* transports);
   void EnqueueMessageInternal(scoped_ptr<MessageInTransit> message);
 
 #ifdef NDEBUG