OLD | NEW |
---|---|
1 // Copyright 2015 The Chromium Authors. All rights reserved. | 1 // Copyright 2015 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 "net/spdy/bidirectional_stream_spdy_impl.h" | 5 #include "net/spdy/bidirectional_stream_spdy_impl.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/logging.h" | 9 #include "base/logging.h" |
10 #include "base/time/time.h" | 10 #include "base/time/time.h" |
(...skipping 42 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... | |
53 bool /*send_request_headers_automatically*/, | 53 bool /*send_request_headers_automatically*/, |
54 BidirectionalStreamImpl::Delegate* delegate, | 54 BidirectionalStreamImpl::Delegate* delegate, |
55 std::unique_ptr<base::Timer> timer) { | 55 std::unique_ptr<base::Timer> timer) { |
56 DCHECK(!stream_); | 56 DCHECK(!stream_); |
57 DCHECK(timer); | 57 DCHECK(timer); |
58 | 58 |
59 delegate_ = delegate; | 59 delegate_ = delegate; |
60 timer_ = std::move(timer); | 60 timer_ = std::move(timer); |
61 | 61 |
62 if (!spdy_session_) { | 62 if (!spdy_session_) { |
63 delegate_->OnFailed(ERR_CONNECTION_CLOSED); | 63 base::ThreadTaskRunnerHandle::Get()->PostTask( |
64 FROM_HERE, | |
65 base::Bind(&BidirectionalStreamSpdyImpl::NotifyError, | |
66 weak_factory_.GetWeakPtr(), ERR_CONNECTION_CLOSED)); | |
64 return; | 67 return; |
65 } | 68 } |
66 | 69 |
67 request_info_ = request_info; | 70 request_info_ = request_info; |
68 | 71 |
69 int rv = stream_request_.StartRequest( | 72 int rv = stream_request_.StartRequest( |
70 SPDY_BIDIRECTIONAL_STREAM, spdy_session_, request_info_->url, | 73 SPDY_BIDIRECTIONAL_STREAM, spdy_session_, request_info_->url, |
71 request_info_->priority, net_log, | 74 request_info_->priority, net_log, |
72 base::Bind(&BidirectionalStreamSpdyImpl::OnStreamInitialized, | 75 base::Bind(&BidirectionalStreamSpdyImpl::OnStreamInitialized, |
73 weak_factory_.GetWeakPtr())); | 76 weak_factory_.GetWeakPtr())); |
(...skipping 23 matching lines...) Expand all Loading... | |
97 // Read will complete asynchronously and Delegate::OnReadCompleted will be | 100 // Read will complete asynchronously and Delegate::OnReadCompleted will be |
98 // called upon completion. | 101 // called upon completion. |
99 read_buffer_ = buf; | 102 read_buffer_ = buf; |
100 read_buffer_len_ = buf_len; | 103 read_buffer_len_ = buf_len; |
101 return ERR_IO_PENDING; | 104 return ERR_IO_PENDING; |
102 } | 105 } |
103 | 106 |
104 void BidirectionalStreamSpdyImpl::SendData(const scoped_refptr<IOBuffer>& data, | 107 void BidirectionalStreamSpdyImpl::SendData(const scoped_refptr<IOBuffer>& data, |
105 int length, | 108 int length, |
106 bool end_stream) { | 109 bool end_stream) { |
110 DCHECK(length > 0 || (length == 0 && end_stream)); | |
111 | |
112 if (!stream_) { | |
113 LOG(ERROR) << "Trying to send data after stream has been destroyed."; | |
114 base::ThreadTaskRunnerHandle::Get()->PostTask( | |
115 FROM_HERE, base::Bind(&BidirectionalStreamSpdyImpl::NotifyError, | |
116 weak_factory_.GetWeakPtr(), ERR_UNEXPECTED)); | |
117 return; | |
118 } | |
119 | |
107 DCHECK(!stream_closed_); | 120 DCHECK(!stream_closed_); |
108 DCHECK(stream_); | |
109 | |
110 stream_->SendData(data.get(), length, | 121 stream_->SendData(data.get(), length, |
111 end_stream ? NO_MORE_DATA_TO_SEND : MORE_DATA_TO_SEND); | 122 end_stream ? NO_MORE_DATA_TO_SEND : MORE_DATA_TO_SEND); |
112 } | 123 } |
113 | 124 |
114 void BidirectionalStreamSpdyImpl::SendvData( | 125 void BidirectionalStreamSpdyImpl::SendvData( |
115 const std::vector<scoped_refptr<IOBuffer>>& buffers, | 126 const std::vector<scoped_refptr<IOBuffer>>& buffers, |
116 const std::vector<int>& lengths, | 127 const std::vector<int>& lengths, |
117 bool end_stream) { | 128 bool end_stream) { |
118 DCHECK(!stream_closed_); | |
119 DCHECK(stream_); | |
120 DCHECK_EQ(buffers.size(), lengths.size()); | 129 DCHECK_EQ(buffers.size(), lengths.size()); |
121 | 130 |
131 if (!stream_) { | |
132 LOG(ERROR) << "Trying to send data after stream has been destroyed."; | |
133 base::ThreadTaskRunnerHandle::Get()->PostTask( | |
134 FROM_HERE, base::Bind(&BidirectionalStreamSpdyImpl::NotifyError, | |
135 weak_factory_.GetWeakPtr(), ERR_UNEXPECTED)); | |
136 return; | |
137 } | |
138 | |
139 DCHECK(!stream_closed_); | |
122 int total_len = 0; | 140 int total_len = 0; |
123 for (int len : lengths) { | 141 for (int len : lengths) { |
124 total_len += len; | 142 total_len += len; |
125 } | 143 } |
126 | 144 |
127 pending_combined_buffer_ = new net::IOBuffer(total_len); | 145 pending_combined_buffer_ = new net::IOBuffer(total_len); |
128 int len = 0; | 146 int len = 0; |
129 // TODO(xunjieli): Get rid of extra copy. Coalesce headers and data frames. | 147 // TODO(xunjieli): Get rid of extra copy. Coalesce headers and data frames. |
130 for (size_t i = 0; i < buffers.size(); ++i) { | 148 for (size_t i = 0; i < buffers.size(); ++i) { |
131 memcpy(pending_combined_buffer_->data() + len, buffers[i]->data(), | 149 memcpy(pending_combined_buffer_->data() + len, buffers[i]->data(), |
132 lengths[i]); | 150 lengths[i]); |
133 len += lengths[i]; | 151 len += lengths[i]; |
134 } | 152 } |
135 stream_->SendData(pending_combined_buffer_.get(), total_len, | 153 stream_->SendData(pending_combined_buffer_.get(), total_len, |
136 end_stream ? NO_MORE_DATA_TO_SEND : MORE_DATA_TO_SEND); | 154 end_stream ? NO_MORE_DATA_TO_SEND : MORE_DATA_TO_SEND); |
137 } | 155 } |
138 | 156 |
139 void BidirectionalStreamSpdyImpl::Cancel() { | 157 void BidirectionalStreamSpdyImpl::Cancel() { |
140 if (!stream_) | 158 if (!stream_) |
141 return; | 159 return; |
142 // Cancels the stream and detaches the delegate so it doesn't get called back. | 160 // Cancels the stream and detaches the delegate so it doesn't get called back. |
143 stream_->DetachDelegate(); | 161 stream_->DetachDelegate(); |
144 DCHECK(!stream_); | 162 DCHECK(!stream_); |
kapishnikov
2016/06/03 22:46:41
nit: This DCheck will always succeed and can be re
xunjieli
2016/06/04 01:31:05
Done.
| |
163 delegate_ = nullptr; | |
kapishnikov
2016/06/03 22:46:41
Should go ahead of "if (!stream_)".
xunjieli
2016/06/04 01:31:05
Done.
| |
164 // Cancel any pending callback. | |
165 weak_factory_.InvalidateWeakPtrs(); | |
145 } | 166 } |
146 | 167 |
147 NextProto BidirectionalStreamSpdyImpl::GetProtocol() const { | 168 NextProto BidirectionalStreamSpdyImpl::GetProtocol() const { |
148 return negotiated_protocol_; | 169 return negotiated_protocol_; |
149 } | 170 } |
150 | 171 |
151 int64_t BidirectionalStreamSpdyImpl::GetTotalReceivedBytes() const { | 172 int64_t BidirectionalStreamSpdyImpl::GetTotalReceivedBytes() const { |
152 if (stream_closed_) | 173 if (stream_closed_) |
153 return closed_stream_received_bytes_; | 174 return closed_stream_received_bytes_; |
154 | 175 |
(...skipping 10 matching lines...) Expand all Loading... | |
165 if (!stream_) | 186 if (!stream_) |
166 return 0; | 187 return 0; |
167 | 188 |
168 return stream_->raw_sent_bytes(); | 189 return stream_->raw_sent_bytes(); |
169 } | 190 } |
170 | 191 |
171 void BidirectionalStreamSpdyImpl::OnRequestHeadersSent() { | 192 void BidirectionalStreamSpdyImpl::OnRequestHeadersSent() { |
172 DCHECK(stream_); | 193 DCHECK(stream_); |
173 | 194 |
174 negotiated_protocol_ = stream_->GetProtocol(); | 195 negotiated_protocol_ = stream_->GetProtocol(); |
175 delegate_->OnStreamReady(/*request_headers_sent=*/true); | 196 if (delegate_) |
197 delegate_->OnStreamReady(/*request_headers_sent=*/true); | |
176 } | 198 } |
177 | 199 |
178 SpdyResponseHeadersStatus BidirectionalStreamSpdyImpl::OnResponseHeadersUpdated( | 200 SpdyResponseHeadersStatus BidirectionalStreamSpdyImpl::OnResponseHeadersUpdated( |
179 const SpdyHeaderBlock& response_headers) { | 201 const SpdyHeaderBlock& response_headers) { |
180 DCHECK(stream_); | 202 DCHECK(stream_); |
181 | 203 |
182 delegate_->OnHeadersReceived(response_headers); | 204 if (delegate_) |
205 delegate_->OnHeadersReceived(response_headers); | |
206 | |
183 return RESPONSE_HEADERS_ARE_COMPLETE; | 207 return RESPONSE_HEADERS_ARE_COMPLETE; |
184 } | 208 } |
185 | 209 |
186 void BidirectionalStreamSpdyImpl::OnDataReceived( | 210 void BidirectionalStreamSpdyImpl::OnDataReceived( |
187 std::unique_ptr<SpdyBuffer> buffer) { | 211 std::unique_ptr<SpdyBuffer> buffer) { |
188 DCHECK(stream_); | 212 DCHECK(stream_); |
189 DCHECK(!stream_closed_); | 213 DCHECK(!stream_closed_); |
190 | 214 |
191 // If |buffer| is null, BidirectionalStreamSpdyImpl::OnClose will be invoked | 215 // If |buffer| is null, BidirectionalStreamSpdyImpl::OnClose will be invoked |
192 // by SpdyStream to indicate the end of stream. | 216 // by SpdyStream to indicate the end of stream. |
193 if (!buffer) | 217 if (!buffer) |
194 return; | 218 return; |
195 | 219 |
196 // When buffer is consumed, SpdyStream::OnReadBufferConsumed will adjust | 220 // When buffer is consumed, SpdyStream::OnReadBufferConsumed will adjust |
197 // recv window size accordingly. | 221 // recv window size accordingly. |
198 read_data_queue_.Enqueue(std::move(buffer)); | 222 read_data_queue_.Enqueue(std::move(buffer)); |
199 if (read_buffer_) { | 223 if (read_buffer_) { |
200 // Handing small chunks of data to the caller creates measurable overhead. | 224 // Handing small chunks of data to the caller creates measurable overhead. |
201 // So buffer data in short time-spans and send a single read notification. | 225 // So buffer data in short time-spans and send a single read notification. |
202 ScheduleBufferedRead(); | 226 ScheduleBufferedRead(); |
203 } | 227 } |
204 } | 228 } |
205 | 229 |
206 void BidirectionalStreamSpdyImpl::OnDataSent() { | 230 void BidirectionalStreamSpdyImpl::OnDataSent() { |
207 DCHECK(stream_); | 231 DCHECK(stream_); |
208 DCHECK(!stream_closed_); | 232 DCHECK(!stream_closed_); |
209 | 233 |
210 pending_combined_buffer_ = nullptr; | 234 pending_combined_buffer_ = nullptr; |
211 delegate_->OnDataSent(); | 235 if (delegate_) |
236 delegate_->OnDataSent(); | |
212 } | 237 } |
213 | 238 |
214 void BidirectionalStreamSpdyImpl::OnTrailers(const SpdyHeaderBlock& trailers) { | 239 void BidirectionalStreamSpdyImpl::OnTrailers(const SpdyHeaderBlock& trailers) { |
215 DCHECK(stream_); | 240 DCHECK(stream_); |
216 DCHECK(!stream_closed_); | 241 DCHECK(!stream_closed_); |
217 | 242 |
218 delegate_->OnTrailersReceived(trailers); | 243 if (delegate_) |
244 delegate_->OnTrailersReceived(trailers); | |
219 } | 245 } |
220 | 246 |
221 void BidirectionalStreamSpdyImpl::OnClose(int status) { | 247 void BidirectionalStreamSpdyImpl::OnClose(int status) { |
222 DCHECK(stream_); | 248 DCHECK(stream_); |
223 | 249 |
224 stream_closed_ = true; | 250 stream_closed_ = true; |
225 closed_stream_status_ = status; | 251 closed_stream_status_ = status; |
226 closed_stream_received_bytes_ = stream_->raw_received_bytes(); | 252 closed_stream_received_bytes_ = stream_->raw_received_bytes(); |
227 closed_stream_sent_bytes_ = stream_->raw_sent_bytes(); | 253 closed_stream_sent_bytes_ = stream_->raw_sent_bytes(); |
228 stream_.reset(); | |
229 | 254 |
230 if (status != OK) { | 255 if (status != OK) { |
231 delegate_->OnFailed(status); | 256 NotifyError(status); |
232 return; | 257 return; |
233 } | 258 } |
259 stream_.reset(); | |
234 // Complete any remaining read, as all data has been buffered. | 260 // Complete any remaining read, as all data has been buffered. |
235 // If user has not called ReadData (i.e |read_buffer_| is nullptr), this will | 261 // If user has not called ReadData (i.e |read_buffer_| is nullptr), this will |
236 // do nothing. | 262 // do nothing. |
237 timer_->Stop(); | 263 timer_->Stop(); |
238 DoBufferedRead(); | 264 DoBufferedRead(); |
239 } | 265 } |
240 | 266 |
241 int BidirectionalStreamSpdyImpl::SendRequestHeadersHelper() { | 267 int BidirectionalStreamSpdyImpl::SendRequestHeadersHelper() { |
242 std::unique_ptr<SpdyHeaderBlock> headers(new SpdyHeaderBlock); | 268 std::unique_ptr<SpdyHeaderBlock> headers(new SpdyHeaderBlock); |
243 HttpRequestInfo http_request_info; | 269 HttpRequestInfo http_request_info; |
(...skipping 16 matching lines...) Expand all Loading... | |
260 stream_ = stream_request_.ReleaseStream(); | 286 stream_ = stream_request_.ReleaseStream(); |
261 stream_->SetDelegate(this); | 287 stream_->SetDelegate(this); |
262 rv = SendRequestHeadersHelper(); | 288 rv = SendRequestHeadersHelper(); |
263 if (rv == OK) { | 289 if (rv == OK) { |
264 OnRequestHeadersSent(); | 290 OnRequestHeadersSent(); |
265 return; | 291 return; |
266 } else if (rv == ERR_IO_PENDING) { | 292 } else if (rv == ERR_IO_PENDING) { |
267 return; | 293 return; |
268 } | 294 } |
269 } | 295 } |
270 delegate_->OnFailed(rv); | 296 NotifyError(rv); |
297 } | |
298 | |
299 void BidirectionalStreamSpdyImpl::NotifyError(int rv) { | |
kapishnikov
2016/06/03 22:46:41
Should we call "stream_->DetachDelegate()" here? T
xunjieli
2016/06/04 01:31:05
Done, I added a if-conditional though. Since Detac
| |
300 stream_.reset(); | |
301 if (!delegate_) | |
302 return; | |
303 BidirectionalStreamImpl::Delegate* delegate = delegate_; | |
304 delegate_ = nullptr; | |
305 // Cancel any pending callback. | |
306 weak_factory_.InvalidateWeakPtrs(); | |
307 delegate->OnFailed(rv); | |
271 } | 308 } |
272 | 309 |
273 void BidirectionalStreamSpdyImpl::ScheduleBufferedRead() { | 310 void BidirectionalStreamSpdyImpl::ScheduleBufferedRead() { |
274 // If there is already a scheduled DoBufferedRead, don't issue | 311 // If there is already a scheduled DoBufferedRead, don't issue |
275 // another one. Mark that we have received more data and return. | 312 // another one. Mark that we have received more data and return. |
276 if (timer_->IsRunning()) { | 313 if (timer_->IsRunning()) { |
277 more_read_data_pending_ = true; | 314 more_read_data_pending_ = true; |
278 return; | 315 return; |
279 } | 316 } |
280 | 317 |
(...skipping 15 matching lines...) Expand all Loading... | |
296 ScheduleBufferedRead(); | 333 ScheduleBufferedRead(); |
297 return; | 334 return; |
298 } | 335 } |
299 | 336 |
300 int rv = 0; | 337 int rv = 0; |
301 if (read_buffer_) { | 338 if (read_buffer_) { |
302 rv = ReadData(read_buffer_.get(), read_buffer_len_); | 339 rv = ReadData(read_buffer_.get(), read_buffer_len_); |
303 DCHECK_NE(ERR_IO_PENDING, rv); | 340 DCHECK_NE(ERR_IO_PENDING, rv); |
304 read_buffer_ = nullptr; | 341 read_buffer_ = nullptr; |
305 read_buffer_len_ = 0; | 342 read_buffer_len_ = 0; |
306 delegate_->OnDataRead(rv); | 343 if (delegate_) |
344 delegate_->OnDataRead(rv); | |
307 } | 345 } |
308 } | 346 } |
309 | 347 |
310 bool BidirectionalStreamSpdyImpl::ShouldWaitForMoreBufferedData() const { | 348 bool BidirectionalStreamSpdyImpl::ShouldWaitForMoreBufferedData() const { |
311 if (stream_closed_) | 349 if (stream_closed_) |
312 return false; | 350 return false; |
313 DCHECK_GT(read_buffer_len_, 0); | 351 DCHECK_GT(read_buffer_len_, 0); |
314 return read_data_queue_.GetTotalSize() < | 352 return read_data_queue_.GetTotalSize() < |
315 static_cast<size_t>(read_buffer_len_); | 353 static_cast<size_t>(read_buffer_len_); |
316 } | 354 } |
317 | 355 |
318 } // namespace net | 356 } // namespace net |
OLD | NEW |