| // Copyright (c) 2013 The Chromium Authors. All rights reserved. |
| // Use of this source code is governed by a BSD-style license that can be |
| // found in the LICENSE file. |
| |
| #include "content/browser/streams/stream.h" |
| |
| #include "base/bind.h" |
| #include "base/location.h" |
| #include "base/message_loop/message_loop_proxy.h" |
| #include "base/values.h" |
| #include "content/browser/streams/stream_handle_impl.h" |
| #include "content/browser/streams/stream_read_observer.h" |
| #include "content/browser/streams/stream_registry.h" |
| #include "content/browser/streams/stream_write_observer.h" |
| #include "net/base/io_buffer.h" |
| #include "net/http/http_response_headers.h" |
| |
| namespace { |
| // Start throttling the connection at about 1MB. |
| const size_t kDeferSizeThreshold = 40 * 32768; |
| } |
| |
| namespace content { |
| |
| Stream::Stream(StreamRegistry* registry, |
| StreamWriteObserver* write_observer, |
| const GURL& url) |
| : can_add_data_(true), |
| url_(url), |
| data_length_(0), |
| data_bytes_read_(0), |
| last_total_buffered_bytes_(0), |
| registry_(registry), |
| read_observer_(NULL), |
| write_observer_(write_observer), |
| stream_handle_(NULL), |
| weak_ptr_factory_(this) { |
| CreateByteStream(base::MessageLoopProxy::current(), |
| base::MessageLoopProxy::current(), |
| kDeferSizeThreshold, |
| &writer_, |
| &reader_); |
| |
| // Setup callback for writing. |
| writer_->RegisterCallback(base::Bind(&Stream::OnSpaceAvailable, |
| weak_ptr_factory_.GetWeakPtr())); |
| reader_->RegisterCallback(base::Bind(&Stream::OnDataAvailable, |
| weak_ptr_factory_.GetWeakPtr())); |
| |
| registry_->RegisterStream(this); |
| } |
| |
| Stream::~Stream() { |
| } |
| |
| bool Stream::SetReadObserver(StreamReadObserver* observer) { |
| if (read_observer_) |
| return false; |
| read_observer_ = observer; |
| return true; |
| } |
| |
| void Stream::RemoveReadObserver(StreamReadObserver* observer) { |
| DCHECK(observer == read_observer_); |
| read_observer_ = NULL; |
| } |
| |
| void Stream::RemoveWriteObserver(StreamWriteObserver* observer) { |
| DCHECK(observer == write_observer_); |
| write_observer_ = NULL; |
| } |
| |
| void Stream::Abort() { |
| // Clear all buffer. It's safe to clear reader_ here since the same thread |
| // is used for both input and output operation. |
| writer_.reset(); |
| reader_.reset(); |
| ClearBuffer(); |
| can_add_data_ = false; |
| registry_->UnregisterStream(url()); |
| } |
| |
| void Stream::AddData(scoped_refptr<net::IOBuffer> buffer, size_t size) { |
| if (!writer_.get()) |
| return; |
| |
| size_t current_buffered_bytes = writer_->GetTotalBufferedBytes(); |
| if (!registry_->UpdateMemoryUsage(url(), current_buffered_bytes, size)) { |
| Abort(); |
| return; |
| } |
| |
| // Now it's guaranteed that this doesn't overflow. This must be done before |
| // Write() since GetTotalBufferedBytes() may return different value after |
| // Write() call, so if we use the new value, information in this instance and |
| // one in |registry_| become inconsistent. |
| last_total_buffered_bytes_ = current_buffered_bytes + size; |
| |
| can_add_data_ = writer_->Write(buffer, size); |
| } |
| |
| void Stream::AddData(const char* data, size_t size) { |
| if (!writer_.get()) |
| return; |
| |
| scoped_refptr<net::IOBuffer> io_buffer(new net::IOBuffer(size)); |
| memcpy(io_buffer->data(), data, size); |
| AddData(io_buffer, size); |
| } |
| |
| void Stream::Finalize() { |
| if (!writer_.get()) |
| return; |
| |
| writer_->Close(0); |
| writer_.reset(); |
| |
| // Continue asynchronously. |
| base::MessageLoopProxy::current()->PostTask( |
| FROM_HERE, |
| base::Bind(&Stream::OnDataAvailable, weak_ptr_factory_.GetWeakPtr())); |
| } |
| |
| Stream::StreamState Stream::ReadRawData(net::IOBuffer* buf, |
| int buf_size, |
| int* bytes_read) { |
| DCHECK(buf); |
| DCHECK(bytes_read); |
| |
| *bytes_read = 0; |
| if (!data_.get()) { |
| DCHECK(!data_length_); |
| DCHECK(!data_bytes_read_); |
| |
| if (!reader_.get()) |
| return STREAM_ABORTED; |
| |
| ByteStreamReader::StreamState state = reader_->Read(&data_, &data_length_); |
| switch (state) { |
| case ByteStreamReader::STREAM_HAS_DATA: |
| break; |
| case ByteStreamReader::STREAM_COMPLETE: |
| registry_->UnregisterStream(url()); |
| return STREAM_COMPLETE; |
| case ByteStreamReader::STREAM_EMPTY: |
| return STREAM_EMPTY; |
| } |
| } |
| |
| const size_t remaining_bytes = data_length_ - data_bytes_read_; |
| size_t to_read = |
| static_cast<size_t>(buf_size) < remaining_bytes ? |
| buf_size : remaining_bytes; |
| memcpy(buf->data(), data_->data() + data_bytes_read_, to_read); |
| data_bytes_read_ += to_read; |
| if (data_bytes_read_ >= data_length_) |
| ClearBuffer(); |
| |
| *bytes_read = to_read; |
| return STREAM_HAS_DATA; |
| } |
| |
| scoped_ptr<StreamHandle> Stream::CreateHandle() { |
| CHECK(!stream_handle_); |
| stream_handle_ = new StreamHandleImpl(weak_ptr_factory_.GetWeakPtr()); |
| return scoped_ptr<StreamHandle>(stream_handle_).Pass(); |
| } |
| |
| void Stream::CloseHandle() { |
| // Prevent deletion until this function ends. |
| scoped_refptr<Stream> ref(this); |
| |
| CHECK(stream_handle_); |
| stream_handle_ = NULL; |
| registry_->UnregisterStream(url()); |
| if (write_observer_) |
| write_observer_->OnClose(this); |
| } |
| |
| void Stream::OnSpaceAvailable() { |
| can_add_data_ = true; |
| if (write_observer_) |
| write_observer_->OnSpaceAvailable(this); |
| } |
| |
| void Stream::OnDataAvailable() { |
| if (read_observer_) |
| read_observer_->OnDataAvailable(this); |
| } |
| |
| void Stream::ClearBuffer() { |
| data_ = NULL; |
| data_length_ = 0; |
| data_bytes_read_ = 0; |
| } |
| |
| } // namespace content |