net/spdy/spdy_http_stream.cc

   1 // Copyright (c) 2012 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/spdy_http_stream.h"
   6
   7 #include <algorithm>
   8 #include <list>
   9
  10 #include "base/bind.h"
  11 #include "base/location.h"
  12 #include "base/logging.h"
  13 #include "base/single_thread_task_runner.h"
  14 #include "base/strings/stringprintf.h"
  15 #include "base/thread_task_runner_handle.h"
  16 #include "base/values.h"
  17 #include "net/base/host_port_pair.h"
  18 #include "net/base/net_util.h"
  19 #include "net/base/upload_data_stream.h"
  20 #include "net/http/http_request_headers.h"
  21 #include "net/http/http_request_info.h"
  22 #include "net/http/http_response_info.h"
  23 #include "net/log/net_log.h"
  24 #include "net/spdy/spdy_header_block.h"
  25 #include "net/spdy/spdy_http_utils.h"
  26 #include "net/spdy/spdy_protocol.h"
  27 #include "net/spdy/spdy_session.h"
  28
  29 namespace net {
  30
  31 SpdyHttpStream::SpdyHttpStream(const base::WeakPtr<SpdySession>& spdy_session,
  32                                bool direct)
  33     : spdy_session_(spdy_session),
  34       is_reused_(spdy_session_->IsReused()),
  35       stream_closed_(false),
  36       closed_stream_status_(ERR_FAILED),
  37       closed_stream_id_(0),
  38       closed_stream_received_bytes_(0),
  39       request_info_(NULL),
  40       response_info_(NULL),
  41       response_headers_status_(RESPONSE_HEADERS_ARE_INCOMPLETE),
  42       user_buffer_len_(0),
  43       request_body_buf_size_(0),
  44       buffered_read_callback_pending_(false),
  45       more_read_data_pending_(false),
  46       direct_(direct),
  47       weak_factory_(this) {
  48   DCHECK(spdy_session_.get());
  49 }
  50
  51 SpdyHttpStream::~SpdyHttpStream() {
  52   if (stream_.get()) {
  53     stream_->DetachDelegate();
  54     DCHECK(!stream_.get());
  55   }
  56 }
  57
  58 int SpdyHttpStream::InitializeStream(const HttpRequestInfo* request_info,
  59                                      RequestPriority priority,
  60                                      const BoundNetLog& stream_net_log,
  61                                      const CompletionCallback& callback) {
  62   DCHECK(!stream_);
  63   if (!spdy_session_)
  64     return ERR_CONNECTION_CLOSED;
  65
  66   request_info_ = request_info;
  67   if (request_info_->method == "GET") {
  68     int error = spdy_session_->GetPushStream(request_info_->url, &stream_,
  69                                              stream_net_log);
  70     if (error != OK)
  71       return error;
  72
  73     // |stream_| may be NULL even if OK was returned.
  74     if (stream_.get()) {
  75       DCHECK_EQ(stream_->type(), SPDY_PUSH_STREAM);
  76       stream_->SetDelegate(this);
  77       return OK;
  78     }
  79   }
  80
  81   int rv = stream_request_.StartRequest(
  82       SPDY_REQUEST_RESPONSE_STREAM, spdy_session_, request_info_->url,
  83       priority, stream_net_log,
  84       base::Bind(&SpdyHttpStream::OnStreamCreated,
  85                  weak_factory_.GetWeakPtr(), callback));
  86
  87   if (rv == OK) {
  88     stream_ = stream_request_.ReleaseStream();
  89     stream_->SetDelegate(this);
  90   }
  91
  92   return rv;
  93 }
  94
  95 UploadProgress SpdyHttpStream::GetUploadProgress() const {
  96   if (!request_info_ || !HasUploadData())
  97     return UploadProgress();
  98
  99   return UploadProgress(request_info_->upload_data_stream->position(),
 100                         request_info_->upload_data_stream->size());
 101 }
 102
 103 int SpdyHttpStream::ReadResponseHeaders(const CompletionCallback& callback) {
 104   CHECK(!callback.is_null());
 105   if (stream_closed_)
 106     return closed_stream_status_;
 107
 108   CHECK(stream_.get());
 109
 110   // Check if we already have the response headers. If so, return synchronously.
 111   if (response_headers_status_ == RESPONSE_HEADERS_ARE_COMPLETE) {
 112     CHECK(!stream_->IsIdle());
 113     return OK;
 114   }
 115
 116   // Still waiting for the response, return IO_PENDING.
 117   CHECK(callback_.is_null());
 118   callback_ = callback;
 119   return ERR_IO_PENDING;
 120 }
 121
 122 int SpdyHttpStream::ReadResponseBody(
 123     IOBuffer* buf, int buf_len, const CompletionCallback& callback) {
 124   if (stream_.get())
 125     CHECK(!stream_->IsIdle());
 126
 127   CHECK(buf);
 128   CHECK(buf_len);
 129   CHECK(!callback.is_null());
 130
 131   // If we have data buffered, complete the IO immediately.
 132   if (!response_body_queue_.IsEmpty()) {
 133     return response_body_queue_.Dequeue(buf->data(), buf_len);
 134   } else if (stream_closed_) {
 135     return closed_stream_status_;
 136   }
 137
 138   CHECK(callback_.is_null());
 139   CHECK(!user_buffer_.get());
 140   CHECK_EQ(0, user_buffer_len_);
 141
 142   callback_ = callback;
 143   user_buffer_ = buf;
 144   user_buffer_len_ = buf_len;
 145   return ERR_IO_PENDING;
 146 }
 147
 148 void SpdyHttpStream::Close(bool not_reusable) {
 149   // Note: the not_reusable flag has no meaning for SPDY streams.
 150
 151   Cancel();
 152   DCHECK(!stream_.get());
 153 }
 154
 155 HttpStream* SpdyHttpStream::RenewStreamForAuth() {
 156   return NULL;
 157 }
 158
 159 bool SpdyHttpStream::IsResponseBodyComplete() const {
 160   return stream_closed_;
 161 }
 162
 163 bool SpdyHttpStream::CanFindEndOfResponse() const {
 164   return true;
 165 }
 166
 167 bool SpdyHttpStream::IsConnectionReused() const {
 168   return is_reused_;
 169 }
 170
 171 void SpdyHttpStream::SetConnectionReused() {
 172   // SPDY doesn't need an indicator here.
 173 }
 174
 175 bool SpdyHttpStream::IsConnectionReusable() const {
 176   // SPDY streams aren't considered reusable.
 177   return false;
 178 }
 179
 180 int64 SpdyHttpStream::GetTotalReceivedBytes() const {
 181   if (stream_closed_)
 182     return closed_stream_received_bytes_;
 183
 184   if (!stream_)
 185     return 0;
 186
 187   return stream_->raw_received_bytes();
 188 }
 189
 190 bool SpdyHttpStream::GetLoadTimingInfo(LoadTimingInfo* load_timing_info) const {
 191   if (stream_closed_) {
 192     if (!closed_stream_has_load_timing_info_)
 193       return false;
 194     *load_timing_info = closed_stream_load_timing_info_;
 195     return true;
 196   }
 197
 198   // If |stream_| has yet to be created, or does not yet have an ID, fail.
 199   // The reused flag can only be correctly set once a stream has an ID.  Streams
 200   // get their IDs once the request has been successfully sent, so this does not
 201   // behave that differently from other stream types.
 202   if (!stream_ || stream_->stream_id() == 0)
 203     return false;
 204
 205   return stream_->GetLoadTimingInfo(load_timing_info);
 206 }
 207
 208 int SpdyHttpStream::SendRequest(const HttpRequestHeaders& request_headers,
 209                                 HttpResponseInfo* response,
 210                                 const CompletionCallback& callback) {
 211   if (stream_closed_) {
 212     return closed_stream_status_;
 213   }
 214
 215   base::Time request_time = base::Time::Now();
 216   CHECK(stream_.get());
 217
 218   stream_->SetRequestTime(request_time);
 219   // This should only get called in the case of a request occurring
 220   // during server push that has already begun but hasn't finished,
 221   // so we set the response's request time to be the actual one
 222   if (response_info_)
 223     response_info_->request_time = request_time;
 224
 225   CHECK(!request_body_buf_.get());
 226   if (HasUploadData()) {
 227     // Use kMaxSpdyFrameChunkSize as the buffer size, since the request
 228     // body data is written with this size at a time.
 229     request_body_buf_ = new IOBufferWithSize(kMaxSpdyFrameChunkSize);
 230     // The request body buffer is empty at first.
 231     request_body_buf_size_ = 0;
 232   }
 233
 234   CHECK(!callback.is_null());
 235   CHECK(response);
 236
 237   // SendRequest can be called in two cases.
 238   //
 239   // a) A client initiated request. In this case, |response_info_| should be
 240   //    NULL to start with.
 241   // b) A client request which matches a response that the server has already
 242   //    pushed.
 243   if (push_response_info_.get()) {
 244     *response = *(push_response_info_.get());
 245     push_response_info_.reset();
 246   } else {
 247     DCHECK_EQ(static_cast<HttpResponseInfo*>(NULL), response_info_);
 248   }
 249
 250   response_info_ = response;
 251
 252   // Put the peer's IP address and port into the response.
 253   IPEndPoint address;
 254   int result = stream_->GetPeerAddress(&address);
 255   if (result != OK)
 256     return result;
 257   response_info_->socket_address = HostPortPair::FromIPEndPoint(address);
 258
 259   if (stream_->type() == SPDY_PUSH_STREAM) {
 260     // Pushed streams do not send any data, and should always be
 261     // idle. However, we still want to return ERR_IO_PENDING to mimic
 262     // non-push behavior. The callback will be called when the
 263     // response is received.
 264     result = ERR_IO_PENDING;
 265   } else {
 266     scoped_ptr<SpdyHeaderBlock> headers(new SpdyHeaderBlock);
 267     CreateSpdyHeadersFromHttpRequest(
 268         *request_info_, request_headers,
 269         stream_->GetProtocolVersion(), direct_,
 270         headers.get());
 271     stream_->net_log().AddEvent(
 272         NetLog::TYPE_HTTP_TRANSACTION_HTTP2_SEND_REQUEST_HEADERS,
 273         base::Bind(&SpdyHeaderBlockNetLogCallback, headers.get()));
 274     result =
 275         stream_->SendRequestHeaders(
 276             headers.Pass(),
 277             HasUploadData() ? MORE_DATA_TO_SEND : NO_MORE_DATA_TO_SEND);
 278   }
 279
 280   if (result == ERR_IO_PENDING) {
 281     CHECK(callback_.is_null());
 282     callback_ = callback;
 283   }
 284   return result;
 285 }
 286
 287 void SpdyHttpStream::Cancel() {
 288   callback_.Reset();
 289   if (stream_.get()) {
 290     stream_->Cancel();
 291     DCHECK(!stream_.get());
 292   }
 293 }
 294
 295 void SpdyHttpStream::OnRequestHeadersSent() {
 296   if (!callback_.is_null())
 297     DoCallback(OK);
 298
 299   // TODO(akalin): Do this immediately after sending the request
 300   // headers.
 301   if (HasUploadData())
 302     ReadAndSendRequestBodyData();
 303 }
 304
 305 SpdyResponseHeadersStatus SpdyHttpStream::OnResponseHeadersUpdated(
 306     const SpdyHeaderBlock& response_headers) {
 307   CHECK_EQ(response_headers_status_, RESPONSE_HEADERS_ARE_INCOMPLETE);
 308
 309   if (!response_info_) {
 310     DCHECK_EQ(stream_->type(), SPDY_PUSH_STREAM);
 311     push_response_info_.reset(new HttpResponseInfo);
 312     response_info_ = push_response_info_.get();
 313   }
 314
 315   if (!SpdyHeadersToHttpResponse(
 316           response_headers, stream_->GetProtocolVersion(), response_info_)) {
 317     // We do not have complete headers yet.
 318     return RESPONSE_HEADERS_ARE_INCOMPLETE;
 319   }
 320
 321   response_info_->response_time = stream_->response_time();
 322   response_headers_status_ = RESPONSE_HEADERS_ARE_COMPLETE;
 323   // Don't store the SSLInfo in the response here, HttpNetworkTransaction
 324   // will take care of that part.
 325   SSLInfo ssl_info;
 326   NextProto protocol_negotiated = kProtoUnknown;
 327   stream_->GetSSLInfo(&ssl_info,
 328                       &response_info_->was_npn_negotiated,
 329                       &protocol_negotiated);
 330   response_info_->npn_negotiated_protocol =
 331       SSLClientSocket::NextProtoToString(protocol_negotiated);
 332   response_info_->request_time = stream_->GetRequestTime();
 333   response_info_->connection_info =
 334       HttpResponseInfo::ConnectionInfoFromNextProto(stream_->GetProtocol());
 335   response_info_->vary_data
 336       .Init(*request_info_, *response_info_->headers.get());
 337
 338   if (!callback_.is_null())
 339     DoCallback(OK);
 340
 341   return RESPONSE_HEADERS_ARE_COMPLETE;
 342 }
 343
 344 void SpdyHttpStream::OnDataReceived(scoped_ptr<SpdyBuffer> buffer) {
 345   CHECK_EQ(response_headers_status_, RESPONSE_HEADERS_ARE_COMPLETE);
 346
 347   // Note that data may be received for a SpdyStream prior to the user calling
 348   // ReadResponseBody(), therefore user_buffer_ may be NULL.  This may often
 349   // happen for server initiated streams.
 350   DCHECK(stream_.get());
 351   DCHECK(!stream_->IsClosed() || stream_->type() == SPDY_PUSH_STREAM);
 352   if (buffer) {
 353     response_body_queue_.Enqueue(buffer.Pass());
 354
 355     if (user_buffer_.get()) {
 356       // Handing small chunks of data to the caller creates measurable overhead.
 357       // We buffer data in short time-spans and send a single read notification.
 358       ScheduleBufferedReadCallback();
 359     }
 360   }
 361 }
 362
 363 void SpdyHttpStream::OnDataSent() {
 364   request_body_buf_size_ = 0;
 365   ReadAndSendRequestBodyData();
 366 }
 367
 368 // TODO(xunjieli): Maybe do something with the trailers. crbug.com/422958.
 369 void SpdyHttpStream::OnTrailers(const SpdyHeaderBlock& trailers) {}
 370
 371 void SpdyHttpStream::OnClose(int status) {
 372   // Cancel any pending reads from the upload data stream.
 373   if (request_info_->upload_data_stream)
 374     request_info_->upload_data_stream->Reset();
 375
 376   if (stream_.get()) {
 377     stream_closed_ = true;
 378     closed_stream_status_ = status;
 379     closed_stream_id_ = stream_->stream_id();
 380     closed_stream_has_load_timing_info_ =
 381         stream_->GetLoadTimingInfo(&closed_stream_load_timing_info_);
 382     closed_stream_received_bytes_ = stream_->raw_received_bytes();
 383   }
 384   stream_.reset();
 385
 386   bool invoked_callback = false;
 387   if (status == OK) {
 388     // We need to complete any pending buffered read now.
 389     invoked_callback = DoBufferedReadCallback();
 390   }
 391   if (!invoked_callback && !callback_.is_null())
 392     DoCallback(status);
 393 }
 394
 395 bool SpdyHttpStream::HasUploadData() const {
 396   CHECK(request_info_);
 397   return
 398       request_info_->upload_data_stream &&
 399       ((request_info_->upload_data_stream->size() > 0) ||
 400        request_info_->upload_data_stream->is_chunked());
 401 }
 402
 403 void SpdyHttpStream::OnStreamCreated(
 404     const CompletionCallback& callback,
 405     int rv) {
 406   if (rv == OK) {
 407     stream_ = stream_request_.ReleaseStream();
 408     stream_->SetDelegate(this);
 409   }
 410   callback.Run(rv);
 411 }
 412
 413 void SpdyHttpStream::ReadAndSendRequestBodyData() {
 414   CHECK(HasUploadData());
 415   CHECK_EQ(request_body_buf_size_, 0);
 416
 417   if (request_info_->upload_data_stream->IsEOF())
 418     return;
 419
 420   // Read the data from the request body stream.
 421   const int rv = request_info_->upload_data_stream
 422       ->Read(request_body_buf_.get(),
 423              request_body_buf_->size(),
 424              base::Bind(&SpdyHttpStream::OnRequestBodyReadCompleted,
 425                         weak_factory_.GetWeakPtr()));
 426
 427   if (rv != ERR_IO_PENDING) {
 428     // ERR_IO_PENDING is the only possible error.
 429     CHECK_GE(rv, 0);
 430     OnRequestBodyReadCompleted(rv);
 431   }
 432 }
 433
 434 void SpdyHttpStream::OnRequestBodyReadCompleted(int status) {
 435   CHECK_GE(status, 0);
 436   request_body_buf_size_ = status;
 437   const bool eof = request_info_->upload_data_stream->IsEOF();
 438   // Only the final fame may have a length of 0.
 439   if (eof) {
 440     CHECK_GE(request_body_buf_size_, 0);
 441   } else {
 442     CHECK_GT(request_body_buf_size_, 0);
 443   }
 444   stream_->SendData(request_body_buf_.get(),
 445                     request_body_buf_size_,
 446                     eof ? NO_MORE_DATA_TO_SEND : MORE_DATA_TO_SEND);
 447 }
 448
 449 void SpdyHttpStream::ScheduleBufferedReadCallback() {
 450   // If there is already a scheduled DoBufferedReadCallback, don't issue
 451   // another one.  Mark that we have received more data and return.
 452   if (buffered_read_callback_pending_) {
 453     more_read_data_pending_ = true;
 454     return;
 455   }
 456
 457   more_read_data_pending_ = false;
 458   buffered_read_callback_pending_ = true;
 459   const base::TimeDelta kBufferTime = base::TimeDelta::FromMilliseconds(1);
 460   base::ThreadTaskRunnerHandle::Get()->PostDelayedTask(
 461       FROM_HERE,
 462       base::Bind(base::IgnoreResult(&SpdyHttpStream::DoBufferedReadCallback),
 463                  weak_factory_.GetWeakPtr()),
 464       kBufferTime);
 465 }
 466
 467 // Checks to see if we should wait for more buffered data before notifying
 468 // the caller.  Returns true if we should wait, false otherwise.
 469 bool SpdyHttpStream::ShouldWaitForMoreBufferedData() const {
 470   // If the response is complete, there is no point in waiting.
 471   if (stream_closed_)
 472     return false;
 473
 474   DCHECK_GT(user_buffer_len_, 0);
 475   return response_body_queue_.GetTotalSize() <
 476       static_cast<size_t>(user_buffer_len_);
 477 }
 478
 479 bool SpdyHttpStream::DoBufferedReadCallback() {
 480   buffered_read_callback_pending_ = false;
 481
 482   // If the transaction is cancelled or errored out, we don't need to complete
 483   // the read.
 484   if (!stream_.get() && !stream_closed_)
 485     return false;
 486
 487   int stream_status =
 488       stream_closed_ ? closed_stream_status_ : stream_->response_status();
 489   if (stream_status != OK)
 490     return false;
 491
 492   // When more_read_data_pending_ is true, it means that more data has
 493   // arrived since we started waiting.  Wait a little longer and continue
 494   // to buffer.
 495   if (more_read_data_pending_ && ShouldWaitForMoreBufferedData()) {
 496     ScheduleBufferedReadCallback();
 497     return false;
 498   }
 499
 500   int rv = 0;
 501   if (user_buffer_.get()) {
 502     rv = ReadResponseBody(user_buffer_.get(), user_buffer_len_, callback_);
 503     CHECK_NE(rv, ERR_IO_PENDING);
 504     user_buffer_ = NULL;
 505     user_buffer_len_ = 0;
 506     DoCallback(rv);
 507     return true;
 508   }
 509   return false;
 510 }
 511
 512 void SpdyHttpStream::DoCallback(int rv) {
 513   CHECK_NE(rv, ERR_IO_PENDING);
 514   CHECK(!callback_.is_null());
 515
 516   // Since Run may result in being called back, clear user_callback_ in advance.
 517   CompletionCallback c = callback_;
 518   callback_.Reset();
 519   c.Run(rv);
 520 }
 521
 522 void SpdyHttpStream::GetSSLInfo(SSLInfo* ssl_info) {
 523   DCHECK(stream_.get());
 524   bool using_npn;
 525   NextProto protocol_negotiated = kProtoUnknown;
 526   stream_->GetSSLInfo(ssl_info, &using_npn, &protocol_negotiated);
 527 }
 528
 529 void SpdyHttpStream::GetSSLCertRequestInfo(
 530     SSLCertRequestInfo* cert_request_info) {
 531   // A SPDY stream cannot request client certificates. Client authentication may
 532   // only occur during the initial SSL handshake.
 533   NOTREACHED();
 534 }
 535
 536 bool SpdyHttpStream::IsSpdyHttpStream() const {
 537   return true;
 538 }
 539
 540 void SpdyHttpStream::Drain(HttpNetworkSession* session) {
 541   Close(false);
 542   delete this;
 543 }
 544
 545 void SpdyHttpStream::SetPriority(RequestPriority priority) {
 546   // TODO(akalin): Plumb this through to |stream_request_| and
 547   // |stream_|.
 548 }
 549
 550 }  // namespace net