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/bind_helpers.h"
  12 #include "base/logging.h"
  13 #include "base/message_loop.h"
  14 #include "base/stringprintf.h"
  15 #include "net/base/host_port_pair.h"
  16 #include "net/base/net_log.h"
  17 #include "net/base/net_util.h"
  18 #include "net/base/upload_data_stream.h"
  19 #include "net/http/http_request_headers.h"
  20 #include "net/http/http_request_info.h"
  21 #include "net/http/http_response_info.h"
  22 #include "net/spdy/spdy_header_block.h"
  23 #include "net/spdy/spdy_http_utils.h"
  24 #include "net/spdy/spdy_session.h"
  25
  26 namespace net {
  27
  28 SpdyHttpStream::SpdyHttpStream(SpdySession* spdy_session,
  29                                bool direct)
  30     : ALLOW_THIS_IN_INITIALIZER_LIST(weak_factory_(this)),
  31       stream_(NULL),
  32       spdy_session_(spdy_session),
  33       request_info_(NULL),
  34       has_upload_data_(false),
  35       response_info_(NULL),
  36       download_finished_(false),
  37       response_headers_received_(false),
  38       user_buffer_len_(0),
  39       buffered_read_callback_pending_(false),
  40       more_read_data_pending_(false),
  41       direct_(direct) { }
  42
  43 void SpdyHttpStream::InitializeWithExistingStream(SpdyStream* spdy_stream) {
  44   stream_ = spdy_stream;
  45   stream_->SetDelegate(this);
  46   response_headers_received_ = true;
  47 }
  48
  49 SpdyHttpStream::~SpdyHttpStream() {
  50   if (stream_)
  51     stream_->DetachDelegate();
  52 }
  53
  54 int SpdyHttpStream::InitializeStream(const HttpRequestInfo* request_info,
  55                                      const BoundNetLog& stream_net_log,
  56                                      const CompletionCallback& callback) {
  57   DCHECK(!stream_.get());
  58   if (spdy_session_->IsClosed())
  59    return ERR_CONNECTION_CLOSED;
  60
  61   request_info_ = request_info;
  62   if (request_info_->method == "GET") {
  63     int error = spdy_session_->GetPushStream(request_info_->url, &stream_,
  64                                              stream_net_log);
  65     if (error != OK)
  66       return error;
  67   }
  68
  69   if (stream_.get())
  70     return OK;
  71
  72   return spdy_session_->CreateStream(request_info_->url,
  73                                      request_info_->priority, &stream_,
  74                                      stream_net_log, callback);
  75 }
  76
  77 const HttpResponseInfo* SpdyHttpStream::GetResponseInfo() const {
  78   return response_info_;
  79 }
  80
  81 UploadProgress SpdyHttpStream::GetUploadProgress() const {
  82   if (!request_info_ || !request_info_->upload_data_stream)
  83     return UploadProgress();
  84
  85   return UploadProgress(request_info_->upload_data_stream->position(),
  86                         request_info_->upload_data_stream->size());
  87 }
  88
  89 int SpdyHttpStream::ReadResponseHeaders(const CompletionCallback& callback) {
  90   CHECK(!callback.is_null());
  91   CHECK(!stream_->cancelled());
  92
  93   if (stream_->closed())
  94     return stream_->response_status();
  95
  96   // Check if we already have the response headers. If so, return synchronously.
  97   if(stream_->response_received()) {
  98     CHECK(stream_->is_idle());
  99     return OK;
 100   }
 101
 102   // Still waiting for the response, return IO_PENDING.
 103   CHECK(callback_.is_null());
 104   callback_ = callback;
 105   return ERR_IO_PENDING;
 106 }
 107
 108 int SpdyHttpStream::ReadResponseBody(
 109     IOBuffer* buf, int buf_len, const CompletionCallback& callback) {
 110   CHECK(stream_->is_idle());
 111   CHECK(buf);
 112   CHECK(buf_len);
 113   CHECK(!callback.is_null());
 114
 115   // If we have data buffered, complete the IO immediately.
 116   if (!response_body_.empty()) {
 117     int bytes_read = 0;
 118     while (!response_body_.empty() && buf_len > 0) {
 119       scoped_refptr<IOBufferWithSize> data = response_body_.front();
 120       const int bytes_to_copy = std::min(buf_len, data->size());
 121       memcpy(&(buf->data()[bytes_read]), data->data(), bytes_to_copy);
 122       buf_len -= bytes_to_copy;
 123       if (bytes_to_copy == data->size()) {
 124         response_body_.pop_front();
 125       } else {
 126         const int bytes_remaining = data->size() - bytes_to_copy;
 127         IOBufferWithSize* new_buffer = new IOBufferWithSize(bytes_remaining);
 128         memcpy(new_buffer->data(), &(data->data()[bytes_to_copy]),
 129                bytes_remaining);
 130         response_body_.pop_front();
 131         response_body_.push_front(make_scoped_refptr(new_buffer));
 132       }
 133       bytes_read += bytes_to_copy;
 134     }
 135     stream_->IncreaseRecvWindowSize(bytes_read);
 136     return bytes_read;
 137   } else if (stream_->closed()) {
 138     return stream_->response_status();
 139   }
 140
 141   CHECK(callback_.is_null());
 142   CHECK(!user_buffer_);
 143   CHECK_EQ(0, user_buffer_len_);
 144
 145   callback_ = callback;
 146   user_buffer_ = buf;
 147   user_buffer_len_ = buf_len;
 148   return ERR_IO_PENDING;
 149 }
 150
 151 void SpdyHttpStream::Close(bool not_reusable) {
 152   // Note: the not_reusable flag has no meaning for SPDY streams.
 153
 154   Cancel();
 155 }
 156
 157 HttpStream* SpdyHttpStream::RenewStreamForAuth() {
 158   return NULL;
 159 }
 160
 161 bool SpdyHttpStream::IsResponseBodyComplete() const {
 162   if (!stream_)
 163     return false;
 164   return stream_->closed();
 165 }
 166
 167 bool SpdyHttpStream::CanFindEndOfResponse() const {
 168   return true;
 169 }
 170
 171 bool SpdyHttpStream::IsMoreDataBuffered() const {
 172   return false;
 173 }
 174
 175 bool SpdyHttpStream::IsConnectionReused() const {
 176   return spdy_session_->IsReused();
 177 }
 178
 179 void SpdyHttpStream::SetConnectionReused() {
 180   // SPDY doesn't need an indicator here.
 181 }
 182
 183 bool SpdyHttpStream::IsConnectionReusable() const {
 184   // SPDY streams aren't considered reusable.
 185   return false;
 186 }
 187
 188 int SpdyHttpStream::SendRequest(const HttpRequestHeaders& request_headers,
 189                                 HttpResponseInfo* response,
 190                                 const CompletionCallback& callback) {
 191   base::Time request_time = base::Time::Now();
 192   CHECK(stream_.get());
 193
 194   stream_->SetDelegate(this);
 195
 196   scoped_ptr<SpdyHeaderBlock> headers(new SpdyHeaderBlock);
 197   CreateSpdyHeadersFromHttpRequest(*request_info_, request_headers,
 198                                    headers.get(), stream_->GetProtocolVersion(),
 199                                    direct_);
 200   stream_->net_log().AddEvent(
 201       NetLog::TYPE_HTTP_TRANSACTION_SPDY_SEND_REQUEST_HEADERS,
 202       base::Bind(&SpdyHeaderBlockNetLogCallback, headers.get()));
 203   stream_->set_spdy_headers(headers.Pass());
 204
 205   stream_->SetRequestTime(request_time);
 206   // This should only get called in the case of a request occurring
 207   // during server push that has already begun but hasn't finished,
 208   // so we set the response's request time to be the actual one
 209   if (response_info_)
 210     response_info_->request_time = request_time;
 211
 212   CHECK(!has_upload_data_);
 213   has_upload_data_ = request_info_->upload_data_stream &&
 214       (request_info_->upload_data_stream->size() ||
 215        request_info_->upload_data_stream->is_chunked());
 216   if (has_upload_data_) {
 217     // Use kMaxSpdyFrameChunkSize as the buffer size, since the request
 218     // body data is written with this size at a time.
 219     raw_request_body_buf_ = new IOBufferWithSize(kMaxSpdyFrameChunkSize);
 220     // The request body buffer is empty at first.
 221     request_body_buf_ = new DrainableIOBuffer(raw_request_body_buf_, 0);
 222   }
 223
 224   CHECK(!callback.is_null());
 225   CHECK(!stream_->cancelled());
 226   CHECK(response);
 227
 228   if (!stream_->pushed() && stream_->closed()) {
 229     if (stream_->response_status() == OK)
 230       return ERR_FAILED;
 231     else
 232       return stream_->response_status();
 233   }
 234
 235   // SendRequest can be called in two cases.
 236   //
 237   // a) A client initiated request. In this case, |response_info_| should be
 238   //    NULL to start with.
 239   // b) A client request which matches a response that the server has already
 240   //    pushed.
 241   if (push_response_info_.get()) {
 242     *response = *(push_response_info_.get());
 243     push_response_info_.reset();
 244   } else {
 245     DCHECK_EQ(static_cast<HttpResponseInfo*>(NULL), response_info_);
 246   }
 247
 248   response_info_ = response;
 249
 250   // Put the peer's IP address and port into the response.
 251   IPEndPoint address;
 252   int result = stream_->GetPeerAddress(&address);
 253   if (result != OK)
 254     return result;
 255   response_info_->socket_address = HostPortPair::FromIPEndPoint(address);
 256
 257   result = stream_->SendRequest(has_upload_data_);
 258   if (result == ERR_IO_PENDING) {
 259     CHECK(callback_.is_null());
 260     callback_ = callback;
 261   }
 262   return result;
 263 }
 264
 265 void SpdyHttpStream::Cancel() {
 266   if (spdy_session_)
 267     spdy_session_->CancelPendingCreateStreams(&stream_);
 268   callback_.Reset();
 269   if (stream_)
 270     stream_->Cancel();
 271 }
 272
 273 int SpdyHttpStream::SendData() {
 274   CHECK(request_info_ && request_info_->upload_data_stream);
 275   CHECK_EQ(0, request_body_buf_->BytesRemaining());
 276
 277   // Read the data from the request body stream.
 278   const int bytes_read = request_info_->upload_data_stream->Read(
 279       raw_request_body_buf_, raw_request_body_buf_->size(),
 280       base::Bind(
 281           base::IgnoreResult(&SpdyHttpStream::OnRequestBodyReadCompleted),
 282           weak_factory_.GetWeakPtr()));
 283
 284   if (bytes_read == ERR_IO_PENDING)
 285     return ERR_IO_PENDING;
 286   // ERR_IO_PENDING is the only possible error.
 287   DCHECK_GE(bytes_read, 0);
 288   return OnRequestBodyReadCompleted(bytes_read);
 289 }
 290
 291 bool SpdyHttpStream::OnSendHeadersComplete(int status) {
 292   if (!callback_.is_null())
 293     DoCallback(status);
 294   return !has_upload_data_;
 295 }
 296
 297 int SpdyHttpStream::OnSendBody() {
 298   CHECK(request_info_ && request_info_->upload_data_stream);
 299   const bool eof = request_info_->upload_data_stream->IsEOF();
 300   if (request_body_buf_->BytesRemaining() > 0) {
 301     return stream_->WriteStreamData(
 302         request_body_buf_,
 303         request_body_buf_->BytesRemaining(),
 304         eof ? DATA_FLAG_FIN : DATA_FLAG_NONE);
 305   }
 306
 307   // The entire body data has been sent.
 308   if (eof)
 309     return OK;
 310
 311   return SendData();
 312 }
 313
 314 int SpdyHttpStream::OnSendBodyComplete(int status, bool* eof) {
 315   // |status| is the number of bytes written to the SPDY stream.
 316   CHECK(request_info_ && request_info_->upload_data_stream);
 317   *eof = false;
 318
 319   if (status > 0) {
 320     request_body_buf_->DidConsume(status);
 321     if (request_body_buf_->BytesRemaining()) {
 322       // Go back to OnSendBody() to send the remaining data.
 323       return OK;
 324     }
 325   }
 326
 327   // Check if the entire body data has been sent.
 328   *eof = (request_info_->upload_data_stream->IsEOF() &&
 329           !request_body_buf_->BytesRemaining());
 330   return OK;
 331 }
 332
 333 int SpdyHttpStream::OnResponseReceived(const SpdyHeaderBlock& response,
 334                                        base::Time response_time,
 335                                        int status) {
 336   if (!response_info_) {
 337     DCHECK(stream_->pushed());
 338     push_response_info_.reset(new HttpResponseInfo);
 339     response_info_ = push_response_info_.get();
 340   }
 341
 342   // If the response is already received, these headers are too late.
 343   if (response_headers_received_) {
 344     LOG(WARNING) << "SpdyHttpStream headers received after response started.";
 345     return OK;
 346   }
 347
 348   // TODO(mbelshe): This is the time of all headers received, not just time
 349   // to first byte.
 350   response_info_->response_time = base::Time::Now();
 351
 352   if (!SpdyHeadersToHttpResponse(response, stream_->GetProtocolVersion(),
 353                                  response_info_)) {
 354     // We might not have complete headers yet.
 355     return ERR_INCOMPLETE_SPDY_HEADERS;
 356   }
 357
 358   response_headers_received_ = true;
 359   // Don't store the SSLInfo in the response here, HttpNetworkTransaction
 360   // will take care of that part.
 361   SSLInfo ssl_info;
 362   NextProto protocol_negotiated = kProtoUnknown;
 363   stream_->GetSSLInfo(&ssl_info,
 364                       &response_info_->was_npn_negotiated,
 365                       &protocol_negotiated);
 366   response_info_->npn_negotiated_protocol =
 367       SSLClientSocket::NextProtoToString(protocol_negotiated);
 368   response_info_->request_time = stream_->GetRequestTime();
 369   response_info_->vary_data.Init(*request_info_, *response_info_->headers);
 370   // TODO(ahendrickson): This is recorded after the entire SYN_STREAM control
 371   // frame has been received and processed.  Move to framer?
 372   response_info_->response_time = response_time;
 373
 374   if (!callback_.is_null())
 375     DoCallback(status);
 376
 377   return status;
 378 }
 379
 380 void SpdyHttpStream::OnHeadersSent() {
 381   // For HTTP streams, no HEADERS frame is sent from the client.
 382   NOTREACHED();
 383 }
 384
 385 int SpdyHttpStream::OnDataReceived(const char* data, int length) {
 386   // SpdyStream won't call us with data if the header block didn't contain a
 387   // valid set of headers.  So we don't expect to not have headers received
 388   // here.
 389   if (!response_headers_received_)
 390     return ERR_INCOMPLETE_SPDY_HEADERS;
 391
 392   // Note that data may be received for a SpdyStream prior to the user calling
 393   // ReadResponseBody(), therefore user_buffer_ may be NULL.  This may often
 394   // happen for server initiated streams.
 395   DCHECK(!stream_->closed() || stream_->pushed());
 396   if (length > 0) {
 397     // Save the received data.
 398     IOBufferWithSize* io_buffer = new IOBufferWithSize(length);
 399     memcpy(io_buffer->data(), data, length);
 400     response_body_.push_back(make_scoped_refptr(io_buffer));
 401
 402     if (user_buffer_) {
 403       // Handing small chunks of data to the caller creates measurable overhead.
 404       // We buffer data in short time-spans and send a single read notification.
 405       ScheduleBufferedReadCallback();
 406     }
 407   }
 408   return OK;
 409 }
 410
 411 void SpdyHttpStream::OnDataSent(int length) {
 412   // For HTTP streams, no data is sent from the client while in the OPEN state,
 413   // so it is never called.
 414   NOTREACHED();
 415 }
 416
 417 void SpdyHttpStream::OnClose(int status) {
 418   bool invoked_callback = false;
 419   if (status == net::OK) {
 420     // We need to complete any pending buffered read now.
 421     invoked_callback = DoBufferedReadCallback();
 422   }
 423   if (!invoked_callback && !callback_.is_null())
 424     DoCallback(status);
 425 }
 426
 427 void SpdyHttpStream::ScheduleBufferedReadCallback() {
 428   // If there is already a scheduled DoBufferedReadCallback, don't issue
 429   // another one.  Mark that we have received more data and return.
 430   if (buffered_read_callback_pending_) {
 431     more_read_data_pending_ = true;
 432     return;
 433   }
 434
 435   more_read_data_pending_ = false;
 436   buffered_read_callback_pending_ = true;
 437   const base::TimeDelta kBufferTime = base::TimeDelta::FromMilliseconds(1);
 438   MessageLoop::current()->PostDelayedTask(
 439       FROM_HERE,
 440       base::Bind(base::IgnoreResult(&SpdyHttpStream::DoBufferedReadCallback),
 441                  weak_factory_.GetWeakPtr()),
 442       kBufferTime);
 443 }
 444
 445 // Checks to see if we should wait for more buffered data before notifying
 446 // the caller.  Returns true if we should wait, false otherwise.
 447 bool SpdyHttpStream::ShouldWaitForMoreBufferedData() const {
 448   // If the response is complete, there is no point in waiting.
 449   if (stream_->closed())
 450     return false;
 451
 452   int bytes_buffered = 0;
 453   std::list<scoped_refptr<IOBufferWithSize> >::const_iterator it;
 454   for (it = response_body_.begin();
 455        it != response_body_.end() && bytes_buffered < user_buffer_len_;
 456        ++it)
 457     bytes_buffered += (*it)->size();
 458
 459   return bytes_buffered < user_buffer_len_;
 460 }
 461
 462 bool SpdyHttpStream::DoBufferedReadCallback() {
 463   weak_factory_.InvalidateWeakPtrs();
 464   buffered_read_callback_pending_ = false;
 465
 466   // If the transaction is cancelled or errored out, we don't need to complete
 467   // the read.
 468   if (!stream_ || stream_->response_status() != OK || stream_->cancelled())
 469     return false;
 470
 471   // When more_read_data_pending_ is true, it means that more data has
 472   // arrived since we started waiting.  Wait a little longer and continue
 473   // to buffer.
 474   if (more_read_data_pending_ && ShouldWaitForMoreBufferedData()) {
 475     ScheduleBufferedReadCallback();
 476     return false;
 477   }
 478
 479   int rv = 0;
 480   if (user_buffer_) {
 481     rv = ReadResponseBody(user_buffer_, user_buffer_len_, callback_);
 482     CHECK_NE(rv, ERR_IO_PENDING);
 483     user_buffer_ = NULL;
 484     user_buffer_len_ = 0;
 485     DoCallback(rv);
 486     return true;
 487   }
 488   return false;
 489 }
 490
 491 void SpdyHttpStream::DoCallback(int rv) {
 492   CHECK_NE(rv, ERR_IO_PENDING);
 493   CHECK(!callback_.is_null());
 494
 495   // Since Run may result in being called back, clear user_callback_ in advance.
 496   CompletionCallback c = callback_;
 497   callback_.Reset();
 498   c.Run(rv);
 499 }
 500
 501 int SpdyHttpStream::OnRequestBodyReadCompleted(int status) {
 502   DCHECK_GE(status, 0);
 503
 504   request_body_buf_ = new DrainableIOBuffer(raw_request_body_buf_, status);
 505
 506   const bool eof = request_info_->upload_data_stream->IsEOF();
 507   return stream_->WriteStreamData(request_body_buf_,
 508                                   request_body_buf_->BytesRemaining(),
 509                                   eof ? DATA_FLAG_FIN : DATA_FLAG_NONE);
 510 }
 511
 512 void SpdyHttpStream::GetSSLInfo(SSLInfo* ssl_info) {
 513   DCHECK(stream_);
 514   bool using_npn;
 515   NextProto protocol_negotiated = kProtoUnknown;
 516   stream_->GetSSLInfo(ssl_info, &using_npn, &protocol_negotiated);
 517 }
 518
 519 void SpdyHttpStream::GetSSLCertRequestInfo(
 520     SSLCertRequestInfo* cert_request_info) {
 521   DCHECK(stream_);
 522   stream_->GetSSLCertRequestInfo(cert_request_info);
 523 }
 524
 525 bool SpdyHttpStream::IsSpdyHttpStream() const {
 526   return true;
 527 }
 528
 529 void SpdyHttpStream::Drain(HttpNetworkSession* session) {
 530   Close(false);
 531   delete this;
 532 }
 533
 534 }  // namespace net