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