| OLD | NEW |
|---|
| (Empty) |
| 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 | |
| 3 // found in the LICENSE file. | |
| 4 | |
| 5 #include "net/spdy/bidirectional_stream_spdy_job.h" | |
| 6 | |
| 7 #include "base/bind.h" | |
| 8 #include "base/location.h" | |
| 9 #include "base/logging.h" | |
| 10 #include "base/time/time.h" | |
| 11 #include "base/timer/timer.h" | |
| 12 #include "net/http/bidirectional_stream_request_info.h" | |
| 13 #include "net/spdy/spdy_buffer.h" | |
| 14 #include "net/spdy/spdy_header_block.h" | |
| 15 #include "net/spdy/spdy_http_utils.h" | |
| 16 #include "net/spdy/spdy_stream.h" | |
| 17 | |
| 18 namespace net { | |
| 19 | |
| 20 namespace { | |
| 21 | |
| 22 // Time to wait in millisecond to notify |delegate_| of data received. | |
| 23 // Handing small chunks of data to the caller creates measurable overhead. | |
| 24 // So buffer data in short time-spans and send a single read notification. | |
| 25 const int kBufferTimeMs = 1; | |
| 26 | |
| 27 } // namespace | |
| 28 | |
| 29 BidirectionalStreamSpdyJob::BidirectionalStreamSpdyJob( | |
| 30 const base::WeakPtr<SpdySession>& spdy_session) | |
| 31 : spdy_session_(spdy_session), | |
| 32 request_info_(nullptr), | |
| 33 delegate_(nullptr), | |
| 34 negotiated_protocol_(kProtoUnknown), | |
| 35 more_read_data_pending_(false), | |
| 36 read_buffer_len_(0), | |
| 37 stream_closed_(false), | |
| 38 closed_stream_status_(ERR_FAILED), | |
| 39 closed_stream_received_bytes_(0), | |
| 40 closed_stream_sent_bytes_(0), | |
| 41 weak_factory_(this) {} | |
| 42 | |
| 43 BidirectionalStreamSpdyJob::~BidirectionalStreamSpdyJob() { | |
| 44 if (stream_) { | |
| 45 stream_->DetachDelegate(); | |
| 46 DCHECK(!stream_); | |
| 47 } | |
| 48 } | |
| 49 | |
| 50 void BidirectionalStreamSpdyJob::Start( | |
| 51 const BidirectionalStreamRequestInfo* request_info, | |
| 52 const BoundNetLog& net_log, | |
| 53 BidirectionalStreamJob::Delegate* delegate, | |
| 54 scoped_ptr<base::Timer> timer) { | |
| 55 DCHECK(!stream_); | |
| 56 DCHECK(timer); | |
| 57 | |
| 58 delegate_ = delegate; | |
| 59 timer_ = std::move(timer); | |
| 60 | |
| 61 if (!spdy_session_) { | |
| 62 delegate_->OnFailed(ERR_CONNECTION_CLOSED); | |
| 63 return; | |
| 64 } | |
| 65 | |
| 66 request_info_ = request_info; | |
| 67 | |
| 68 int rv = stream_request_.StartRequest( | |
| 69 SPDY_BIDIRECTIONAL_STREAM, spdy_session_, request_info_->url, | |
| 70 request_info_->priority, net_log, | |
| 71 base::Bind(&BidirectionalStreamSpdyJob::OnStreamInitialized, | |
| 72 weak_factory_.GetWeakPtr())); | |
| 73 if (rv != ERR_IO_PENDING) | |
| 74 OnStreamInitialized(rv); | |
| 75 } | |
| 76 | |
| 77 int BidirectionalStreamSpdyJob::ReadData(IOBuffer* buf, int buf_len) { | |
| 78 if (stream_) | |
| 79 DCHECK(!stream_->IsIdle()); | |
| 80 | |
| 81 DCHECK(buf); | |
| 82 DCHECK(buf_len); | |
| 83 DCHECK(!timer_->IsRunning()) << "There should be only one ReadData in flight"; | |
| 84 | |
| 85 // If there is data buffered, complete the IO immediately. | |
| 86 if (!read_data_queue_.IsEmpty()) { | |
| 87 return read_data_queue_.Dequeue(buf->data(), buf_len); | |
| 88 } else if (stream_closed_) { | |
| 89 return closed_stream_status_; | |
| 90 } | |
| 91 // Read will complete asynchronously and Delegate::OnReadCompleted will be | |
| 92 // called upon completion. | |
| 93 read_buffer_ = buf; | |
| 94 read_buffer_len_ = buf_len; | |
| 95 return ERR_IO_PENDING; | |
| 96 } | |
| 97 | |
| 98 void BidirectionalStreamSpdyJob::SendData(IOBuffer* data, | |
| 99 int length, | |
| 100 bool end_stream) { | |
| 101 DCHECK(!stream_closed_); | |
| 102 DCHECK(stream_); | |
| 103 | |
| 104 stream_->SendData(data, length, | |
| 105 end_stream ? NO_MORE_DATA_TO_SEND : MORE_DATA_TO_SEND); | |
| 106 } | |
| 107 | |
| 108 void BidirectionalStreamSpdyJob::Cancel() { | |
| 109 if (!stream_) | |
| 110 return; | |
| 111 // Cancels the stream and detaches the delegate so it doesn't get called back. | |
| 112 stream_->DetachDelegate(); | |
| 113 DCHECK(!stream_); | |
| 114 } | |
| 115 | |
| 116 NextProto BidirectionalStreamSpdyJob::GetProtocol() const { | |
| 117 return negotiated_protocol_; | |
| 118 } | |
| 119 | |
| 120 int64_t BidirectionalStreamSpdyJob::GetTotalReceivedBytes() const { | |
| 121 if (stream_closed_) | |
| 122 return closed_stream_received_bytes_; | |
| 123 | |
| 124 if (!stream_) | |
| 125 return 0; | |
| 126 | |
| 127 return stream_->raw_received_bytes(); | |
| 128 } | |
| 129 | |
| 130 int64_t BidirectionalStreamSpdyJob::GetTotalSentBytes() const { | |
| 131 if (stream_closed_) | |
| 132 return closed_stream_sent_bytes_; | |
| 133 | |
| 134 if (!stream_) | |
| 135 return 0; | |
| 136 | |
| 137 return stream_->raw_sent_bytes(); | |
| 138 } | |
| 139 | |
| 140 void BidirectionalStreamSpdyJob::OnRequestHeadersSent() { | |
| 141 DCHECK(stream_); | |
| 142 | |
| 143 negotiated_protocol_ = stream_->GetProtocol(); | |
| 144 delegate_->OnHeadersSent(); | |
| 145 } | |
| 146 | |
| 147 SpdyResponseHeadersStatus BidirectionalStreamSpdyJob::OnResponseHeadersUpdated( | |
| 148 const SpdyHeaderBlock& response_headers) { | |
| 149 DCHECK(stream_); | |
| 150 | |
| 151 delegate_->OnHeadersReceived(response_headers); | |
| 152 return RESPONSE_HEADERS_ARE_COMPLETE; | |
| 153 } | |
| 154 | |
| 155 void BidirectionalStreamSpdyJob::OnDataReceived(scoped_ptr<SpdyBuffer> buffer) { | |
| 156 DCHECK(stream_); | |
| 157 DCHECK(!stream_closed_); | |
| 158 | |
| 159 // If |buffer| is null, BidirectionalStreamSpdyJob::OnClose will be invoked by | |
| 160 // SpdyStream to indicate the end of stream. | |
| 161 if (!buffer) | |
| 162 return; | |
| 163 | |
| 164 // When buffer is consumed, SpdyStream::OnReadBufferConsumed will adjust | |
| 165 // recv window size accordingly. | |
| 166 read_data_queue_.Enqueue(std::move(buffer)); | |
| 167 if (read_buffer_) { | |
| 168 // Handing small chunks of data to the caller creates measurable overhead. | |
| 169 // So buffer data in short time-spans and send a single read notification. | |
| 170 ScheduleBufferedRead(); | |
| 171 } | |
| 172 } | |
| 173 | |
| 174 void BidirectionalStreamSpdyJob::OnDataSent() { | |
| 175 DCHECK(stream_); | |
| 176 DCHECK(!stream_closed_); | |
| 177 | |
| 178 delegate_->OnDataSent(); | |
| 179 } | |
| 180 | |
| 181 void BidirectionalStreamSpdyJob::OnTrailers(const SpdyHeaderBlock& trailers) { | |
| 182 DCHECK(stream_); | |
| 183 DCHECK(!stream_closed_); | |
| 184 | |
| 185 delegate_->OnTrailersReceived(trailers); | |
| 186 } | |
| 187 | |
| 188 void BidirectionalStreamSpdyJob::OnClose(int status) { | |
| 189 DCHECK(stream_); | |
| 190 | |
| 191 stream_closed_ = true; | |
| 192 closed_stream_status_ = status; | |
| 193 closed_stream_received_bytes_ = stream_->raw_received_bytes(); | |
| 194 closed_stream_sent_bytes_ = stream_->raw_sent_bytes(); | |
| 195 stream_.reset(); | |
| 196 | |
| 197 if (status != OK) { | |
| 198 delegate_->OnFailed(status); | |
| 199 return; | |
| 200 } | |
| 201 // Complete any remaining read, as all data has been buffered. | |
| 202 // If user has not called ReadData (i.e |read_buffer_| is nullptr), this will | |
| 203 // do nothing. | |
| 204 timer_->Stop(); | |
| 205 DoBufferedRead(); | |
| 206 } | |
| 207 | |
| 208 void BidirectionalStreamSpdyJob::SendRequestHeaders() { | |
| 209 scoped_ptr<SpdyHeaderBlock> headers(new SpdyHeaderBlock); | |
| 210 HttpRequestInfo http_request_info; | |
| 211 http_request_info.url = request_info_->url; | |
| 212 http_request_info.method = request_info_->method; | |
| 213 http_request_info.extra_headers = request_info_->extra_headers; | |
| 214 | |
| 215 CreateSpdyHeadersFromHttpRequest( | |
| 216 http_request_info, http_request_info.extra_headers, | |
| 217 stream_->GetProtocolVersion(), true, headers.get()); | |
| 218 stream_->SendRequestHeaders(std::move(headers), | |
| 219 request_info_->end_stream_on_headers | |
| 220 ? NO_MORE_DATA_TO_SEND | |
| 221 : MORE_DATA_TO_SEND); | |
| 222 } | |
| 223 | |
| 224 void BidirectionalStreamSpdyJob::OnStreamInitialized(int rv) { | |
| 225 DCHECK_NE(ERR_IO_PENDING, rv); | |
| 226 if (rv == OK) { | |
| 227 stream_ = stream_request_.ReleaseStream(); | |
| 228 stream_->SetDelegate(this); | |
| 229 SendRequestHeaders(); | |
| 230 return; | |
| 231 } | |
| 232 delegate_->OnFailed(rv); | |
| 233 } | |
| 234 | |
| 235 void BidirectionalStreamSpdyJob::ScheduleBufferedRead() { | |
| 236 // If there is already a scheduled DoBufferedRead, don't issue | |
| 237 // another one. Mark that we have received more data and return. | |
| 238 if (timer_->IsRunning()) { | |
| 239 more_read_data_pending_ = true; | |
| 240 return; | |
| 241 } | |
| 242 | |
| 243 more_read_data_pending_ = false; | |
| 244 timer_->Start(FROM_HERE, base::TimeDelta::FromMilliseconds(kBufferTimeMs), | |
| 245 base::Bind(&BidirectionalStreamSpdyJob::DoBufferedRead, | |
| 246 weak_factory_.GetWeakPtr())); | |
| 247 } | |
| 248 | |
| 249 void BidirectionalStreamSpdyJob::DoBufferedRead() { | |
| 250 DCHECK(!timer_->IsRunning()); | |
| 251 // Check to see that the stream has not errored out. | |
| 252 DCHECK(stream_ || stream_closed_); | |
| 253 DCHECK(!stream_closed_ || closed_stream_status_ == OK); | |
| 254 | |
| 255 // When |more_read_data_pending_| is true, it means that more data has arrived | |
| 256 // since started waiting. Wait a little longer and continue to buffer. | |
| 257 if (more_read_data_pending_ && ShouldWaitForMoreBufferedData()) { | |
| 258 ScheduleBufferedRead(); | |
| 259 return; | |
| 260 } | |
| 261 | |
| 262 int rv = 0; | |
| 263 if (read_buffer_) { | |
| 264 rv = ReadData(read_buffer_.get(), read_buffer_len_); | |
| 265 DCHECK_NE(ERR_IO_PENDING, rv); | |
| 266 read_buffer_ = nullptr; | |
| 267 read_buffer_len_ = 0; | |
| 268 delegate_->OnDataRead(rv); | |
| 269 } | |
| 270 } | |
| 271 | |
| 272 bool BidirectionalStreamSpdyJob::ShouldWaitForMoreBufferedData() const { | |
| 273 if (stream_closed_) | |
| 274 return false; | |
| 275 DCHECK_GT(read_buffer_len_, 0); | |
| 276 return read_data_queue_.GetTotalSize() < | |
| 277 static_cast<size_t>(read_buffer_len_); | |
| 278 } | |
| 279 | |
| 280 } // namespace net | |
| OLD | NEW |
|---|
Chromium Code Reviews
Issue 1812823010:
Rename net::BidirectionalStream*Job to net::BidirectionalStream*Impl (Closed)
Base URL: https://chromium.googlesource.com/chromium/src.git@master