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

mojo/edk/system/local_data_pipe_impl.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.
 
 // TODO(vtl): I currently potentially overflow in doing index calculations.
 // E.g., |start_index_| and |current_num_bytes_| fit into a |uint32_t|, but
 // their sum may not. This is bad and poses a security risk. (We're currently
 // saved by the limit on capacity -- the maximum size of the buffer, checked in
 // |DataPipe::ValidateOptions()|, is currently sufficiently small.)
 
@@ skipping 161 matching lines @@
     embedder::PlatformHandleVector* platform_handles) {
   SerializedDataPipeProducerDispatcher* s =
       static_cast<SerializedDataPipeProducerDispatcher*>(destination);
   s->validated_options = validated_options();
   void* destination_for_endpoint = static_cast<char*>(destination) +
                                    sizeof(SerializedDataPipeProducerDispatcher);
 
   if (!consumer_open()) {
     // Case 1: The consumer is closed.
     s->consumer_num_bytes = static_cast<size_t>(-1);
-    owner()->ProducerCloseNoLock();
     *actual_size = sizeof(SerializedDataPipeProducerDispatcher);
     return true;
   }
 
   // Case 2: The consumer isn't closed. We'll replace ourselves with a
   // |RemoteProducerDataPipeImpl|.
 
+  s->consumer_num_bytes = current_num_bytes_;
   // Note: We don't use |port|.
   scoped_refptr<ChannelEndpoint> channel_endpoint =
       channel->SerializeEndpointWithLocalPeer(destination_for_endpoint, nullptr,
                                               owner(), 0);
   // Note: Keep |*this| alive until the end of this method, to make things
   // slightly easier on ourselves.
   scoped_ptr<DataPipeImpl> self(owner()->ReplaceImplNoLock(make_scoped_ptr(
       new RemoteProducerDataPipeImpl(channel_endpoint.get(), buffer_.Pass(),
                                      start_index_, current_num_bytes_))));
 
@@ skipping 149 matching lines @@
     size_t* actual_size,
     embedder::PlatformHandleVector* platform_handles) {
   SerializedDataPipeConsumerDispatcher* s =
       static_cast<SerializedDataPipeConsumerDispatcher*>(destination);
   s->validated_options = validated_options();
   void* destination_for_endpoint = static_cast<char*>(destination) +
                                    sizeof(SerializedDataPipeConsumerDispatcher);
 
   size_t old_num_bytes = current_num_bytes_;
   MessageInTransitQueue message_queue;
-  ConvertDataToMessages(&message_queue);
+  ConvertDataToMessages(buffer_.get(), &start_index_, &current_num_bytes_,
+                        &message_queue);
+  start_index_ = 0;
+  current_num_bytes_ = 0;
 
   if (!producer_open()) {
     // Case 1: The producer is closed.
     channel->SerializeEndpointWithClosedPeer(destination_for_endpoint,
                                              &message_queue);
-    owner()->ConsumerCloseNoLock();
     *actual_size = sizeof(SerializedDataPipeConsumerDispatcher) +
                    channel->GetSerializedEndpointSize();
     return true;
   }
 
   // Case 2: The producer isn't closed. We'll replace ourselves with a
   // |RemoteConsumerDataPipeImpl|.
 
   // Note: We don't use |port|.
   scoped_refptr<ChannelEndpoint> channel_endpoint =
@@ skipping 56 matching lines @@
   return current_num_bytes_;
 }
 
 void LocalDataPipeImpl::MarkDataAsConsumed(size_t num_bytes) {
   DCHECK_LE(num_bytes, current_num_bytes_);
   start_index_ += num_bytes;
   start_index_ %= capacity_num_bytes();
   current_num_bytes_ -= num_bytes;
 }
 
-void LocalDataPipeImpl::ConvertDataToMessages(
-    MessageInTransitQueue* message_queue) {
-  // The maximum amount of data to send per message (make it a multiple of the
-  // element size.
-  size_t max_message_num_bytes = GetConfiguration().max_message_num_bytes;
-  max_message_num_bytes -= max_message_num_bytes % element_num_bytes();
-  DCHECK_GT(max_message_num_bytes, 0u);
-
-  while (current_num_bytes_ > 0) {
-    size_t message_num_bytes =
-        std::min(max_message_num_bytes, GetMaxNumBytesToRead());
-
-    // Note: |message_num_bytes| fits in a |uint32_t| since the capacity does.
-    scoped_ptr<MessageInTransit> message(new MessageInTransit(
-        MessageInTransit::kTypeEndpoint, MessageInTransit::kSubtypeEndpointData,
-        static_cast<uint32_t>(message_num_bytes),
-        buffer_.get() + start_index_));
-    message_queue->AddMessage(message.Pass());
-
-    MarkDataAsConsumed(message_num_bytes);
-  }
-}
-
 }  // namespace system
 }  // namespace mojo