OLD | NEW |
---|---|
1 // Copyright (c) 2013 The Chromium Authors. All rights reserved. | 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 | 2 // Use of this source code is governed by a BSD-style license that can be |
3 // found in the LICENSE file. | 3 // found in the LICENSE file. |
4 | 4 |
5 #include "content/browser/streams/stream.h" | 5 #include "content/browser/streams/stream.h" |
6 | 6 |
7 #include "base/bind.h" | 7 #include "base/bind.h" |
8 #include "base/location.h" | 8 #include "base/location.h" |
9 #include "base/message_loop/message_loop_proxy.h" | 9 #include "base/message_loop/message_loop_proxy.h" |
10 #include "content/browser/streams/stream_handle_impl.h" | 10 #include "content/browser/streams/stream_handle_impl.h" |
11 #include "content/browser/streams/stream_read_observer.h" | 11 #include "content/browser/streams/stream_read_observer.h" |
12 #include "content/browser/streams/stream_registry.h" | 12 #include "content/browser/streams/stream_registry.h" |
13 #include "content/browser/streams/stream_write_observer.h" | 13 #include "content/browser/streams/stream_write_observer.h" |
14 #include "net/base/io_buffer.h" | 14 #include "net/base/io_buffer.h" |
15 | 15 |
16 namespace { | 16 namespace { |
17 // Start throttling the connection at about 1MB. | 17 // Start throttling the connection at about 1MB. |
18 const size_t kDeferSizeThreshold = 40 * 32768; | 18 const size_t kDeferSizeThreshold = 40 * 32768; |
19 } | 19 } |
20 | 20 |
21 namespace content { | 21 namespace content { |
22 | 22 |
23 Stream::Stream(StreamRegistry* registry, | 23 Stream::Stream(StreamRegistry* registry, |
24 StreamWriteObserver* write_observer, | 24 StreamWriteObserver* write_observer, |
25 const GURL& url) | 25 const GURL& url) |
26 : data_bytes_read_(0), | 26 : data_bytes_read_(0), |
27 can_add_data_(true), | 27 can_add_data_(true), |
28 url_(url), | 28 url_(url), |
29 data_length_(0), | 29 data_length_(0), |
30 last_total_buffered_bytes_(0), | |
30 registry_(registry), | 31 registry_(registry), |
31 read_observer_(NULL), | 32 read_observer_(NULL), |
32 write_observer_(write_observer), | 33 write_observer_(write_observer), |
33 stream_handle_(NULL), | 34 stream_handle_(NULL), |
34 weak_ptr_factory_(this) { | 35 weak_ptr_factory_(this) { |
35 CreateByteStream(base::MessageLoopProxy::current(), | 36 CreateByteStream(base::MessageLoopProxy::current(), |
36 base::MessageLoopProxy::current(), | 37 base::MessageLoopProxy::current(), |
37 kDeferSizeThreshold, | 38 kDeferSizeThreshold, |
38 &writer_, | 39 &writer_, |
39 &reader_); | 40 &reader_); |
(...skipping 21 matching lines...) Expand all Loading... | |
61 DCHECK(observer == read_observer_); | 62 DCHECK(observer == read_observer_); |
62 read_observer_ = NULL; | 63 read_observer_ = NULL; |
63 } | 64 } |
64 | 65 |
65 void Stream::RemoveWriteObserver(StreamWriteObserver* observer) { | 66 void Stream::RemoveWriteObserver(StreamWriteObserver* observer) { |
66 DCHECK(observer == write_observer_); | 67 DCHECK(observer == write_observer_); |
67 write_observer_ = NULL; | 68 write_observer_ = NULL; |
68 } | 69 } |
69 | 70 |
70 void Stream::AddData(scoped_refptr<net::IOBuffer> buffer, size_t size) { | 71 void Stream::AddData(scoped_refptr<net::IOBuffer> buffer, size_t size) { |
72 if (!writer_.get()) | |
73 return; | |
74 | |
75 size_t new_size = writer_->TotalBufferedBytes() + size; | |
kinuko
2013/08/14 09:32:23
ditto. Could this overflow?
tyoshino (SeeGerritForStatus)
2013/08/15 04:15:33
Added overflow check.
| |
76 if (!registry_->CanIncreaseMemoryUsage( | |
77 url(), last_total_buffered_bytes_, new_size)) { | |
78 // Clear all buffer. It's safe to clear reader_ here since the same thread | |
79 // is used for both input and output operation. | |
80 writer_.reset(); | |
81 reader_.reset(); | |
82 can_add_data_ = false; | |
83 registry_->UnregisterStream(url()); | |
84 return; | |
85 } | |
71 can_add_data_ = writer_->Write(buffer, size); | 86 can_add_data_ = writer_->Write(buffer, size); |
87 last_total_buffered_bytes_ = new_size; | |
72 } | 88 } |
73 | 89 |
74 void Stream::AddData(const char* data, size_t size) { | 90 void Stream::AddData(const char* data, size_t size) { |
75 scoped_refptr<net::IOBuffer> io_buffer(new net::IOBuffer(size)); | 91 scoped_refptr<net::IOBuffer> io_buffer(new net::IOBuffer(size)); |
76 memcpy(io_buffer->data(), data, size); | 92 memcpy(io_buffer->data(), data, size); |
77 can_add_data_ = writer_->Write(io_buffer, size); | 93 AddData(io_buffer, size); |
78 } | 94 } |
79 | 95 |
80 void Stream::Finalize() { | 96 void Stream::Finalize() { |
97 if (!writer_.get()) | |
98 return; | |
99 | |
81 writer_->Close(0); | 100 writer_->Close(0); |
82 writer_.reset(NULL); | 101 writer_.reset(); |
83 | 102 |
84 // Continue asynchronously. | 103 // Continue asynchronously. |
85 base::MessageLoopProxy::current()->PostTask( | 104 base::MessageLoopProxy::current()->PostTask( |
86 FROM_HERE, | 105 FROM_HERE, |
87 base::Bind(&Stream::OnDataAvailable, weak_ptr_factory_.GetWeakPtr())); | 106 base::Bind(&Stream::OnDataAvailable, weak_ptr_factory_.GetWeakPtr())); |
88 } | 107 } |
89 | 108 |
90 Stream::StreamState Stream::ReadRawData(net::IOBuffer* buf, | 109 Stream::StreamState Stream::ReadRawData(net::IOBuffer* buf, |
91 int buf_size, | 110 int buf_size, |
92 int* bytes_read) { | 111 int* bytes_read) { |
93 *bytes_read = 0; | 112 *bytes_read = 0; |
94 if (!data_.get()) { | 113 if (!data_.get()) { |
114 // TODO(tyoshino): Add STREAM_ABORTED type to tell the reader that this | |
115 // stream is aborted. | |
116 if (!reader_.get()) | |
117 return STREAM_EMPTY; | |
118 | |
95 data_length_ = 0; | 119 data_length_ = 0; |
96 data_bytes_read_ = 0; | 120 data_bytes_read_ = 0; |
97 ByteStreamReader::StreamState state = reader_->Read(&data_, &data_length_); | 121 ByteStreamReader::StreamState state = reader_->Read(&data_, &data_length_); |
98 switch (state) { | 122 switch (state) { |
99 case ByteStreamReader::STREAM_HAS_DATA: | 123 case ByteStreamReader::STREAM_HAS_DATA: |
100 break; | 124 break; |
101 case ByteStreamReader::STREAM_COMPLETE: | 125 case ByteStreamReader::STREAM_COMPLETE: |
102 registry_->UnregisterStream(url()); | 126 registry_->UnregisterStream(url()); |
103 return STREAM_COMPLETE; | 127 return STREAM_COMPLETE; |
104 case ByteStreamReader::STREAM_EMPTY: | 128 case ByteStreamReader::STREAM_EMPTY: |
(...skipping 39 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... | |
144 if (write_observer_) | 168 if (write_observer_) |
145 write_observer_->OnSpaceAvailable(this); | 169 write_observer_->OnSpaceAvailable(this); |
146 } | 170 } |
147 | 171 |
148 void Stream::OnDataAvailable() { | 172 void Stream::OnDataAvailable() { |
149 if (read_observer_) | 173 if (read_observer_) |
150 read_observer_->OnDataAvailable(this); | 174 read_observer_->OnDataAvailable(this); |
151 } | 175 } |
152 | 176 |
153 } // namespace content | 177 } // namespace content |
OLD | NEW |