1 // Copyright (c) 2013 The Chromium Authors. All rights reserved.
2 // Use of this source code is governed by a BSD-style license that can be
3 // found in the LICENSE file.
4
5 #include "content/browser/streams/stream.h"
6
7 #include "base/bind.h"
8 #include "base/location.h"
9 #include "base/message_loop/message_loop_proxy.h"
10 #include "content/browser/streams/stream_handle_impl.h"
11 #include "content/browser/streams/stream_read_observer.h"
12 #include "content/browser/streams/stream_registry.h"
13 #include "content/browser/streams/stream_write_observer.h"
14 #include "net/base/io_buffer.h"
15
16 namespace {
17 // Start throttling the connection at about 1MB.
18 const size_t kDeferSizeThreshold = 40 * 32768;
19 }
20
21 namespace content {
22
Stream(StreamRegistry * registry,StreamWriteObserver * write_observer,const GURL & url)23 Stream::Stream(StreamRegistry* registry,
24 StreamWriteObserver* write_observer,
25 const GURL& url)
26 : can_add_data_(true),
27 url_(url),
28 data_length_(0),
29 data_bytes_read_(0),
30 last_total_buffered_bytes_(0),
31 registry_(registry),
32 read_observer_(NULL),
33 write_observer_(write_observer),
34 stream_handle_(NULL),
35 weak_ptr_factory_(this) {
36 CreateByteStream(base::MessageLoopProxy::current(),
37 base::MessageLoopProxy::current(),
38 kDeferSizeThreshold,
39 &writer_,
40 &reader_);
41
42 // Setup callback for writing.
43 writer_->RegisterCallback(base::Bind(&Stream::OnSpaceAvailable,
44 weak_ptr_factory_.GetWeakPtr()));
45 reader_->RegisterCallback(base::Bind(&Stream::OnDataAvailable,
46 weak_ptr_factory_.GetWeakPtr()));
47
48 registry_->RegisterStream(this);
49 }
50
~Stream()51 Stream::~Stream() {
52 }
53
SetReadObserver(StreamReadObserver * observer)54 bool Stream::SetReadObserver(StreamReadObserver* observer) {
55 if (read_observer_)
56 return false;
57 read_observer_ = observer;
58 return true;
59 }
60
RemoveReadObserver(StreamReadObserver * observer)61 void Stream::RemoveReadObserver(StreamReadObserver* observer) {
62 DCHECK(observer == read_observer_);
63 read_observer_ = NULL;
64 }
65
RemoveWriteObserver(StreamWriteObserver * observer)66 void Stream::RemoveWriteObserver(StreamWriteObserver* observer) {
67 DCHECK(observer == write_observer_);
68 write_observer_ = NULL;
69 }
70
Abort()71 void Stream::Abort() {
72 // Clear all buffer. It's safe to clear reader_ here since the same thread
73 // is used for both input and output operation.
74 writer_.reset();
75 reader_.reset();
76 ClearBuffer();
77 can_add_data_ = false;
78 registry_->UnregisterStream(url());
79 }
80
AddData(scoped_refptr<net::IOBuffer> buffer,size_t size)81 void Stream::AddData(scoped_refptr<net::IOBuffer> buffer, size_t size) {
82 if (!writer_.get())
83 return;
84
85 size_t current_buffered_bytes = writer_->GetTotalBufferedBytes();
86 if (!registry_->UpdateMemoryUsage(url(), current_buffered_bytes, size)) {
87 Abort();
88 return;
89 }
90
91 // Now it's guaranteed that this doesn't overflow. This must be done before
92 // Write() since GetTotalBufferedBytes() may return different value after
93 // Write() call, so if we use the new value, information in this instance and
94 // one in |registry_| become inconsistent.
95 last_total_buffered_bytes_ = current_buffered_bytes + size;
96
97 can_add_data_ = writer_->Write(buffer, size);
98 }
99
AddData(const char * data,size_t size)100 void Stream::AddData(const char* data, size_t size) {
101 if (!writer_.get())
102 return;
103
104 scoped_refptr<net::IOBuffer> io_buffer(new net::IOBuffer(size));
105 memcpy(io_buffer->data(), data, size);
106 AddData(io_buffer, size);
107 }
108
Finalize()109 void Stream::Finalize() {
110 if (!writer_.get())
111 return;
112
113 writer_->Close(0);
114 writer_.reset();
115
116 // Continue asynchronously.
117 base::MessageLoopProxy::current()->PostTask(
118 FROM_HERE,
119 base::Bind(&Stream::OnDataAvailable, weak_ptr_factory_.GetWeakPtr()));
120 }
121
ReadRawData(net::IOBuffer * buf,int buf_size,int * bytes_read)122 Stream::StreamState Stream::ReadRawData(net::IOBuffer* buf,
123 int buf_size,
124 int* bytes_read) {
125 DCHECK(buf);
126 DCHECK(bytes_read);
127
128 *bytes_read = 0;
129 if (!data_.get()) {
130 DCHECK(!data_length_);
131 DCHECK(!data_bytes_read_);
132
133 if (!reader_.get())
134 return STREAM_ABORTED;
135
136 ByteStreamReader::StreamState state = reader_->Read(&data_, &data_length_);
137 switch (state) {
138 case ByteStreamReader::STREAM_HAS_DATA:
139 break;
140 case ByteStreamReader::STREAM_COMPLETE:
141 registry_->UnregisterStream(url());
142 return STREAM_COMPLETE;
143 case ByteStreamReader::STREAM_EMPTY:
144 return STREAM_EMPTY;
145 }
146 }
147
148 const size_t remaining_bytes = data_length_ - data_bytes_read_;
149 size_t to_read =
150 static_cast<size_t>(buf_size) < remaining_bytes ?
151 buf_size : remaining_bytes;
152 memcpy(buf->data(), data_->data() + data_bytes_read_, to_read);
153 data_bytes_read_ += to_read;
154 if (data_bytes_read_ >= data_length_)
155 ClearBuffer();
156
157 *bytes_read = to_read;
158 return STREAM_HAS_DATA;
159 }
160
CreateHandle(const GURL & original_url,const std::string & mime_type)161 scoped_ptr<StreamHandle> Stream::CreateHandle(const GURL& original_url,
162 const std::string& mime_type) {
163 CHECK(!stream_handle_);
164 stream_handle_ = new StreamHandleImpl(weak_ptr_factory_.GetWeakPtr(),
165 original_url,
166 mime_type);
167 return scoped_ptr<StreamHandle>(stream_handle_).Pass();
168 }
169
CloseHandle()170 void Stream::CloseHandle() {
171 // Prevent deletion until this function ends.
172 scoped_refptr<Stream> ref(this);
173
174 CHECK(stream_handle_);
175 stream_handle_ = NULL;
176 registry_->UnregisterStream(url());
177 if (write_observer_)
178 write_observer_->OnClose(this);
179 }
180
OnSpaceAvailable()181 void Stream::OnSpaceAvailable() {
182 can_add_data_ = true;
183 if (write_observer_)
184 write_observer_->OnSpaceAvailable(this);
185 }
186
OnDataAvailable()187 void Stream::OnDataAvailable() {
188 if (read_observer_)
189 read_observer_->OnDataAvailable(this);
190 }
191
ClearBuffer()192 void Stream::ClearBuffer() {
193 data_ = NULL;
194 data_length_ = 0;
195 data_bytes_read_ = 0;
196 }
197
198 } // namespace content
199