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