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" |
(...skipping 14 matching lines...) Expand all Loading... |
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 registry_(registry), | 30 registry_(registry), |
31 read_observer_(NULL), | 31 read_observer_(NULL), |
32 write_observer_(write_observer), | 32 write_observer_(write_observer), |
33 stream_handle_(NULL), | 33 stream_handle_(NULL), |
34 weak_ptr_factory_(this) { | 34 weak_ptr_factory_(this) { |
35 CreateByteStream(base::MessageLoopProxy::current(), | 35 CreateByteStream<bool>(base::MessageLoopProxy::current(), |
36 base::MessageLoopProxy::current(), | 36 base::MessageLoopProxy::current(), |
37 kDeferSizeThreshold, | 37 kDeferSizeThreshold, |
38 &writer_, | 38 &writer_, |
39 &reader_); | 39 &reader_); |
40 | 40 |
41 // Setup callback for writing. | 41 // Setup callback for writing. |
42 writer_->RegisterCallback(base::Bind(&Stream::OnSpaceAvailable, | 42 writer_->RegisterCallback(base::Bind(&Stream::OnSpaceAvailable, |
43 weak_ptr_factory_.GetWeakPtr())); | 43 weak_ptr_factory_.GetWeakPtr())); |
44 reader_->RegisterCallback(base::Bind(&Stream::OnDataAvailable, | 44 reader_->RegisterCallback(base::Bind(&Stream::OnDataAvailable, |
45 weak_ptr_factory_.GetWeakPtr())); | 45 weak_ptr_factory_.GetWeakPtr())); |
46 | 46 |
47 registry_->RegisterStream(this); | 47 registry_->RegisterStream(this); |
48 } | 48 } |
49 | 49 |
(...skipping 15 matching lines...) Expand all Loading... |
65 void Stream::RemoveWriteObserver(StreamWriteObserver* observer) { | 65 void Stream::RemoveWriteObserver(StreamWriteObserver* observer) { |
66 DCHECK(observer == write_observer_); | 66 DCHECK(observer == write_observer_); |
67 write_observer_ = NULL; | 67 write_observer_ = NULL; |
68 } | 68 } |
69 | 69 |
70 void Stream::AddData(scoped_refptr<net::IOBuffer> buffer, size_t size) { | 70 void Stream::AddData(scoped_refptr<net::IOBuffer> buffer, size_t size) { |
71 can_add_data_ = writer_->Write(buffer, size); | 71 can_add_data_ = writer_->Write(buffer, size); |
72 } | 72 } |
73 | 73 |
74 void Stream::Finalize() { | 74 void Stream::Finalize() { |
75 writer_->Close(DOWNLOAD_INTERRUPT_REASON_NONE); | 75 // We don't use the status passing feature of ByteStream. |
| 76 writer_->Close(false /* status */); |
76 writer_.reset(NULL); | 77 writer_.reset(NULL); |
77 | 78 |
78 // Continue asynchronously. | 79 // Continue asynchronously. |
79 base::MessageLoopProxy::current()->PostTask( | 80 base::MessageLoopProxy::current()->PostTask( |
80 FROM_HERE, | 81 FROM_HERE, |
81 base::Bind(&Stream::OnDataAvailable, weak_ptr_factory_.GetWeakPtr())); | 82 base::Bind(&Stream::OnDataAvailable, weak_ptr_factory_.GetWeakPtr())); |
82 } | 83 } |
83 | 84 |
84 Stream::StreamState Stream::ReadRawData(net::IOBuffer* buf, | 85 Stream::StreamState Stream::ReadRawData(net::IOBuffer* buf, |
85 int buf_size, | 86 int buf_size, |
86 int* bytes_read) { | 87 int* bytes_read) { |
87 *bytes_read = 0; | 88 *bytes_read = 0; |
88 if (!data_.get()) { | 89 if (!data_.get()) { |
89 data_length_ = 0; | 90 data_length_ = 0; |
90 data_bytes_read_ = 0; | 91 data_bytes_read_ = 0; |
91 ByteStreamReader::StreamState state = reader_->Read(&data_, &data_length_); | 92 ByteStreamReader<bool>::StreamState state = |
| 93 reader_->Read(&data_, &data_length_); |
92 switch (state) { | 94 switch (state) { |
93 case ByteStreamReader::STREAM_HAS_DATA: | 95 case ByteStreamReader<bool>::STREAM_HAS_DATA: |
94 break; | 96 break; |
95 case ByteStreamReader::STREAM_COMPLETE: | 97 case ByteStreamReader<bool>::STREAM_COMPLETE: |
96 registry_->UnregisterStream(url()); | 98 registry_->UnregisterStream(url()); |
97 return STREAM_COMPLETE; | 99 return STREAM_COMPLETE; |
98 case ByteStreamReader::STREAM_EMPTY: | 100 case ByteStreamReader<bool>::STREAM_EMPTY: |
99 return STREAM_EMPTY; | 101 return STREAM_EMPTY; |
100 } | 102 } |
101 } | 103 } |
102 | 104 |
103 const size_t remaining_bytes = data_length_ - data_bytes_read_; | 105 const size_t remaining_bytes = data_length_ - data_bytes_read_; |
104 size_t to_read = | 106 size_t to_read = |
105 static_cast<size_t>(buf_size) < remaining_bytes ? | 107 static_cast<size_t>(buf_size) < remaining_bytes ? |
106 buf_size : remaining_bytes; | 108 buf_size : remaining_bytes; |
107 memcpy(buf->data(), data_->data() + data_bytes_read_, to_read); | 109 memcpy(buf->data(), data_->data() + data_bytes_read_, to_read); |
108 data_bytes_read_ += to_read; | 110 data_bytes_read_ += to_read; |
(...skipping 29 matching lines...) Expand all Loading... |
138 if (write_observer_) | 140 if (write_observer_) |
139 write_observer_->OnSpaceAvailable(this); | 141 write_observer_->OnSpaceAvailable(this); |
140 } | 142 } |
141 | 143 |
142 void Stream::OnDataAvailable() { | 144 void Stream::OnDataAvailable() { |
143 if (read_observer_) | 145 if (read_observer_) |
144 read_observer_->OnDataAvailable(this); | 146 read_observer_->OnDataAvailable(this); |
145 } | 147 } |
146 | 148 |
147 } // namespace content | 149 } // namespace content |
OLD | NEW |