| 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_); |
| 163 delegate_ = nullptr; |
| 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) { |
| 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 |