Allow data pipe producer/consumer handles to be re-sent.

mojo/edk/system/remote_producer_data_pipe_impl.cc

 // Copyright 2015 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/remote_producer_data_pipe_impl.h"
 
 #include <string.h>
 
 #include <algorithm>
 
 #include "base/logging.h"
 #include "base/memory/scoped_ptr.h"
+#include "mojo/edk/system/channel.h"
 #include "mojo/edk/system/channel_endpoint.h"
 #include "mojo/edk/system/configuration.h"
 #include "mojo/edk/system/data_pipe.h"
 #include "mojo/edk/system/message_in_transit.h"
 #include "mojo/edk/system/message_in_transit_queue.h"
 #include "mojo/edk/system/remote_consumer_data_pipe_impl.h"
 #include "mojo/edk/system/remote_data_pipe_ack.h"
 
 namespace mojo {
 namespace system {
@@ skipping 272 matching lines @@
   if (!producer_open())
     rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
   rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
   return rv;
 }
 
 void RemoteProducerDataPipeImpl::ConsumerStartSerialize(
     Channel* channel,
     size_t* max_size,
     size_t* max_platform_handles) {
-  // TODO(vtl): Support serializing consumer data pipe handles.
-  NOTIMPLEMENTED();  // FIXME
-  *max_size = 0;
+  *max_size = sizeof(SerializedDataPipeConsumerDispatcher) +
+              channel->GetSerializedEndpointSize();
   *max_platform_handles = 0;
 }
 
 bool RemoteProducerDataPipeImpl::ConsumerEndSerialize(
     Channel* channel,
     void* destination,
     size_t* actual_size,
     embedder::PlatformHandleVector* platform_handles) {
-  // TODO(vtl): Support serializing consumer data pipe handles.
-  NOTIMPLEMENTED();  // FIXME
-  owner()->ConsumerCloseNoLock();
-  return false;
+  SerializedDataPipeConsumerDispatcher* s =
+      static_cast<SerializedDataPipeConsumerDispatcher*>(destination);
+  s->validated_options = validated_options();
+  void* destination_for_endpoint = static_cast<char*>(destination) +
+                                   sizeof(SerializedDataPipeConsumerDispatcher);
+
+  MessageInTransitQueue message_queue;
+  ConvertDataToMessages(buffer_.get(), &start_index_, &current_num_bytes_,
+                        &message_queue);
+
+  if (!producer_open()) {
+    // Case 1: The producer is closed.
+    channel->SerializeEndpointWithClosedPeer(destination_for_endpoint,
+                                             &message_queue);
+    *actual_size = sizeof(SerializedDataPipeConsumerDispatcher) +
+                   channel->GetSerializedEndpointSize();
+    return true;
+  }
+
+  // Case 2: The producer isn't closed. We pass |channel_endpoint| back to the
+  // |Channel|. There's no reason for us to continue to exist afterwards.
+
+  // Note: We don't use |port|.
+  scoped_refptr<ChannelEndpoint> channel_endpoint;
+  channel_endpoint.swap(channel_endpoint_);
+  channel->SerializeEndpointWithRemotePeer(destination_for_endpoint,
+                                           &message_queue, channel_endpoint);
+  owner()->SetProducerClosedNoLock();
+
+  *actual_size = sizeof(SerializedDataPipeConsumerDispatcher) +
+                 channel->GetSerializedEndpointSize();
+  return true;
 }
 
 bool RemoteProducerDataPipeImpl::OnReadMessage(unsigned /*port*/,
                                                MessageInTransit* message) {
   // Always take ownership of the message. (This means that we should always
   // return true.)
   scoped_ptr<MessageInTransit> msg(message);
 
   if (!producer_open()) {
     DCHECK(!channel_endpoint_);
@@ skipping 108 matching lines @@
   if (!consumer_open() || !current_num_bytes_) {
     // Note: There can only be a two-phase *read* (by the consumer) if we still
     // have data.
     DCHECK(!consumer_in_two_phase_read());
     DestroyBuffer();
   }
 }
 
 }  // namespace system
 }  // namespace mojo