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

21 files changed, 666 insertions, 294 deletions

diff --git a/remoting/base/compound_buffer.cc b/remoting/base/compound_buffer.cc
new file mode 100644
index 0000000..9e8b8c3
--- /dev/null
+++ b/remoting/base/compound_buffer.cc
@@ -0,0 +1,232 @@
+// Copyright (c) 2010 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 <functional>
+
+#include "base/logging.h"
+#include "net/base/io_buffer.h"
+#include "remoting/base/compound_buffer.h"
+
+namespace remoting {
+
+CompoundBuffer::DataChunk::DataChunk(
+    net::IOBuffer* buffer_value, const char* start_value, int size_value)
+    : buffer(buffer_value),
+      start(start_value),
+      size(size_value) {
+}
+
+CompoundBuffer::CompoundBuffer()
+    : total_bytes_(0),
+      locked_(false) {
+}
+
+CompoundBuffer::~CompoundBuffer() {
+}
+
+void CompoundBuffer::Clear() {
+  CHECK(!locked_);
+  chunks_.clear();
+  total_bytes_ = 0;
+}
+
+void CompoundBuffer::Append(net::IOBuffer* buffer,
+                            const char* start, int size) {
+  // A weak check that the |start| is within |buffer|.
+  DCHECK_GE(start, buffer->data());
+  DCHECK_GT(size, 0);
+
+  CHECK(!locked_);
+
+  chunks_.push_back(DataChunk(buffer, start, size));
+  total_bytes_ += size;
+}
+
+void CompoundBuffer::Append(net::IOBuffer* buffer, int size) {
+  Append(buffer, buffer->data(), size);
+}
+
+void CompoundBuffer::Append(const CompoundBuffer& buffer) {
+  for (DataChunkList::const_iterator it = buffer.chunks_.begin();
+       it != buffer.chunks_.end(); ++it) {
+    Append(it->buffer, it->start, it->size);
+  }
+}
+
+void CompoundBuffer::Prepend(net::IOBuffer* buffer,
+                             const char* start, int size) {
+  // A weak check that the |start| is within |buffer|.
+  DCHECK_GE(start, buffer->data());
+  DCHECK_GT(size, 0);
+
+  CHECK(!locked_);
+
+  chunks_.push_front(DataChunk(buffer, start, size));
+  total_bytes_ += size;
+}
+
+void CompoundBuffer::Prepend(net::IOBuffer* buffer, int size) {
+  Prepend(buffer, buffer->data(), size);
+}
+
+void CompoundBuffer::Prepend(const CompoundBuffer& buffer) {
+  for (DataChunkList::const_iterator it = buffer.chunks_.begin();
+       it != buffer.chunks_.end(); ++it) {
+    Prepend(it->buffer, it->start, it->size);
+  }
+}
+void CompoundBuffer::AppendCopyOf(const char* data, int size) {
+  net::IOBuffer* buffer = new net::IOBuffer(size);
+  memcpy(buffer->data(), data, size);
+  Append(buffer, size);
+}
+
+void CompoundBuffer::PrependCopyOf(const char* data, int size) {
+  net::IOBuffer* buffer = new net::IOBuffer(size);
+  memcpy(buffer->data(), data, size);
+  Prepend(buffer, size);
+}
+
+void CompoundBuffer::Lock() {
+  locked_ = true;
+}
+
+net::IOBufferWithSize* CompoundBuffer::ToIOBufferWithSize() const {
+  net::IOBufferWithSize* result = new net::IOBufferWithSize(total_bytes_);
+  CopyTo(result->data(), total_bytes_);
+  return result;
+}
+
+void CompoundBuffer::CopyTo(char* data, int size) const {
+  char* pos = data;
+  for (DataChunkList::const_iterator it = chunks_.begin();
+       it != chunks_.end(); ++it) {
+    CHECK_LE(pos + it->size, data + size);
+    memcpy(pos, it->start, it->size);
+    pos += it->size;
+  }
+}
+
+void CompoundBuffer::CopyFrom(const CompoundBuffer& source,
+                              int start, int end) {
+  // Check that 0 <= |start| <= |end| <= |total_bytes_|.
+  DCHECK_LE(0, start);
+  DCHECK_LE(start, end);
+  DCHECK_LE(end, source.total_bytes());
+
+  Clear();
+
+  if (end == start) {
+    return;
+  }
+
+  // Iterate over chunks in the |source| and add those that we need.
+  int pos = 0;
+  for (DataChunkList::const_iterator it = source.chunks_.begin();
+       it != source.chunks_.end(); ++it) {
+
+    // Add data from the current chunk only if it is in the specified interval.
+    if (pos + it->size > start && pos < end) {
+      int relative_start = std::max(0, start - pos);
+      int relative_end = std::min(it->size, end - pos);
+      DCHECK_LE(0, relative_start);
+      DCHECK_LT(relative_start, relative_end);
+      DCHECK_LE(relative_end, it->size);
+      Append(it->buffer.get(), it->start + relative_start,
+             relative_end - relative_start);
+    }
+
+    pos += it->size;
+    if (pos >= end) {
+      // We've got all the data we need.
+      break;
+    }
+  }
+
+  DCHECK_EQ(total_bytes_, end - start);
+}
+
+CompoundBufferInputStream::CompoundBufferInputStream(
+    const CompoundBuffer* buffer)
+    : buffer_(buffer),
+      current_chunk_(0),
+      current_chunk_position_(0),
+      position_(0),
+      last_returned_size_(0) {
+  DCHECK(buffer_->locked());
+}
+
+CompoundBufferInputStream::~CompoundBufferInputStream() {
+}
+
+bool CompoundBufferInputStream::Next(const void** data, int* size) {
+  if (current_chunk_ < buffer_->chunks_.size()) {
+    // Reply with the number of bytes remaining in the current buffer.
+    const CompoundBuffer::DataChunk& chunk = buffer_->chunks_[current_chunk_];
+    int read_size = chunk.size - current_chunk_position_;
+    *data = chunk.start + current_chunk_position_;
+    *size = read_size;
+
+    // Adjust position.
+    ++current_chunk_;
+    current_chunk_position_ = 0;
+    position_ += read_size;
+
+    last_returned_size_ = read_size;
+    return true;
+  }
+
+  DCHECK_EQ(position_, buffer_->total_bytes());
+
+  // We've reached the end of the stream. So reset |last_returned_size_|
+  // to zero to prevent any backup request.
+  // This is the same as in ArrayInputStream.
+  // See google/protobuf/io/zero_copy_stream_impl_lite.cc.
+  last_returned_size_ = 0;
+  return false;
+}
+
+void CompoundBufferInputStream::BackUp(int count) {
+  DCHECK_LE(count, last_returned_size_);
+  DCHECK_GT(current_chunk_, 0u);
+
+  // Rewind one buffer and rewind data offset by |count| bytes.
+  --current_chunk_;
+  const CompoundBuffer::DataChunk& chunk = buffer_->chunks_[current_chunk_];
+  current_chunk_position_ = chunk.size - count;
+  position_ -= count;
+  DCHECK_GE(position_, 0);
+
+  // Prevent additional backups.
+  last_returned_size_ = 0;
+}
+
+bool CompoundBufferInputStream::Skip(int count) {
+  DCHECK_GE(count, 0);
+  last_returned_size_ = 0;
+
+  while (count > 0 && current_chunk_ < buffer_->chunks_.size()) {
+    const CompoundBuffer::DataChunk& chunk = buffer_->chunks_[current_chunk_];
+    int read = std::min(count, chunk.size - current_chunk_position_);
+
+    // Advance the current buffer offset and position.
+    current_chunk_position_ += read;
+    position_ += read;
+    count -= read;
+
+    // If the current buffer is fully read, then advance to the next buffer.
+    if (current_chunk_position_ == chunk.size) {
+      ++current_chunk_;
+      current_chunk_position_ = 0;
+    }
+  }
+
+  return count == 0;
+}
+
+int64 CompoundBufferInputStream::ByteCount() const {
+  return position_;
+}
+
+}  // namespace remoting
diff --git a/remoting/base/compound_buffer.h b/remoting/base/compound_buffer.h
new file mode 100644
index 0000000..050182a
--- /dev/null
+++ b/remoting/base/compound_buffer.h
@@ -0,0 +1,120 @@
+// Copyright (c) 2010 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.
+
+// CompoundBuffer implements a data buffer that is composed of several pieces,
+// each stored in a refcounted IOBuffer. It is needed for encoding/decoding
+// video pipeline to represent data packet and minimize data copying.
+// It is particularly useful for splitting data between multiple RTP packets
+// and assembling them into one buffer on the receiving side.
+//
+// CompoundBufferInputStream implements ZeroCopyInputStream interface
+// to be used by protobuf to decode data stored in CompoundBuffer into
+// a protocol buffer message.
+//
+// Mutations to the buffer are not thread-safe. Immutability can be ensured
+// with the Lock() method.
+
+#ifndef REMOTING_BASE_COMPOUND_BUFFER_H_
+#define REMOTING_BASE_COMPOUND_BUFFER_H_
+
+#include <deque>
+
+#include "base/basictypes.h"
+#include "base/ref_counted.h"
+#include "google/protobuf/io/zero_copy_stream.h"
+
+namespace net {
+class IOBuffer;
+class IOBufferWithSize;
+}  // namespace net
+
+namespace remoting {
+
+class CompoundBuffer {
+ public:
+  CompoundBuffer();
+  ~CompoundBuffer();
+
+  void Clear();
+
+  // Adds new chunk to the buffer. |start| defines position of the chunk
+  // within the |buffer|. |size| is the size of the chunk that is being
+  // added, not size of the |buffer|.
+  void Append(net::IOBuffer* buffer, int size);
+  void Append(net::IOBuffer* buffer, const char* start, int size);
+  void Append(const CompoundBuffer& buffer);
+  void Prepend(net::IOBuffer* buffer, int size);
+  void Prepend(net::IOBuffer* buffer, const char* start, int size);
+  void Prepend(const CompoundBuffer& buffer);
+
+  // Same as above, but creates new IOBuffer and copies the data.
+  void AppendCopyOf(const char* data, int data_size);
+  void PrependCopyOf(const char* data, int data_size);
+
+  // Current size of the buffer.
+  int total_bytes() const { return total_bytes_; }
+
+  // Locks the buffer. After the buffer is locked, no data can be
+  // added or removed (content can still be changed if some other
+  // object holds reference to the IOBuffer objects).
+  void Lock();
+
+  // Returns true if content is locked.
+  bool locked() const { return locked_; }
+
+  // Creates new IOBufferWithSize object and copies all data into it.
+  // Ownership of the result is given to the caller.
+  net::IOBufferWithSize* ToIOBufferWithSize() const;
+
+  // Copies all data into given location.
+  void CopyTo(char* data, int data_size) const;
+
+  // Clears the buffer, and initializes it with the interval from |buffer|
+  // starting at |start| and ending at |end|. The data itself isn't copied.
+  void CopyFrom(const CompoundBuffer& source, int start, int end);
+
+ private:
+  friend class CompoundBufferInputStream;
+
+  struct DataChunk {
+    DataChunk(net::IOBuffer* buffer, const char* start, int size);
+
+    scoped_refptr<net::IOBuffer> buffer;
+    const char* start;
+    int size;
+  };
+  typedef std::deque<DataChunk> DataChunkList;
+
+  DataChunkList chunks_;
+  int total_bytes_;
+  bool locked_;
+
+  DISALLOW_COPY_AND_ASSIGN(CompoundBuffer);
+};
+
+class CompoundBufferInputStream
+    : public google::protobuf::io::ZeroCopyInputStream  {
+ public:
+  // Caller keeps ownership of |buffer|. |buffer| must be locked.
+  explicit CompoundBufferInputStream(const CompoundBuffer* buffer);
+  virtual ~CompoundBufferInputStream();
+
+  // google::protobuf::io::ZeroCopyInputStream interface.
+  virtual bool Next(const void** data, int* size);
+  virtual void BackUp(int count);
+  virtual bool Skip(int count);
+  virtual int64 ByteCount() const;
+
+ private:
+  const CompoundBuffer* buffer_;
+
+  size_t current_chunk_;
+  int current_chunk_position_;
+  int position_;
+  int last_returned_size_;
+};
+
+}  // namespace remoting
+
+#endif  // REMOTING_BASE_COMPOUND_BUFFER_H_
diff --git a/remoting/base/compound_buffer_unittest.cc b/remoting/base/compound_buffer_unittest.cc
new file mode 100644
index 0000000..63ede5a
--- /dev/null
+++ b/remoting/base/compound_buffer_unittest.cc
@@ -0,0 +1,249 @@
+// Copyright (c) 2010 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 <string>
+
+#include "base/callback.h"
+#include "base/scoped_ptr.h"
+#include "net/base/io_buffer.h"
+#include "remoting/base/compound_buffer.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+using net::IOBuffer;
+
+namespace remoting {
+
+namespace {
+const int kDataSize = 1024;
+
+// Chunk sizes used to append and prepend data to the buffer.
+const int kChunkSizes0[] = {kDataSize, -1};
+const int kChunkSizes1[] = {1, 10, 20, -1};
+
+// Chunk sizes used to test CopyFrom().
+const int kCopySizes0[] = {10, 3, -1};
+const int kCopySizes1[] = {20, -1};
+}  // namespace
+
+class CompoundBufferTest : public testing::Test {
+ public:
+
+  // Following 5 methods are used with IterateOverPieces().
+  void Append(int pos, int size) {
+    target_.Append(data_, data_->data() + pos, size);
+  }
+
+  void AppendCopyOf(int pos, int size) {
+    target_.AppendCopyOf(data_->data() + pos, size);
+  }
+
+  void Prepend(int pos, int size) {
+    target_.Prepend(data_, data_->data() + kDataSize - pos - size, size);
+  }
+
+  void PrependCopyOf(int pos, int size) {
+    target_.PrependCopyOf(data_->data() + (kDataSize - pos - size), size);
+  }
+
+  void TestCopyFrom(int pos, int size) {
+    CompoundBuffer copy;
+    copy.CopyFrom(target_, pos, pos + size);
+    EXPECT_TRUE(CompareData(copy, data_->data() + pos, size));
+  }
+
+ protected:
+  virtual void SetUp() {
+    data_ = new IOBuffer(kDataSize);
+    for (int i = 0; i < kDataSize; ++i) {
+      data_->data()[i] = i;
+    }
+  }
+
+  // Iterate over chunks of data with sizes specified in |sizes| in the
+  // interval [0..kDataSize]. |function| is called for each chunk.
+  void IterateOverPieces(const int sizes[],
+                         Callback2<int, int>::Type* function) {
+    DCHECK_GT(sizes[0], 0);
+
+    int pos = 0;
+    int index = 0;
+    while (pos < kDataSize) {
+      int size = std::min(sizes[index], kDataSize - pos);
+      ++index;
+      if (sizes[index] <= 0)
+        index = 0;
+
+      function->Run(pos, size);
+
+      pos += size;
+    }
+    delete function;
+  }
+
+  bool CompareData(const CompoundBuffer& buffer, char* data, int size) {
+    scoped_refptr<IOBuffer> buffer_data = buffer.ToIOBufferWithSize();
+    return buffer.total_bytes() == size &&
+        memcmp(buffer_data->data(), data, size) == 0;
+  }
+
+  static size_t ReadFromInput(CompoundBufferInputStream* input,
+                              void* data, size_t size) {
+    uint8* out = reinterpret_cast<uint8*>(data);
+    int out_size = size;
+
+    const void* in;
+    int in_size = 0;
+
+    while (true) {
+      if (!input->Next(&in, &in_size)) {
+        return size - out_size;
+      }
+      EXPECT_GT(in_size, -1);
+
+      if (out_size <= in_size) {
+        memcpy(out, in, out_size);
+        if (in_size > out_size) {
+          input->BackUp(in_size - out_size);
+        }
+        return size;  // Copied all of it.
+      }
+
+      memcpy(out, in, in_size);
+      out += in_size;
+      out_size -= in_size;
+    }
+  }
+
+  static void ReadString(CompoundBufferInputStream* input,
+                         const std::string& str) {
+    SCOPED_TRACE(str);
+    scoped_array<char> buffer(new char[str.size() + 1]);
+    buffer[str.size()] = '\0';
+    EXPECT_EQ(ReadFromInput(input, buffer.get(), str.size()), str.size());
+    EXPECT_STREQ(str.data(), buffer.get());
+  }
+
+  // Construct and prepare data in the |buffer|.
+  static void PrepareData(scoped_ptr<CompoundBuffer>* buffer) {
+    static const std::string kTestData =
+        "Hello world!"
+        "This is testing"
+        "MultipleArrayInputStream"
+        "for Chromoting";
+
+    // Determine how many segments to split kTestData. We split the data in
+    // 1 character, 2 characters, 1 character, 2 characters ...
+    int segments = (kTestData.length() / 3) * 2;
+    int remaining_chars = kTestData.length() % 3;
+    if (remaining_chars) {
+      if (remaining_chars == 1)
+        ++segments;
+      else
+        segments += 2;
+    }
+
+    CompoundBuffer* result = new CompoundBuffer();
+    const char* data = kTestData.data();
+    for (int i = 0; i < segments; ++i) {
+      int size = i % 2 == 0 ? 1 : 2;
+      result->Append(new net::WrappedIOBuffer(data), size);
+      data += size;
+    }
+    result->Lock();
+    buffer->reset(result);
+  }
+
+  CompoundBuffer target_;
+  scoped_refptr<IOBuffer> data_;
+};
+
+TEST_F(CompoundBufferTest, Append) {
+  target_.Clear();
+  IterateOverPieces(kChunkSizes0, NewCallback(
+      static_cast<CompoundBufferTest*>(this), &CompoundBufferTest::Append));
+  EXPECT_TRUE(CompareData(target_, data_->data(), kDataSize));
+
+  target_.Clear();
+  IterateOverPieces(kChunkSizes1, NewCallback(
+      static_cast<CompoundBufferTest*>(this), &CompoundBufferTest::Append));
+  EXPECT_TRUE(CompareData(target_, data_->data(), kDataSize));
+}
+
+TEST_F(CompoundBufferTest, AppendCopyOf) {
+  target_.Clear();
+  IterateOverPieces(kChunkSizes0, NewCallback(
+      static_cast<CompoundBufferTest*>(this),
+      &CompoundBufferTest::AppendCopyOf));
+  EXPECT_TRUE(CompareData(target_, data_->data(), kDataSize));
+
+  target_.Clear();
+  IterateOverPieces(kChunkSizes1, NewCallback(
+      static_cast<CompoundBufferTest*>(this),
+      &CompoundBufferTest::AppendCopyOf));
+  EXPECT_TRUE(CompareData(target_, data_->data(), kDataSize));
+}
+
+TEST_F(CompoundBufferTest, Prepend) {
+  target_.Clear();
+  IterateOverPieces(kChunkSizes0, NewCallback(
+      static_cast<CompoundBufferTest*>(this), &CompoundBufferTest::Prepend));
+  EXPECT_TRUE(CompareData(target_, data_->data(), kDataSize));
+
+  target_.Clear();
+  IterateOverPieces(kChunkSizes1, NewCallback(
+      static_cast<CompoundBufferTest*>(this), &CompoundBufferTest::Prepend));
+  EXPECT_TRUE(CompareData(target_, data_->data(), kDataSize));
+}
+
+TEST_F(CompoundBufferTest, PrependCopyOf) {
+  target_.Clear();
+  IterateOverPieces(kChunkSizes0, NewCallback(
+      static_cast<CompoundBufferTest*>(this),
+      &CompoundBufferTest::PrependCopyOf));
+  EXPECT_TRUE(CompareData(target_, data_->data(), kDataSize));
+
+  target_.Clear();
+  IterateOverPieces(kChunkSizes1, NewCallback(
+      static_cast<CompoundBufferTest*>(this),
+      &CompoundBufferTest::PrependCopyOf));
+  EXPECT_TRUE(CompareData(target_, data_->data(), kDataSize));
+}
+
+TEST_F(CompoundBufferTest, CopyFrom) {
+  target_.Clear();
+  IterateOverPieces(kChunkSizes1, NewCallback(
+      static_cast<CompoundBufferTest*>(this), &CompoundBufferTest::Append));
+  {
+    SCOPED_TRACE("CopyFrom.kCopySizes0");
+    IterateOverPieces(kCopySizes0, NewCallback(
+        static_cast<CompoundBufferTest*>(this),
+        &CompoundBufferTest::TestCopyFrom));
+  }
+  {
+    SCOPED_TRACE("CopyFrom.kCopySizes1");
+    IterateOverPieces(kCopySizes1, NewCallback(
+        static_cast<CompoundBufferTest*>(this),
+        &CompoundBufferTest::TestCopyFrom));
+  }
+}
+
+TEST_F(CompoundBufferTest, InputStream) {
+  scoped_ptr<CompoundBuffer> buffer;
+  PrepareData(&buffer);
+  CompoundBufferInputStream stream(buffer.get());
+
+  ReadString(&stream, "Hello world!");
+  ReadString(&stream, "This ");
+  ReadString(&stream, "is test");
+  EXPECT_TRUE(stream.Skip(3));
+  ReadString(&stream, "MultipleArrayInput");
+  EXPECT_TRUE(stream.Skip(6));
+  ReadString(&stream, "f");
+  ReadString(&stream, "o");
+  ReadString(&stream, "r");
+  ReadString(&stream, " ");
+  ReadString(&stream, "Chromoting");
+}
+
+}  // namespace remoting
diff --git a/remoting/base/multiple_array_input_stream.cc b/remoting/base/multiple_array_input_stream.cc
deleted file mode 100644
index 67aa2ff..0000000
--- a/remoting/base/multiple_array_input_stream.cc
+++ /dev/null
@@ -1,87 +0,0 @@
-// Copyright (c) 2010 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 <functional>
-
-#include "base/logging.h"
-#include "net/base/io_buffer.h"
-#include "remoting/base/multiple_array_input_stream.h"
-
-namespace remoting {
-
-MultipleArrayInputStream::MultipleArrayInputStream()
-    : current_buffer_(0),
-      position_(0),
-      last_returned_size_(0) {
-}
-
-MultipleArrayInputStream::~MultipleArrayInputStream() {
-}
-
-void MultipleArrayInputStream::AddBuffer(net::IOBuffer* buffer, int size) {
-  DCHECK_EQ(position_, 0); // Haven't started reading.
-  buffers_.push_back(make_scoped_refptr(
-      new net::DrainableIOBuffer(buffer, size)));
-}
-
-bool MultipleArrayInputStream::Next(const void** data, int* size) {
-  if (current_buffer_ < buffers_.size()) {
-    // Reply with the number of bytes remaining in the current buffer.
-    scoped_refptr<net::DrainableIOBuffer> buffer = buffers_[current_buffer_];
-    last_returned_size_ = buffer->BytesRemaining();
-    *data = buffer->data();
-    *size = last_returned_size_;
-
-    // After reading the current buffer then advance to the next buffer.
-    buffer->DidConsume(last_returned_size_);
-    ++current_buffer_;
-    position_ += last_returned_size_;
-    return true;
-  }
-
-  // We've reached the end of the stream. So reset |last_returned_size_|
-  // to zero to prevent any backup request.
-  // This is the same as in ArrayInputStream.
-  // See google/protobuf/io/zero_copy_stream_impl_lite.cc.
-  last_returned_size_ = 0;
-  return false;
-}
-
-void MultipleArrayInputStream::BackUp(int count) {
-  DCHECK_LE(count, last_returned_size_);
-  DCHECK_GT(current_buffer_, 0u);
-
-  // Rewind one buffer and rewind data offset by |count| bytes.
-  --current_buffer_;
-  scoped_refptr<net::DrainableIOBuffer> buffer = buffers_[current_buffer_];
-  buffer->SetOffset(buffer->size() - count);
-  position_ -= count;
-  DCHECK_GE(position_, 0);
-}
-
-bool MultipleArrayInputStream::Skip(int count) {
-  DCHECK_GE(count, 0);
-  last_returned_size_ = 0;
-
-  while (count && current_buffer_ < buffers_.size()) {
-    scoped_refptr<net::DrainableIOBuffer> buffer = buffers_[current_buffer_];
-    int read = std::min(count, buffer->BytesRemaining());
-
-    // Advance the current buffer offset and position.
-    buffer->DidConsume(read);
-    position_ += read;
-    count -= read;
-
-    // If the current buffer is fully read, then advance to the next buffer.
-    if (!buffer->BytesRemaining())
-      ++current_buffer_;
-  }
-  return count == 0;
-}
-
-int64 MultipleArrayInputStream::ByteCount() const {
-  return position_;
-}
-
-}  // namespace remoting
diff --git a/remoting/base/multiple_array_input_stream.h b/remoting/base/multiple_array_input_stream.h
deleted file mode 100644
index a848248..0000000
--- a/remoting/base/multiple_array_input_stream.h
+++ /dev/null
@@ -1,57 +0,0 @@
-// Copyright (c) 2010 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.
-
-// MultipleArrayInputStream implements ZeroCopyInputStream to be used by
-// protobuf to decode bytes into a protocol buffer message.
-//
-// This input stream is made of multiple IOBuffers received from the network.
-// This object retains the IOBuffers added to it.
-//
-// Internally, we wrap each added IOBuffer in a DrainableIOBuffer. This allows
-// us to track how much data has been consumed from each IOBuffer.
-
-#ifndef REMOTING_BASE_MULTIPLE_ARRAY_INPUT_STREAM_H_
-#define REMOTING_BASE_MULTIPLE_ARRAY_INPUT_STREAM_H_
-
-#include <vector>
-
-#include "base/basictypes.h"
-#include "base/ref_counted.h"
-#include "google/protobuf/io/zero_copy_stream.h"
-
-namespace net {
-class DrainableIOBuffer;
-class IOBuffer;
-}  // namespace net
-
-namespace remoting {
-
-class MultipleArrayInputStream :
-      public google::protobuf::io::ZeroCopyInputStream {
- public:
-  MultipleArrayInputStream();
-  virtual ~MultipleArrayInputStream();
-
-  // Add a buffer to the list. |buffer| is retained by this object.
-  void AddBuffer(net::IOBuffer* buffer, int size);
-
-  // google::protobuf::io::ZeroCopyInputStream interface.
-  virtual bool Next(const void** data, int* size);
-  virtual void BackUp(int count);
-  virtual bool Skip(int count);
-  virtual int64 ByteCount() const;
-
- private:
-  std::vector<scoped_refptr<net::DrainableIOBuffer> > buffers_;
-
-  size_t current_buffer_;
-  int position_;
-  int last_returned_size_;
-
-  DISALLOW_COPY_AND_ASSIGN(MultipleArrayInputStream);
-};
-
-}  // namespace remoting
-
-#endif  // REMOTING_BASE_MULTIPLE_ARRAY_INPUT_STREAM_H_
diff --git a/remoting/base/multiple_array_input_stream_unittest.cc b/remoting/base/multiple_array_input_stream_unittest.cc
deleted file mode 100644
index 1a21add..0000000
--- a/remoting/base/multiple_array_input_stream_unittest.cc
+++ /dev/null
@@ -1,98 +0,0 @@
-// Copyright (c) 2010 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 <string>
-
-#include "base/scoped_ptr.h"
-#include "net/base/io_buffer.h"
-#include "remoting/base/multiple_array_input_stream.h"
-#include "testing/gtest/include/gtest/gtest.h"
-
-namespace remoting {
-
-// TODO(sergeyu): Add SCOPED_TRACE() for ReadFromInput() and ReadString().
-
-static size_t ReadFromInput(MultipleArrayInputStream* input,
-                         void* data, size_t size) {
-  uint8* out = reinterpret_cast<uint8*>(data);
-  int out_size = size;
-
-  const void* in;
-  int in_size = 0;
-
-  while (true) {
-    if (!input->Next(&in, &in_size)) {
-      return size - out_size;
-    }
-    EXPECT_GT(in_size, -1);
-
-    if (out_size <= in_size) {
-      memcpy(out, in, out_size);
-      if (in_size > out_size) {
-        input->BackUp(in_size - out_size);
-      }
-      return size;  // Copied all of it.
-    }
-
-    memcpy(out, in, in_size);
-    out += in_size;
-    out_size -= in_size;
-  }
-}
-
-static void ReadString(MultipleArrayInputStream* input,
-                       const std::string& str) {
-  scoped_array<char> buffer(new char[str.size() + 1]);
-  buffer[str.size()] = '\0';
-  EXPECT_EQ(ReadFromInput(input, buffer.get(), str.size()), str.size());
-  EXPECT_STREQ(str.data(), buffer.get());
-}
-
-// Construct and prepare data in the |output_stream|.
-static void PrepareData(scoped_ptr<MultipleArrayInputStream>* stream) {
-  static const std::string kTestData =
-      "Hello world!"
-      "This is testing"
-      "MultipleArrayInputStream"
-      "for Chromoting";
-
-  // Determine how many segments to split kTestData. We split the data in
-  // 1 character, 2 characters, 1 character, 2 characters ...
-  int segments = (kTestData.length() / 3) * 2;
-  int remaining_chars = kTestData.length() % 3;
-  if (remaining_chars) {
-    if (remaining_chars == 1)
-      ++segments;
-    else
-      segments += 2;
-  }
-
-  MultipleArrayInputStream* mstream = new MultipleArrayInputStream();
-  const char* data = kTestData.data();
-  for (int i = 0; i < segments; ++i) {
-    int size = i % 2 == 0 ? 1 : 2;
-    mstream->AddBuffer(new net::StringIOBuffer(std::string(data, size)), size);
-    data += size;
-  }
-  stream->reset(mstream);
-}
-
-TEST(MultipleArrayInputStreamTest, BasicOperations) {
-  scoped_ptr<MultipleArrayInputStream> stream;
-  PrepareData(&stream);
-
-  ReadString(stream.get(), "Hello world!");
-  ReadString(stream.get(), "This ");
-  ReadString(stream.get(), "is test");
-  EXPECT_TRUE(stream->Skip(3));
-  ReadString(stream.get(), "MultipleArrayInput");
-  EXPECT_TRUE(stream->Skip(6));
-  ReadString(stream.get(), "f");
-  ReadString(stream.get(), "o");
-  ReadString(stream.get(), "r");
-  ReadString(stream.get(), " ");
-  ReadString(stream.get(), "Chromoting");
-}
-
-}  // namespace remoting
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;
 
diff --git a/remoting/remoting.gyp b/remoting/remoting.gyp
index 49f28a2..4402230 100644
--- a/remoting/remoting.gyp
+++ b/remoting/remoting.gyp
@@ -132,6 +132,8 @@
       'sources': [
         'base/capture_data.cc',
         'base/capture_data.h',
+        'base/compound_buffer.cc',
+        'base/compound_buffer.h',
         'base/compressor.h',
         'base/compressor_zlib.cc',
         'base/compressor_zlib.h',
@@ -154,8 +156,6 @@
         'base/encoder_vp8.h',
         'base/encoder_zlib.cc',
         'base/encoder_zlib.h',
-        'base/multiple_array_input_stream.cc',
-        'base/multiple_array_input_stream.h',
         'base/tracer.cc',
         'base/tracer.h',
         'base/types.h',
@@ -448,6 +448,7 @@
       'sources': [
 # BUG57351        'base/codec_test.cc',
 # BUG57351        'base/codec_test.h',
+        'base/compound_buffer_unittest.cc',
         'base/compressor_zlib_unittest.cc',
 # BUG57351        'base/decoder_vp8_unittest.cc',
         'base/decompressor_zlib_unittest.cc',
@@ -456,7 +457,6 @@
 # BUG57351        'base/encoder_vp8_unittest.cc',
 # BUG57351        'base/encoder_zlib_unittest.cc',
         'base/mock_objects.h',
-        'base/multiple_array_input_stream_unittest.cc',
 # BUG57351        'client/chromoting_view_unittest.cc',
         'client/mock_objects.h',
         'host/access_verifier_unittest.cc',