Added CompoundBuffer that will be used to store data in the encoding/decoding

pipeline.

BUG=None
TEST=Unittests

Review URL: http://codereview.chromium.org/4779001

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

14 files changed, 62 insertions, 49 deletions

diff --git a/remoting/protocol/connection_to_client.cc b/remoting/protocol/connection_to_client.cc
index 417600b..c3c695d 100644
--- a/remoting/protocol/connection_to_client.cc
+++ b/remoting/protocol/connection_to_client.cc
@@ -9,7 +9,7 @@
 #include "remoting/protocol/client_control_sender.h"
 
 // TODO(hclam): Remove this header once MessageDispatcher is used.
-#include "remoting/base/multiple_array_input_stream.h"
+#include "remoting/base/compound_buffer.h"
 
 namespace remoting {
 namespace protocol {
diff --git a/remoting/protocol/host_message_dispatcher.cc b/remoting/protocol/host_message_dispatcher.cc
index d75de95..7ad9cac 100644
--- a/remoting/protocol/host_message_dispatcher.cc
+++ b/remoting/protocol/host_message_dispatcher.cc
@@ -3,7 +3,6 @@
 // found in the LICENSE file.
 
 #include "net/base/io_buffer.h"
-#include "remoting/base/multiple_array_input_stream.h"
 #include "remoting/proto/control.pb.h"
 #include "remoting/proto/event.pb.h"
 #include "remoting/protocol/host_message_dispatcher.h"
diff --git a/remoting/protocol/jingle_connection_to_host.cc b/remoting/protocol/jingle_connection_to_host.cc
index 065ae55..5b34c5d 100644
--- a/remoting/protocol/jingle_connection_to_host.cc
+++ b/remoting/protocol/jingle_connection_to_host.cc
@@ -7,8 +7,6 @@
 #include "base/callback.h"
 #include "base/message_loop.h"
 #include "remoting/base/constants.h"
-// TODO(hclam): Remove this header once MessageDispatcher is used.
-#include "remoting/base/multiple_array_input_stream.h"
 #include "remoting/jingle_glue/jingle_thread.h"
 #include "remoting/protocol/jingle_session_manager.h"
 #include "remoting/protocol/video_reader.h"
diff --git a/remoting/protocol/message_decoder.cc b/remoting/protocol/message_decoder.cc
index 1200120..2e31bbc 100644
--- a/remoting/protocol/message_decoder.cc
+++ b/remoting/protocol/message_decoder.cc
@@ -6,7 +6,7 @@
 
 #include "base/logging.h"
 #include "net/base/io_buffer.h"
-#include "remoting/base/multiple_array_input_stream.h"
+#include "remoting/base/compound_buffer.h"
 #include "remoting/proto/internal.pb.h"
 #include "third_party/libjingle/source/talk/base/byteorder.h"
 
@@ -27,7 +27,7 @@ void MessageDecoder::AddBuffer(scoped_refptr<net::IOBuffer> data,
   available_bytes_ += data_size;
 }
 
-MultipleArrayInputStream* MessageDecoder::CreateInputStreamFromData() {
+CompoundBuffer* MessageDecoder::CreateCompoundBufferFromData() {
   // Determine the payload size. If we already know it then skip this part.
   // We may not have enough data to determine the payload size so use a
   // utility function to find out.
@@ -47,17 +47,15 @@ MultipleArrayInputStream* MessageDecoder::CreateInputStreamFromData() {
   // The following loop gather buffers in |buffer_list_| that sum up to
   // |next_payload_| bytes. These buffers are added to |stream|.
 
-  // Create a MultipleArrayInputStream for parsing.
-  // TODO(hclam): Avoid creating this object everytime.
-  MultipleArrayInputStream* stream = new MultipleArrayInputStream();
+  // Create a CompoundBuffer for parsing.
+  CompoundBuffer* result = new CompoundBuffer();
   while (next_payload_ > 0 && !buffer_list_.empty()) {
     scoped_refptr<net::DrainableIOBuffer> buffer = buffer_list_.front();
     int read_bytes = std::min(buffer->BytesRemaining(), next_payload_);
 
-    // This call creates a new instance of DrainableIOBuffer internally.
     // This will reference the same base pointer but maintain it's own
     // version of data pointer.
-    stream->AddBuffer(buffer, read_bytes);
+    result->Append(buffer, read_bytes);
 
     // Adjust counters.
     buffer->DidConsume(read_bytes);
@@ -70,7 +68,8 @@ MultipleArrayInputStream* MessageDecoder::CreateInputStreamFromData() {
   }
   DCHECK_EQ(0, next_payload_);
   DCHECK_LE(0, available_bytes_);
-  return stream;
+  result->Lock();
+  return result;
 }
 
 static int GetHeaderSize(const std::string& header) {
diff --git a/remoting/protocol/message_decoder.h b/remoting/protocol/message_decoder.h
index 207dee5..ea7acee 100644
--- a/remoting/protocol/message_decoder.h
+++ b/remoting/protocol/message_decoder.h
@@ -10,8 +10,9 @@
 
 #include "base/ref_counted.h"
 #include "base/scoped_ptr.h"
-#include "google/protobuf/message_lite.h"
-#include "remoting/base/multiple_array_input_stream.h"
+#include "net/base/io_buffer.h"
+#include "remoting/base/compound_buffer.h"
+#include "third_party/protobuf/src/google/protobuf/message_lite.h"
 
 namespace net {
 class DrainableIOBuffer;
@@ -27,9 +28,9 @@ class ClientEventMessage;
 class HostControlMessage;
 class HostEventMessage;
 
-// MessageDecoder uses MultipleArrayInputStream to decode bytes into
-// protocol buffer messages. This can be used to decode bytes received from
-// the network.
+// MessageDecoder uses CompoundBuffer to decode bytes into protocol
+// buffer messages. This can be used to decode bytes received from the
+// network.
 //
 // It provides ParseMessages() which accepts an IOBuffer. If enough bytes
 // are collected to produce protocol buffer messages then the bytes will be
@@ -72,21 +73,21 @@ class MessageDecoder {
   // Parse one message from |buffer_list_|. Return true if sucessful.
   template <class MessageType>
   bool ParseOneMessage(MessageType** message) {
-    scoped_ptr<MultipleArrayInputStream> stream(CreateInputStreamFromData());
-    if (!stream.get())
+    scoped_ptr<CompoundBuffer> buffer(CreateCompoundBufferFromData());
+    if (!buffer.get())
       return false;
 
+    CompoundBufferInputStream stream(buffer.get());
     *message = new MessageType();
-    bool ret = (*message)->ParseFromZeroCopyStream(stream.get());
-    if (!ret) {
+    bool ret = (*message)->ParseFromZeroCopyStream(&stream);
+    if (!ret)
       delete *message;
-    }
     return ret;
   }
 
   void AddBuffer(scoped_refptr<net::IOBuffer> data, int data_size);
 
-  MultipleArrayInputStream* CreateInputStreamFromData();
+  CompoundBuffer* CreateCompoundBufferFromData();
 
   // Retrieves the read payload size of the current protocol buffer via |size|.
   // Returns false and leaves |size| unmodified, if we do not have enough data
diff --git a/remoting/protocol/message_decoder_unittest.cc b/remoting/protocol/message_decoder_unittest.cc
index c893d37..d237f49 100644
--- a/remoting/protocol/message_decoder_unittest.cc
+++ b/remoting/protocol/message_decoder_unittest.cc
@@ -6,8 +6,6 @@
 
 #include "base/scoped_ptr.h"
 #include "base/stl_util-inl.h"
-#include "net/base/io_buffer.h"
-#include "remoting/base/multiple_array_input_stream.h"
 #include "remoting/proto/internal.pb.h"
 #include "remoting/protocol/message_decoder.h"
 #include "remoting/protocol/util.h"
diff --git a/remoting/protocol/message_reader.cc b/remoting/protocol/message_reader.cc
index be07f86..ce0ca0f 100644
--- a/remoting/protocol/message_reader.cc
+++ b/remoting/protocol/message_reader.cc
@@ -8,7 +8,7 @@
 #include "net/base/io_buffer.h"
 #include "net/base/net_errors.h"
 #include "net/socket/socket.h"
-#include "remoting/base/multiple_array_input_stream.h"
+#include "remoting/base/compound_buffer.h"
 #include "remoting/proto/internal.pb.h"
 
 namespace remoting {
diff --git a/remoting/protocol/rtp_reader.cc b/remoting/protocol/rtp_reader.cc
index 9609a6d..b989c44 100644
--- a/remoting/protocol/rtp_reader.cc
+++ b/remoting/protocol/rtp_reader.cc
@@ -26,14 +26,13 @@ void RtpReader::Init(net::Socket* socket,
 void RtpReader::OnDataReceived(net::IOBuffer* buffer, int data_size) {
   RtpPacket packet;
   int header_size = UnpackRtpHeader(reinterpret_cast<uint8*>(buffer->data()),
-                                    data_size, &packet.header);
+                                    data_size, packet.mutable_header());
   if (header_size < 0) {
     LOG(WARNING) << "Received invalid RTP packet.";
     return;
   }
-  packet.data = buffer;
-  packet.payload = buffer->data() + header_size;
-  packet.payload_size = data_size - header_size;
+  packet.mutable_payload()->Append(buffer, buffer->data() + header_size,
+                                   data_size - header_size);
 
   on_message_callback_->Run(packet);
 }
diff --git a/remoting/protocol/rtp_reader.h b/remoting/protocol/rtp_reader.h
index c159f8d..e7b42de 100644
--- a/remoting/protocol/rtp_reader.h
+++ b/remoting/protocol/rtp_reader.h
@@ -6,20 +6,26 @@
 #define REMOTING_PROTOCOL_RTP_READER_H_
 
 #include "base/scoped_ptr.h"
+#include "remoting/base/compound_buffer.h"
 #include "remoting/protocol/rtp_utils.h"
 #include "remoting/protocol/socket_reader_base.h"
 
 namespace remoting {
 namespace protocol {
 
-struct RtpPacket {
+class RtpPacket {
+ public:
   RtpPacket();
   ~RtpPacket();
 
-  RtpHeader header;
-  scoped_refptr<net::IOBuffer> data;
-  char* payload;
-  int payload_size;
+  const RtpHeader& header() const { return header_; }
+  RtpHeader* mutable_header() { return &header_; }
+  const CompoundBuffer& payload() const { return payload_; }
+  CompoundBuffer* mutable_payload() { return &payload_; }
+
+ private:
+  RtpHeader header_;
+  CompoundBuffer payload_;
 };
 
 class RtpReader : public SocketReaderBase {
diff --git a/remoting/protocol/rtp_video_reader.cc b/remoting/protocol/rtp_video_reader.cc
index dbc2555..8a17339 100644
--- a/remoting/protocol/rtp_video_reader.cc
+++ b/remoting/protocol/rtp_video_reader.cc
@@ -27,9 +27,13 @@ void RtpVideoReader::Close() {
 void RtpVideoReader::OnRtpPacket(const RtpPacket& rtp_packet) {
   VideoPacket* packet = new VideoPacket();
 
-  packet->set_data(rtp_packet.payload, rtp_packet.payload_size);
+  packet->mutable_data()->resize(rtp_packet.payload().total_bytes());
+  rtp_packet.payload().CopyTo(
+      const_cast<char*>(packet->mutable_data()->data()),
+      packet->data().size());
+
   packet->mutable_format()->set_encoding(VideoPacketFormat::ENCODING_VP8);
-  packet->set_flags(rtp_packet.header.marker ? VideoPacket::LAST_PACKET : 0);
+  packet->set_flags(rtp_packet.header().marker ? VideoPacket::LAST_PACKET : 0);
 
   video_stub_->ProcessVideoPacket(packet, new DeleteTask<VideoPacket>(packet));
 }
diff --git a/remoting/protocol/rtp_video_writer.cc b/remoting/protocol/rtp_video_writer.cc
index 1a7ead4..1aefbe8 100644
--- a/remoting/protocol/rtp_video_writer.cc
+++ b/remoting/protocol/rtp_video_writer.cc
@@ -4,9 +4,11 @@
 
 #include "remoting/protocol/rtp_video_writer.h"
 
+#include "net/base/io_buffer.h"
+#include "remoting/base/compound_buffer.h"
 #include "remoting/proto/video.pb.h"
-#include "remoting/protocol/session.h"
 #include "remoting/protocol/rtp_writer.h"
+#include "remoting/protocol/session.h"
 
 namespace remoting {
 namespace protocol {
@@ -20,8 +22,10 @@ void RtpVideoWriter::Init(protocol::Session* session) {
 }
 
 void RtpVideoWriter::SendPacket(const VideoPacket& packet) {
-  rtp_writer_.SendPacket(packet.data().data(), packet.data().size(),
-                         packet.timestamp());
+  CompoundBuffer payload;
+  payload.AppendCopyOf(packet.data().data(), packet.data().size());
+
+  rtp_writer_.SendPacket(payload, packet.timestamp());
 }
 
 int RtpVideoWriter::GetPendingPackets() {
diff --git a/remoting/protocol/rtp_writer.cc b/remoting/protocol/rtp_writer.cc
index c27260e..e143222 100644
--- a/remoting/protocol/rtp_writer.cc
+++ b/remoting/protocol/rtp_writer.cc
@@ -6,6 +6,7 @@
 
 #include "net/base/io_buffer.h"
 #include "net/base/net_errors.h"
+#include "remoting/base/compound_buffer.h"
 #include "remoting/protocol/rtp_utils.h"
 
 namespace remoting {
@@ -33,8 +34,7 @@ void RtpWriter::Init(net::Socket* rtp_socket, net::Socket* rtcp_socket) {
   rtcp_socket_ = rtcp_socket;
 }
 
-void RtpWriter::SendPacket(const char* payload, int payload_size,
-                           uint32 timestamp) {
+void RtpWriter::SendPacket(const CompoundBuffer& payload, uint32 timestamp) {
   RtpHeader header;
   header.padding = false;
   header.extension = false;
@@ -50,15 +50,15 @@ void RtpWriter::SendPacket(const char* payload, int payload_size,
   // TODO(sergeyu): Add VP8 payload header.
 
   int position = 0;
-  while (position < payload_size) {
+  while (position < payload.total_bytes()) {
     // Allocate buffer.
-    int size = std::min(kMtu, payload_size - position);
+    int size = std::min(kMtu, payload.total_bytes() - position);
     int header_size = GetRtpHeaderSize(header.sources);
     int total_size = size + header_size;
     net::IOBufferWithSize* buffer = new net::IOBufferWithSize(total_size);
 
     // Set marker if this is the last frame.
-    header.marker = (position + size) == payload_size;
+    header.marker = (position + size) == payload.total_bytes();
 
     // TODO(sergeyu): Handle sequence number wrapping.
     header.sequence_number = last_packet_number_;
@@ -69,7 +69,9 @@ void RtpWriter::SendPacket(const char* payload, int payload_size,
                   header);
 
     // Copy data to the buffer.
-    memcpy(buffer->data() + header_size, payload + position, size);
+    CompoundBuffer chunk;
+    chunk.CopyFrom(payload, position, position + size);
+    chunk.CopyTo(buffer->data() + header_size, size);
 
     // Send it.
     buffered_rtp_writer_->Write(buffer);
@@ -77,7 +79,7 @@ void RtpWriter::SendPacket(const char* payload, int payload_size,
     position += size;
   }
 
-  DCHECK_EQ(position, payload_size);
+  DCHECK_EQ(position, payload.total_bytes());
 }
 
 int RtpWriter::GetPendingPackets() {
diff --git a/remoting/protocol/rtp_writer.h b/remoting/protocol/rtp_writer.h
index 86e92c0..1f410e5 100644
--- a/remoting/protocol/rtp_writer.h
+++ b/remoting/protocol/rtp_writer.h
@@ -9,6 +9,9 @@
 #include "remoting/protocol/buffered_socket_writer.h"
 
 namespace remoting {
+
+class CompoundBuffer;
+
 namespace protocol {
 
 class RtpWriter {
@@ -21,7 +24,7 @@ class RtpWriter {
   void Init(net::Socket* rtp_socket, net::Socket* rtcp_socket);
 
   // Sends next packet.
-  void SendPacket(const char* payload, int payload_size, uint32 timestamp);
+  void SendPacket(const CompoundBuffer& payload, uint32 timestamp);
 
   // Returns number of packets queued in the buffer.
   int GetPendingPackets();
diff --git a/remoting/protocol/util.h b/remoting/protocol/util.h
index 8c460f0..d8435b3 100644
--- a/remoting/protocol/util.h
+++ b/remoting/protocol/util.h
@@ -8,8 +8,8 @@
 #ifndef REMOTING_PROTOCOL_UTIL_H_
 #define REMOTING_PROTOCOL_UTIL_H_
 
-#include "google/protobuf/message_lite.h"
 #include "net/base/io_buffer.h"
+#include "third_party/protobuf/src/google/protobuf/message_lite.h"
 
 class Task;