diff options
author | sergeyu@chromium.org <sergeyu@chromium.org@0039d316-1c4b-4281-b951-d872f2087c98> | 2010-11-16 01:27:46 +0000 |
---|---|---|
committer | sergeyu@chromium.org <sergeyu@chromium.org@0039d316-1c4b-4281-b951-d872f2087c98> | 2010-11-16 01:27:46 +0000 |
commit | b3c03409a5538a9aa4577b10f340e14fd71d99dc (patch) | |
tree | 679974a34a0d1414ea8699f555c52b9fc6ad408f | |
parent | aa32bccb292a8e9e153bd44544558e36f440f21e (diff) | |
download | chromium_src-b3c03409a5538a9aa4577b10f340e14fd71d99dc.zip chromium_src-b3c03409a5538a9aa4577b10f340e14fd71d99dc.tar.gz chromium_src-b3c03409a5538a9aa4577b10f340e14fd71d99dc.tar.bz2 |
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', |