This source file includes following definitions.
- peer_
- SetPeer
- Flush
- Close
- RegisterCallback
- GetTotalBufferedBytes
- UpdateWindow
- UpdateWindowInternal
- PostToPeer
- peer_
- SetPeer
- GetStatus
- RegisterCallback
- TransferData
- TransferDataInternal
- MaybeUpdateInput
- CreateByteStream
#include "content/browser/byte_stream.h"
#include <deque>
#include <set>
#include <utility>
#include "base/bind.h"
#include "base/location.h"
#include "base/memory/ref_counted.h"
#include "base/sequenced_task_runner.h"
namespace content {
namespace {
typedef std::deque<std::pair<scoped_refptr<net::IOBuffer>, size_t> >
ContentVector;
class ByteStreamReaderImpl;
struct LifetimeFlag : public base::RefCountedThreadSafe<LifetimeFlag> {
public:
LifetimeFlag() : is_alive(true) { }
bool is_alive;
protected:
friend class base::RefCountedThreadSafe<LifetimeFlag>;
virtual ~LifetimeFlag() { }
private:
DISALLOW_COPY_AND_ASSIGN(LifetimeFlag);
};
class ByteStreamWriterImpl : public ByteStreamWriter {
public:
ByteStreamWriterImpl(scoped_refptr<base::SequencedTaskRunner> task_runner,
scoped_refptr<LifetimeFlag> lifetime_flag,
size_t buffer_size);
virtual ~ByteStreamWriterImpl();
void SetPeer(ByteStreamReaderImpl* peer,
scoped_refptr<base::SequencedTaskRunner> peer_task_runner,
scoped_refptr<LifetimeFlag> peer_lifetime_flag);
virtual bool Write(scoped_refptr<net::IOBuffer> buffer,
size_t byte_count) OVERRIDE;
virtual void Flush() OVERRIDE;
virtual void Close(int status) OVERRIDE;
virtual void RegisterCallback(const base::Closure& source_callback) OVERRIDE;
virtual size_t GetTotalBufferedBytes() const OVERRIDE;
static void UpdateWindow(scoped_refptr<LifetimeFlag> lifetime_flag,
ByteStreamWriterImpl* target,
size_t bytes_consumed);
private:
void UpdateWindowInternal(size_t bytes_consumed);
void PostToPeer(bool complete, int status);
const size_t total_buffer_size_;
scoped_refptr<base::SequencedTaskRunner> my_task_runner_;
scoped_refptr<LifetimeFlag> my_lifetime_flag_;
base::Closure space_available_callback_;
ContentVector input_contents_;
size_t input_contents_size_;
scoped_refptr<base::SequencedTaskRunner> peer_task_runner_;
size_t output_size_used_;
scoped_refptr<LifetimeFlag> peer_lifetime_flag_;
ByteStreamReaderImpl* peer_;
};
class ByteStreamReaderImpl : public ByteStreamReader {
public:
ByteStreamReaderImpl(scoped_refptr<base::SequencedTaskRunner> task_runner,
scoped_refptr<LifetimeFlag> lifetime_flag,
size_t buffer_size);
virtual ~ByteStreamReaderImpl();
void SetPeer(ByteStreamWriterImpl* peer,
scoped_refptr<base::SequencedTaskRunner> peer_task_runner,
scoped_refptr<LifetimeFlag> peer_lifetime_flag);
virtual StreamState Read(scoped_refptr<net::IOBuffer>* data,
size_t* length) OVERRIDE;
virtual int GetStatus() const OVERRIDE;
virtual void RegisterCallback(const base::Closure& sink_callback) OVERRIDE;
static void TransferData(
scoped_refptr<LifetimeFlag> object_lifetime_flag,
ByteStreamReaderImpl* target,
scoped_ptr<ContentVector> transfer_buffer,
size_t transfer_buffer_bytes,
bool source_complete,
int status);
private:
void TransferDataInternal(
scoped_ptr<ContentVector> transfer_buffer,
size_t transfer_buffer_bytes,
bool source_complete,
int status);
void MaybeUpdateInput();
const size_t total_buffer_size_;
scoped_refptr<base::SequencedTaskRunner> my_task_runner_;
scoped_refptr<LifetimeFlag> my_lifetime_flag_;
ContentVector available_contents_;
bool received_status_;
int status_;
base::Closure data_available_callback_;
base::Time last_non_full_time_;
scoped_refptr<base::SequencedTaskRunner> peer_task_runner_;
size_t unreported_consumed_bytes_;
scoped_refptr<LifetimeFlag> peer_lifetime_flag_;
ByteStreamWriterImpl* peer_;
};
ByteStreamWriterImpl::ByteStreamWriterImpl(
scoped_refptr<base::SequencedTaskRunner> task_runner,
scoped_refptr<LifetimeFlag> lifetime_flag,
size_t buffer_size)
: total_buffer_size_(buffer_size),
my_task_runner_(task_runner),
my_lifetime_flag_(lifetime_flag),
input_contents_size_(0),
output_size_used_(0),
peer_(NULL) {
DCHECK(my_lifetime_flag_.get());
my_lifetime_flag_->is_alive = true;
}
ByteStreamWriterImpl::~ByteStreamWriterImpl() {
my_lifetime_flag_->is_alive = false;
}
void ByteStreamWriterImpl::SetPeer(
ByteStreamReaderImpl* peer,
scoped_refptr<base::SequencedTaskRunner> peer_task_runner,
scoped_refptr<LifetimeFlag> peer_lifetime_flag) {
peer_ = peer;
peer_task_runner_ = peer_task_runner;
peer_lifetime_flag_ = peer_lifetime_flag;
}
bool ByteStreamWriterImpl::Write(
scoped_refptr<net::IOBuffer> buffer, size_t byte_count) {
DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
size_t space_limit = std::numeric_limits<size_t>::max() -
GetTotalBufferedBytes();
if (byte_count > space_limit) {
return false;
}
input_contents_.push_back(std::make_pair(buffer, byte_count));
input_contents_size_ += byte_count;
if (input_contents_size_ > total_buffer_size_ / kFractionBufferBeforeSending)
PostToPeer(false, 0);
return GetTotalBufferedBytes() <= total_buffer_size_;
}
void ByteStreamWriterImpl::Flush() {
DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
if (input_contents_size_ > 0)
PostToPeer(false, 0);
}
void ByteStreamWriterImpl::Close(int status) {
DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
PostToPeer(true, status);
}
void ByteStreamWriterImpl::RegisterCallback(
const base::Closure& source_callback) {
DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
space_available_callback_ = source_callback;
}
size_t ByteStreamWriterImpl::GetTotalBufferedBytes() const {
DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
return input_contents_size_ + output_size_used_;
}
void ByteStreamWriterImpl::UpdateWindow(
scoped_refptr<LifetimeFlag> lifetime_flag, ByteStreamWriterImpl* target,
size_t bytes_consumed) {
if (!lifetime_flag->is_alive) return;
target->UpdateWindowInternal(bytes_consumed);
}
void ByteStreamWriterImpl::UpdateWindowInternal(size_t bytes_consumed) {
DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
bool was_above_limit = GetTotalBufferedBytes() > total_buffer_size_;
DCHECK_GE(output_size_used_, bytes_consumed);
output_size_used_ -= bytes_consumed;
bool no_longer_above_limit = GetTotalBufferedBytes() <= total_buffer_size_;
if (no_longer_above_limit && was_above_limit &&
!space_available_callback_.is_null())
space_available_callback_.Run();
}
void ByteStreamWriterImpl::PostToPeer(bool complete, int status) {
DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
DCHECK(complete || 0 != input_contents_size_);
scoped_ptr<ContentVector> transfer_buffer;
size_t buffer_size = 0;
if (0 != input_contents_size_) {
transfer_buffer.reset(new ContentVector);
transfer_buffer->swap(input_contents_);
buffer_size = input_contents_size_;
output_size_used_ += input_contents_size_;
input_contents_size_ = 0;
}
peer_task_runner_->PostTask(
FROM_HERE, base::Bind(
&ByteStreamReaderImpl::TransferData,
peer_lifetime_flag_,
peer_,
base::Passed(&transfer_buffer),
buffer_size,
complete,
status));
}
ByteStreamReaderImpl::ByteStreamReaderImpl(
scoped_refptr<base::SequencedTaskRunner> task_runner,
scoped_refptr<LifetimeFlag> lifetime_flag,
size_t buffer_size)
: total_buffer_size_(buffer_size),
my_task_runner_(task_runner),
my_lifetime_flag_(lifetime_flag),
received_status_(false),
status_(0),
unreported_consumed_bytes_(0),
peer_(NULL) {
DCHECK(my_lifetime_flag_.get());
my_lifetime_flag_->is_alive = true;
}
ByteStreamReaderImpl::~ByteStreamReaderImpl() {
my_lifetime_flag_->is_alive = false;
}
void ByteStreamReaderImpl::SetPeer(
ByteStreamWriterImpl* peer,
scoped_refptr<base::SequencedTaskRunner> peer_task_runner,
scoped_refptr<LifetimeFlag> peer_lifetime_flag) {
peer_ = peer;
peer_task_runner_ = peer_task_runner;
peer_lifetime_flag_ = peer_lifetime_flag;
}
ByteStreamReaderImpl::StreamState
ByteStreamReaderImpl::Read(scoped_refptr<net::IOBuffer>* data,
size_t* length) {
DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
if (available_contents_.size()) {
*data = available_contents_.front().first;
*length = available_contents_.front().second;
available_contents_.pop_front();
unreported_consumed_bytes_ += *length;
MaybeUpdateInput();
return STREAM_HAS_DATA;
}
if (received_status_) {
return STREAM_COMPLETE;
}
return STREAM_EMPTY;
}
int ByteStreamReaderImpl::GetStatus() const {
DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
DCHECK(received_status_);
return status_;
}
void ByteStreamReaderImpl::RegisterCallback(
const base::Closure& sink_callback) {
DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
data_available_callback_ = sink_callback;
}
void ByteStreamReaderImpl::TransferData(
scoped_refptr<LifetimeFlag> object_lifetime_flag,
ByteStreamReaderImpl* target,
scoped_ptr<ContentVector> transfer_buffer,
size_t buffer_size,
bool source_complete,
int status) {
if (!object_lifetime_flag->is_alive) return;
target->TransferDataInternal(
transfer_buffer.Pass(), buffer_size, source_complete, status);
}
void ByteStreamReaderImpl::TransferDataInternal(
scoped_ptr<ContentVector> transfer_buffer,
size_t buffer_size,
bool source_complete,
int status) {
DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
bool was_empty = available_contents_.empty();
if (transfer_buffer) {
available_contents_.insert(available_contents_.end(),
transfer_buffer->begin(),
transfer_buffer->end());
}
if (source_complete) {
received_status_ = true;
status_ = status;
}
if (((was_empty && !available_contents_.empty()) ||
source_complete) &&
!data_available_callback_.is_null())
data_available_callback_.Run();
}
void ByteStreamReaderImpl::MaybeUpdateInput() {
DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
if (unreported_consumed_bytes_ <=
total_buffer_size_ / kFractionReadBeforeWindowUpdate)
return;
peer_task_runner_->PostTask(
FROM_HERE, base::Bind(
&ByteStreamWriterImpl::UpdateWindow,
peer_lifetime_flag_,
peer_,
unreported_consumed_bytes_));
unreported_consumed_bytes_ = 0;
}
}
const int ByteStreamWriter::kFractionBufferBeforeSending = 3;
const int ByteStreamReader::kFractionReadBeforeWindowUpdate = 3;
ByteStreamReader::~ByteStreamReader() { }
ByteStreamWriter::~ByteStreamWriter() { }
void CreateByteStream(
scoped_refptr<base::SequencedTaskRunner> input_task_runner,
scoped_refptr<base::SequencedTaskRunner> output_task_runner,
size_t buffer_size,
scoped_ptr<ByteStreamWriter>* input,
scoped_ptr<ByteStreamReader>* output) {
scoped_refptr<LifetimeFlag> input_flag(new LifetimeFlag());
scoped_refptr<LifetimeFlag> output_flag(new LifetimeFlag());
ByteStreamWriterImpl* in = new ByteStreamWriterImpl(
input_task_runner, input_flag, buffer_size);
ByteStreamReaderImpl* out = new ByteStreamReaderImpl(
output_task_runner, output_flag, buffer_size);
in->SetPeer(out, output_task_runner, output_flag);
out->SetPeer(in, input_task_runner, input_flag);
input->reset(in);
output->reset(out);
}
}