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