net/quic/quic_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/quic/quic_http_stream.h"
   6
   7 #include "base/callback_helpers.h"
   8 #include "base/strings/stringprintf.h"
   9 #include "net/base/io_buffer.h"
  10 #include "net/base/net_errors.h"
  11 #include "net/http/http_response_headers.h"
  12 #include "net/http/http_util.h"
  13 #include "net/quic/quic_client_session.h"
  14 #include "net/quic/quic_http_utils.h"
  15 #include "net/quic/quic_reliable_client_stream.h"
  16 #include "net/quic/quic_utils.h"
  17 #include "net/socket/next_proto.h"
  18 #include "net/spdy/spdy_frame_builder.h"
  19 #include "net/spdy/spdy_framer.h"
  20 #include "net/spdy/spdy_http_utils.h"
  21 #include "net/ssl/ssl_info.h"
  22
  23 namespace net {
  24
  25 static const size_t kHeaderBufInitialSize = 4096;
  26
  27 QuicHttpStream::QuicHttpStream(const base::WeakPtr<QuicClientSession>& session)
  28     : next_state_(STATE_NONE),
  29       session_(session),
  30       session_error_(OK),
  31       was_handshake_confirmed_(session->IsCryptoHandshakeConfirmed()),
  32       stream_(NULL),
  33       request_info_(NULL),
  34       request_body_stream_(NULL),
  35       priority_(MINIMUM_PRIORITY),
  36       response_info_(NULL),
  37       response_status_(OK),
  38       response_headers_received_(false),
  39       read_buf_(new GrowableIOBuffer()),
  40       user_buffer_len_(0),
  41       weak_factory_(this) {
  42   DCHECK(session_);
  43   session_->AddObserver(this);
  44 }
  45
  46 QuicHttpStream::~QuicHttpStream() {
  47   Close(false);
  48   if (session_)
  49     session_->RemoveObserver(this);
  50 }
  51
  52 int QuicHttpStream::InitializeStream(const HttpRequestInfo* request_info,
  53                                      RequestPriority priority,
  54                                      const BoundNetLog& stream_net_log,
  55                                      const CompletionCallback& callback) {
  56   DCHECK(!stream_);
  57   if (!session_)
  58     return was_handshake_confirmed_ ? ERR_CONNECTION_CLOSED :
  59         ERR_QUIC_HANDSHAKE_FAILED;
  60
  61   stream_net_log_ = stream_net_log;
  62   request_info_ = request_info;
  63   priority_ = priority;
  64
  65   int rv = stream_request_.StartRequest(
  66       session_, &stream_, base::Bind(&QuicHttpStream::OnStreamReady,
  67                                      weak_factory_.GetWeakPtr()));
  68   if (rv == ERR_IO_PENDING) {
  69     callback_ = callback;
  70   } else if (rv == OK) {
  71     stream_->SetDelegate(this);
  72   } else if (!was_handshake_confirmed_) {
  73     rv = ERR_QUIC_HANDSHAKE_FAILED;
  74   }
  75
  76   return rv;
  77 }
  78
  79 void QuicHttpStream::OnStreamReady(int rv) {
  80   DCHECK(rv == OK || !stream_);
  81   if (rv == OK) {
  82     stream_->SetDelegate(this);
  83   } else if (!was_handshake_confirmed_) {
  84     rv = ERR_QUIC_HANDSHAKE_FAILED;
  85   }
  86
  87   ResetAndReturn(&callback_).Run(rv);
  88 }
  89
  90 int QuicHttpStream::SendRequest(const HttpRequestHeaders& request_headers,
  91                                 HttpResponseInfo* response,
  92                                 const CompletionCallback& callback) {
  93   CHECK(stream_);
  94   CHECK(!request_body_stream_);
  95   CHECK(!response_info_);
  96   CHECK(!callback.is_null());
  97   CHECK(response);
  98
  99   QuicPriority priority = ConvertRequestPriorityToQuicPriority(priority_);
 100   stream_->set_priority(priority);
 101   // Store the serialized request headers.
 102   SpdyHeaderBlock headers;
 103   CreateSpdyHeadersFromHttpRequest(*request_info_, request_headers,
 104                                    &headers, 3, /*direct=*/true);
 105   request_ = stream_->compressor()->CompressHeadersWithPriority(priority,
 106                                                                 headers);
 107   // Log the actual request with the URL Request's net log.
 108   stream_net_log_.AddEvent(
 109       NetLog::TYPE_HTTP_TRANSACTION_SPDY_SEND_REQUEST_HEADERS,
 110       base::Bind(&SpdyHeaderBlockNetLogCallback, &headers));
 111   // Also log to the QuicSession's net log.
 112   stream_->net_log().AddEvent(
 113       NetLog::TYPE_QUIC_HTTP_STREAM_SEND_REQUEST_HEADERS,
 114       base::Bind(&SpdyHeaderBlockNetLogCallback, &headers));
 115
 116   // Store the request body.
 117   request_body_stream_ = request_info_->upload_data_stream;
 118   if (request_body_stream_) {
 119     // TODO(rch): Can we be more precise about when to allocate
 120     // raw_request_body_buf_. Removed the following check. DoReadRequestBody()
 121     // was being called even if we didn't yet allocate raw_request_body_buf_.
 122     //   && (request_body_stream_->size() ||
 123     //       request_body_stream_->is_chunked()))
 124     //
 125     // Use kMaxPacketSize as the buffer size, since the request
 126     // body data is written with this size at a time.
 127     // TODO(rch): use a smarter value since we can't write an entire
 128     // packet due to overhead.
 129     raw_request_body_buf_ = new IOBufferWithSize(kMaxPacketSize);
 130     // The request body buffer is empty at first.
 131     request_body_buf_ = new DrainableIOBuffer(raw_request_body_buf_.get(), 0);
 132   }
 133
 134   // Store the response info.
 135   response_info_ = response;
 136
 137   next_state_ = STATE_SEND_HEADERS;
 138   int rv = DoLoop(OK);
 139   if (rv == ERR_IO_PENDING)
 140     callback_ = callback;
 141
 142   return rv > 0 ? OK : rv;
 143 }
 144
 145 UploadProgress QuicHttpStream::GetUploadProgress() const {
 146   if (!request_body_stream_)
 147     return UploadProgress();
 148
 149   return UploadProgress(request_body_stream_->position(),
 150                         request_body_stream_->size());
 151 }
 152
 153 int QuicHttpStream::ReadResponseHeaders(const CompletionCallback& callback) {
 154   CHECK(!callback.is_null());
 155
 156   if (stream_ == NULL)
 157     return response_status_;
 158
 159   // Check if we already have the response headers. If so, return synchronously.
 160   if (response_headers_received_)
 161     return OK;
 162
 163   // Still waiting for the response, return IO_PENDING.
 164   CHECK(callback_.is_null());
 165   callback_ = callback;
 166   return ERR_IO_PENDING;
 167 }
 168
 169 const HttpResponseInfo* QuicHttpStream::GetResponseInfo() const {
 170   return response_info_;
 171 }
 172
 173 int QuicHttpStream::ReadResponseBody(
 174     IOBuffer* buf, int buf_len, const CompletionCallback& callback) {
 175   CHECK(buf);
 176   CHECK(buf_len);
 177   CHECK(!callback.is_null());
 178
 179   // If we have data buffered, complete the IO immediately.
 180   if (!response_body_.empty()) {
 181     int bytes_read = 0;
 182     while (!response_body_.empty() && buf_len > 0) {
 183       scoped_refptr<IOBufferWithSize> data = response_body_.front();
 184       const int bytes_to_copy = std::min(buf_len, data->size());
 185       memcpy(&(buf->data()[bytes_read]), data->data(), bytes_to_copy);
 186       buf_len -= bytes_to_copy;
 187       if (bytes_to_copy == data->size()) {
 188         response_body_.pop_front();
 189       } else {
 190         const int bytes_remaining = data->size() - bytes_to_copy;
 191         IOBufferWithSize* new_buffer = new IOBufferWithSize(bytes_remaining);
 192         memcpy(new_buffer->data(), &(data->data()[bytes_to_copy]),
 193                bytes_remaining);
 194         response_body_.pop_front();
 195         response_body_.push_front(make_scoped_refptr(new_buffer));
 196       }
 197       bytes_read += bytes_to_copy;
 198     }
 199     return bytes_read;
 200   }
 201
 202   if (!stream_) {
 203     // If the stream is already closed, there is no body to read.
 204     return response_status_;
 205   }
 206
 207   CHECK(callback_.is_null());
 208   CHECK(!user_buffer_.get());
 209   CHECK_EQ(0, user_buffer_len_);
 210
 211   callback_ = callback;
 212   user_buffer_ = buf;
 213   user_buffer_len_ = buf_len;
 214   return ERR_IO_PENDING;
 215 }
 216
 217 void QuicHttpStream::Close(bool not_reusable) {
 218   // Note: the not_reusable flag has no meaning for SPDY streams.
 219   if (stream_) {
 220     stream_->SetDelegate(NULL);
 221     stream_->Close(QUIC_STREAM_CANCELLED);
 222     stream_ = NULL;
 223   }
 224 }
 225
 226 HttpStream* QuicHttpStream::RenewStreamForAuth() {
 227   return NULL;
 228 }
 229
 230 bool QuicHttpStream::IsResponseBodyComplete() const {
 231   return next_state_ == STATE_OPEN && !stream_;
 232 }
 233
 234 bool QuicHttpStream::CanFindEndOfResponse() const {
 235   return true;
 236 }
 237
 238 bool QuicHttpStream::IsConnectionReused() const {
 239   // TODO(rch): do something smarter here.
 240   return stream_ && stream_->id() > 1;
 241 }
 242
 243 void QuicHttpStream::SetConnectionReused() {
 244   // QUIC doesn't need an indicator here.
 245 }
 246
 247 bool QuicHttpStream::IsConnectionReusable() const {
 248   // QUIC streams aren't considered reusable.
 249   return false;
 250 }
 251
 252 bool QuicHttpStream::GetLoadTimingInfo(LoadTimingInfo* load_timing_info) const {
 253   // TODO(mmenke):  Figure out what to do here.
 254   return true;
 255 }
 256
 257 void QuicHttpStream::GetSSLInfo(SSLInfo* ssl_info) {
 258   DCHECK(stream_);
 259   stream_->GetSSLInfo(ssl_info);
 260 }
 261
 262 void QuicHttpStream::GetSSLCertRequestInfo(
 263     SSLCertRequestInfo* cert_request_info) {
 264   DCHECK(stream_);
 265   NOTIMPLEMENTED();
 266 }
 267
 268 bool QuicHttpStream::IsSpdyHttpStream() const {
 269   return false;
 270 }
 271
 272 void QuicHttpStream::Drain(HttpNetworkSession* session) {
 273   Close(false);
 274   delete this;
 275 }
 276
 277 void QuicHttpStream::SetPriority(RequestPriority priority) {
 278   priority_ = priority;
 279 }
 280
 281 int QuicHttpStream::OnSendData() {
 282   // TODO(rch): Change QUIC IO to provide notifications to the streams.
 283   NOTREACHED();
 284   return OK;
 285 }
 286
 287 int QuicHttpStream::OnSendDataComplete(int status, bool* eof) {
 288   // TODO(rch): Change QUIC IO to provide notifications to the streams.
 289   NOTREACHED();
 290   return OK;
 291 }
 292
 293 int QuicHttpStream::OnDataReceived(const char* data, int length) {
 294   DCHECK_NE(0, length);
 295   // Are we still reading the response headers.
 296   if (!response_headers_received_) {
 297     // Grow the read buffer if necessary.
 298     if (read_buf_->RemainingCapacity() < length) {
 299       size_t additional_capacity = length - read_buf_->RemainingCapacity();
 300       if (additional_capacity < kHeaderBufInitialSize)
 301         additional_capacity = kHeaderBufInitialSize;
 302       read_buf_->SetCapacity(read_buf_->capacity() + additional_capacity);
 303     }
 304     memcpy(read_buf_->data(), data, length);
 305     read_buf_->set_offset(read_buf_->offset() + length);
 306     int rv = ParseResponseHeaders();
 307     if (rv != ERR_IO_PENDING && !callback_.is_null()) {
 308       DoCallback(rv);
 309     }
 310     return OK;
 311   }
 312
 313   if (callback_.is_null()) {
 314     BufferResponseBody(data, length);
 315     return OK;
 316   }
 317
 318   if (length <= user_buffer_len_) {
 319     memcpy(user_buffer_->data(), data, length);
 320   } else {
 321     memcpy(user_buffer_->data(), data, user_buffer_len_);
 322     int delta = length - user_buffer_len_;
 323     BufferResponseBody(data + user_buffer_len_, delta);
 324     length = user_buffer_len_;
 325   }
 326
 327   user_buffer_ = NULL;
 328   user_buffer_len_ = 0;
 329   DoCallback(length);
 330   return OK;
 331 }
 332
 333 void QuicHttpStream::OnClose(QuicErrorCode error) {
 334   if (error != QUIC_NO_ERROR) {
 335     response_status_ = was_handshake_confirmed_ ?
 336         ERR_QUIC_PROTOCOL_ERROR : ERR_QUIC_HANDSHAKE_FAILED;
 337   } else if (!response_headers_received_) {
 338     response_status_ = ERR_ABORTED;
 339   }
 340
 341   stream_ = NULL;
 342   if (!callback_.is_null())
 343     DoCallback(response_status_);
 344 }
 345
 346 void QuicHttpStream::OnError(int error) {
 347   stream_ = NULL;
 348   response_status_ = was_handshake_confirmed_ ?
 349       error : ERR_QUIC_HANDSHAKE_FAILED;
 350   if (!callback_.is_null())
 351     DoCallback(response_status_);
 352 }
 353
 354 bool QuicHttpStream::HasSendHeadersComplete() {
 355   return next_state_ > STATE_SEND_HEADERS_COMPLETE;
 356 }
 357
 358 void QuicHttpStream::OnCryptoHandshakeConfirmed() {
 359   was_handshake_confirmed_ = true;
 360 }
 361
 362 void QuicHttpStream::OnSessionClosed(int error) {
 363   session_error_ = error;
 364   session_.reset();
 365 }
 366
 367 void QuicHttpStream::OnIOComplete(int rv) {
 368   rv = DoLoop(rv);
 369
 370   if (rv != ERR_IO_PENDING && !callback_.is_null()) {
 371     DoCallback(rv);
 372   }
 373 }
 374
 375 void QuicHttpStream::DoCallback(int rv) {
 376   CHECK_NE(rv, ERR_IO_PENDING);
 377   CHECK(!callback_.is_null());
 378
 379   // The client callback can do anything, including destroying this class,
 380   // so any pending callback must be issued after everything else is done.
 381   base::ResetAndReturn(&callback_).Run(rv);
 382 }
 383
 384 int QuicHttpStream::DoLoop(int rv) {
 385   do {
 386     State state = next_state_;
 387     next_state_ = STATE_NONE;
 388     switch (state) {
 389       case STATE_SEND_HEADERS:
 390         CHECK_EQ(OK, rv);
 391         rv = DoSendHeaders();
 392         break;
 393       case STATE_SEND_HEADERS_COMPLETE:
 394         rv = DoSendHeadersComplete(rv);
 395         break;
 396       case STATE_READ_REQUEST_BODY:
 397         CHECK_EQ(OK, rv);
 398         rv = DoReadRequestBody();
 399         break;
 400       case STATE_READ_REQUEST_BODY_COMPLETE:
 401         rv = DoReadRequestBodyComplete(rv);
 402         break;
 403       case STATE_SEND_BODY:
 404         CHECK_EQ(OK, rv);
 405         rv = DoSendBody();
 406         break;
 407       case STATE_SEND_BODY_COMPLETE:
 408         rv = DoSendBodyComplete(rv);
 409         break;
 410       case STATE_OPEN:
 411         CHECK_EQ(OK, rv);
 412         break;
 413       default:
 414         NOTREACHED() << "next_state_: " << next_state_;
 415         break;
 416     }
 417   } while (next_state_ != STATE_NONE && next_state_ != STATE_OPEN &&
 418            rv != ERR_IO_PENDING);
 419
 420   return rv;
 421 }
 422
 423 int QuicHttpStream::DoSendHeaders() {
 424   if (!stream_)
 425     return ERR_UNEXPECTED;
 426
 427   bool has_upload_data = request_body_stream_ != NULL;
 428
 429   next_state_ = STATE_SEND_HEADERS_COMPLETE;
 430   return stream_->WriteStreamData(
 431       request_, !has_upload_data,
 432       base::Bind(&QuicHttpStream::OnIOComplete, weak_factory_.GetWeakPtr()));
 433 }
 434
 435 int QuicHttpStream::DoSendHeadersComplete(int rv) {
 436   if (rv < 0)
 437     return rv;
 438
 439   next_state_ = request_body_stream_ ?
 440       STATE_READ_REQUEST_BODY : STATE_OPEN;
 441
 442   return OK;
 443 }
 444
 445 int QuicHttpStream::DoReadRequestBody() {
 446   next_state_ = STATE_READ_REQUEST_BODY_COMPLETE;
 447   return request_body_stream_->Read(
 448       raw_request_body_buf_.get(),
 449       raw_request_body_buf_->size(),
 450       base::Bind(&QuicHttpStream::OnIOComplete, weak_factory_.GetWeakPtr()));
 451 }
 452
 453 int QuicHttpStream::DoReadRequestBodyComplete(int rv) {
 454   // |rv| is the result of read from the request body from the last call to
 455   // DoSendBody().
 456   if (rv < 0)
 457     return rv;
 458
 459   request_body_buf_ = new DrainableIOBuffer(raw_request_body_buf_.get(), rv);
 460   if (rv == 0) {  // Reached the end.
 461     DCHECK(request_body_stream_->IsEOF());
 462   }
 463
 464   next_state_ = STATE_SEND_BODY;
 465   return OK;
 466 }
 467
 468 int QuicHttpStream::DoSendBody() {
 469   if (!stream_)
 470     return ERR_UNEXPECTED;
 471
 472   CHECK(request_body_stream_);
 473   CHECK(request_body_buf_.get());
 474   const bool eof = request_body_stream_->IsEOF();
 475   int len = request_body_buf_->BytesRemaining();
 476   if (len > 0 || eof) {
 477     next_state_ = STATE_SEND_BODY_COMPLETE;
 478     base::StringPiece data(request_body_buf_->data(), len);
 479     return stream_->WriteStreamData(
 480         data, eof,
 481         base::Bind(&QuicHttpStream::OnIOComplete, weak_factory_.GetWeakPtr()));
 482   }
 483
 484   next_state_ = STATE_OPEN;
 485   return OK;
 486 }
 487
 488 int QuicHttpStream::DoSendBodyComplete(int rv) {
 489   if (rv < 0)
 490     return rv;
 491
 492   request_body_buf_->DidConsume(request_body_buf_->BytesRemaining());
 493
 494   if (!request_body_stream_->IsEOF()) {
 495     next_state_ = STATE_READ_REQUEST_BODY;
 496     return OK;
 497   }
 498
 499   next_state_ = STATE_OPEN;
 500   return OK;
 501 }
 502
 503 int QuicHttpStream::ParseResponseHeaders() {
 504   size_t read_buf_len = static_cast<size_t>(read_buf_->offset());
 505   SpdyFramer framer(SPDY3);
 506   SpdyHeaderBlock headers;
 507   char* data = read_buf_->StartOfBuffer();
 508   size_t len = framer.ParseHeaderBlockInBuffer(data, read_buf_->offset(),
 509                                                &headers);
 510
 511   if (len == 0) {
 512     return ERR_IO_PENDING;
 513   }
 514
 515   // Save the remaining received data.
 516   size_t delta = read_buf_len - len;
 517   if (delta > 0) {
 518     BufferResponseBody(data + len, delta);
 519   }
 520
 521   // The URLRequest logs these headers, so only log to the QuicSession's
 522   // net log.
 523   stream_->net_log().AddEvent(
 524       NetLog::TYPE_QUIC_HTTP_STREAM_READ_RESPONSE_HEADERS,
 525       base::Bind(&SpdyHeaderBlockNetLogCallback, &headers));
 526
 527   if (!SpdyHeadersToHttpResponse(headers, 3, response_info_)) {
 528     DLOG(WARNING) << "Invalid headers";
 529     return ERR_QUIC_PROTOCOL_ERROR;
 530   }
 531   // Put the peer's IP address and port into the response.
 532   IPEndPoint address = stream_->GetPeerAddress();
 533   response_info_->socket_address = HostPortPair::FromIPEndPoint(address);
 534   response_info_->connection_info =
 535       HttpResponseInfo::CONNECTION_INFO_QUIC1_SPDY3;
 536   response_info_->vary_data
 537       .Init(*request_info_, *response_info_->headers.get());
 538   response_info_->was_npn_negotiated = true;
 539   response_info_->npn_negotiated_protocol = "quic/1+spdy/3";
 540   response_headers_received_ = true;
 541
 542   return OK;
 543 }
 544
 545 void QuicHttpStream::BufferResponseBody(const char* data, int length) {
 546   if (length == 0)
 547     return;
 548   IOBufferWithSize* io_buffer = new IOBufferWithSize(length);
 549   memcpy(io_buffer->data(), data, length);
 550   response_body_.push_back(make_scoped_refptr(io_buffer));
 551 }
 552
 553 }  // namespace net