net/spdy/spdy_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_stream.h"
   6
   7 #include <limits>
   8
   9 #include "base/bind.h"
  10 #include "base/compiler_specific.h"
  11 #include "base/logging.h"
  12 #include "base/message_loop.h"
  13 #include "base/stringprintf.h"
  14 #include "base/strings/string_number_conversions.h"
  15 #include "base/values.h"
  16 #include "net/spdy/spdy_buffer_producer.h"
  17 #include "net/spdy/spdy_http_utils.h"
  18 #include "net/spdy/spdy_session.h"
  19
  20 namespace net {
  21
  22 namespace {
  23
  24 Value* NetLogSpdyStreamErrorCallback(SpdyStreamId stream_id,
  25                                      int status,
  26                                      const std::string* description,
  27                                      NetLog::LogLevel /* log_level */) {
  28   DictionaryValue* dict = new DictionaryValue();
  29   dict->SetInteger("stream_id", static_cast<int>(stream_id));
  30   dict->SetInteger("status", status);
  31   dict->SetString("description", *description);
  32   return dict;
  33 }
  34
  35 Value* NetLogSpdyStreamWindowUpdateCallback(SpdyStreamId stream_id,
  36                                             int32 delta,
  37                                             int32 window_size,
  38                                             NetLog::LogLevel /* log_level */) {
  39   DictionaryValue* dict = new DictionaryValue();
  40   dict->SetInteger("stream_id", stream_id);
  41   dict->SetInteger("delta", delta);
  42   dict->SetInteger("window_size", window_size);
  43   return dict;
  44 }
  45
  46 bool ContainsUpperAscii(const std::string& str) {
  47   for (std::string::const_iterator i(str.begin()); i != str.end(); ++i) {
  48     if (*i >= 'A' && *i <= 'Z') {
  49       return true;
  50     }
  51   }
  52   return false;
  53 }
  54
  55 }  // namespace
  56
  57 // A wrapper around a stream that calls into ProduceSynStreamFrame().
  58 class SpdyStream::SynStreamBufferProducer : public SpdyBufferProducer {
  59  public:
  60   SynStreamBufferProducer(const base::WeakPtr<SpdyStream>& stream)
  61       : stream_(stream) {
  62     DCHECK(stream_);
  63   }
  64
  65   virtual ~SynStreamBufferProducer() {}
  66
  67   virtual scoped_ptr<SpdyBuffer> ProduceBuffer() OVERRIDE {
  68     if (!stream_) {
  69       NOTREACHED();
  70       return scoped_ptr<SpdyBuffer>();
  71     }
  72     DCHECK_GT(stream_->stream_id(), 0u);
  73     return scoped_ptr<SpdyBuffer>(
  74         new SpdyBuffer(stream_->ProduceSynStreamFrame()));
  75   }
  76
  77  private:
  78   const base::WeakPtr<SpdyStream> stream_;
  79 };
  80
  81 // A wrapper around a stream that calls into ProduceHeaderFrame() with
  82 // a given header block.
  83 class SpdyStream::HeaderBufferProducer : public SpdyBufferProducer {
  84  public:
  85   HeaderBufferProducer(const base::WeakPtr<SpdyStream>& stream,
  86                       scoped_ptr<SpdyHeaderBlock> headers)
  87       : stream_(stream),
  88         headers_(headers.Pass()) {
  89     DCHECK(stream_);
  90     DCHECK(headers_);
  91   }
  92
  93   virtual ~HeaderBufferProducer() {}
  94
  95   virtual scoped_ptr<SpdyBuffer> ProduceBuffer() OVERRIDE {
  96     if (!stream_) {
  97       NOTREACHED();
  98       return scoped_ptr<SpdyBuffer>();
  99     }
 100     DCHECK_GT(stream_->stream_id(), 0u);
 101     return scoped_ptr<SpdyBuffer>(
 102         new SpdyBuffer(stream_->ProduceHeaderFrame(headers_.Pass())));
 103   }
 104
 105  private:
 106   const base::WeakPtr<SpdyStream> stream_;
 107   scoped_ptr<SpdyHeaderBlock> headers_;
 108 };
 109
 110 SpdyStream::SpdyStream(SpdySession* session,
 111                        const std::string& path,
 112                        RequestPriority priority,
 113                        int32 initial_send_window_size,
 114                        int32 initial_recv_window_size,
 115                        bool pushed,
 116                        const BoundNetLog& net_log)
 117     : weak_ptr_factory_(ALLOW_THIS_IN_INITIALIZER_LIST(this)),
 118       continue_buffering_data_(true),
 119       stream_id_(0),
 120       path_(path),
 121       priority_(priority),
 122       slot_(0),
 123       send_stalled_by_flow_control_(false),
 124       send_window_size_(initial_send_window_size),
 125       recv_window_size_(initial_recv_window_size),
 126       unacked_recv_window_bytes_(0),
 127       pushed_(pushed),
 128       response_received_(false),
 129       session_(session),
 130       delegate_(NULL),
 131       request_time_(base::Time::Now()),
 132       response_(new SpdyHeaderBlock),
 133       io_state_(STATE_NONE),
 134       response_status_(OK),
 135       cancelled_(false),
 136       has_upload_data_(false),
 137       net_log_(net_log),
 138       send_bytes_(0),
 139       recv_bytes_(0),
 140       domain_bound_cert_type_(CLIENT_CERT_INVALID_TYPE),
 141       just_completed_frame_type_(DATA),
 142       just_completed_frame_size_(0) {
 143 }
 144
 145 SpdyStream::~SpdyStream() {
 146   UpdateHistograms();
 147 }
 148
 149 void SpdyStream::SetDelegate(Delegate* delegate) {
 150   CHECK(delegate);
 151   delegate_ = delegate;
 152
 153   if (pushed_) {
 154     CHECK(response_received());
 155     MessageLoop::current()->PostTask(
 156         FROM_HERE, base::Bind(&SpdyStream::PushedStreamReplayData, this));
 157   } else {
 158     continue_buffering_data_ = false;
 159   }
 160 }
 161
 162 void SpdyStream::PushedStreamReplayData() {
 163   if (cancelled_ || !delegate_)
 164     return;
 165
 166   continue_buffering_data_ = false;
 167
 168   int rv = delegate_->OnResponseReceived(*response_, response_time_, OK);
 169   if (rv == ERR_INCOMPLETE_SPDY_HEADERS) {
 170     // We don't have complete headers.  Assume we're waiting for another
 171     // HEADERS frame.  Since we don't have headers, we had better not have
 172     // any pending data frames.
 173     if (pending_buffers_.size() != 0U) {
 174       LogStreamError(ERR_SPDY_PROTOCOL_ERROR,
 175                      "HEADERS incomplete headers, but pending data frames.");
 176       session_->CloseStream(stream_id_, ERR_SPDY_PROTOCOL_ERROR);
 177     }
 178     return;
 179   }
 180
 181   std::vector<SpdyBuffer*> buffers;
 182   pending_buffers_.release(&buffers);
 183   for (size_t i = 0; i < buffers.size(); ++i) {
 184     // It is always possible that a callback to the delegate results in
 185     // the delegate no longer being available.
 186     if (!delegate_)
 187       break;
 188     if (buffers[i]) {
 189       delegate_->OnDataReceived(scoped_ptr<SpdyBuffer>(buffers[i]));
 190     } else {
 191       delegate_->OnDataReceived(scoped_ptr<SpdyBuffer>());
 192       session_->CloseStream(stream_id_, OK);
 193       // Note: |this| may be deleted after calling CloseStream.
 194       DCHECK_EQ(buffers.size() - 1, i);
 195     }
 196   }
 197 }
 198
 199 scoped_ptr<SpdyFrame> SpdyStream::ProduceSynStreamFrame() {
 200   CHECK_EQ(io_state_, STATE_SEND_HEADERS_COMPLETE);
 201   CHECK(request_.get());
 202   CHECK_GT(stream_id_, 0u);
 203
 204   SpdyControlFlags flags =
 205       has_upload_data_ ? CONTROL_FLAG_NONE : CONTROL_FLAG_FIN;
 206   scoped_ptr<SpdyFrame> frame(session_->CreateSynStream(
 207       stream_id_, priority_, slot_, flags, *request_));
 208   send_time_ = base::TimeTicks::Now();
 209   return frame.Pass();
 210 }
 211
 212 scoped_ptr<SpdyFrame> SpdyStream::ProduceHeaderFrame(
 213     scoped_ptr<SpdyHeaderBlock> header_block) {
 214   CHECK(!cancelled());
 215   // We must need to write stream data.
 216   // Until the headers have been completely sent, we can not be sure
 217   // that our stream_id is correct.
 218   DCHECK_GT(io_state_, STATE_SEND_HEADERS_COMPLETE);
 219   DCHECK_GT(stream_id_, 0u);
 220
 221   // Create actual HEADERS frame just in time because it depends on
 222   // compression context and should not be reordered after the creation.
 223   scoped_ptr<SpdyFrame> header_frame(session_->CreateHeadersFrame(
 224       stream_id_, *header_block, SpdyControlFlags()));
 225   return header_frame.Pass();
 226 }
 227
 228 void SpdyStream::DetachDelegate() {
 229   delegate_ = NULL;
 230   if (!closed())
 231     Cancel();
 232 }
 233
 234 const SpdyHeaderBlock& SpdyStream::spdy_headers() const {
 235   DCHECK(request_ != NULL);
 236   return *request_.get();
 237 }
 238
 239 void SpdyStream::set_spdy_headers(scoped_ptr<SpdyHeaderBlock> headers) {
 240   request_.reset(headers.release());
 241 }
 242
 243 void SpdyStream::AdjustSendWindowSize(int32 delta_window_size) {
 244   DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
 245
 246   if (closed())
 247     return;
 248
 249   // Check for wraparound.
 250   if (send_window_size_ > 0) {
 251     DCHECK_LE(delta_window_size, kint32max - send_window_size_);
 252   }
 253   if (send_window_size_ < 0) {
 254     DCHECK_GE(delta_window_size, kint32min - send_window_size_);
 255   }
 256   send_window_size_ += delta_window_size;
 257   PossiblyResumeIfSendStalled();
 258 }
 259
 260 void SpdyStream::OnWriteBufferConsumed(
 261     size_t frame_payload_size,
 262     size_t consume_size,
 263     SpdyBuffer::ConsumeSource consume_source) {
 264   DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
 265   if (consume_source == SpdyBuffer::DISCARD) {
 266     // If we're discarding a frame or part of it, increase the send
 267     // window by the number of discarded bytes. (Although if we're
 268     // discarding part of a frame, it's probably because of a write
 269     // error and we'll be tearing down the stream soon.)
 270     size_t remaining_payload_bytes = std::min(consume_size, frame_payload_size);
 271     DCHECK_GT(remaining_payload_bytes, 0u);
 272     IncreaseSendWindowSize(static_cast<int32>(remaining_payload_bytes));
 273   }
 274   // For consumed bytes, the send window is increased when we receive
 275   // a WINDOW_UPDATE frame.
 276 }
 277
 278 void SpdyStream::IncreaseSendWindowSize(int32 delta_window_size) {
 279   DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
 280   DCHECK_GE(delta_window_size, 1);
 281
 282   // Ignore late WINDOW_UPDATEs.
 283   if (closed())
 284     return;
 285
 286   if (send_window_size_ > 0) {
 287     // Check for overflow.
 288     int32 max_delta_window_size = kint32max - send_window_size_;
 289     if (delta_window_size > max_delta_window_size) {
 290       std::string desc = base::StringPrintf(
 291           "Received WINDOW_UPDATE [delta: %d] for stream %d overflows "
 292           "send_window_size_ [current: %d]", delta_window_size, stream_id_,
 293           send_window_size_);
 294       session_->ResetStream(stream_id_, RST_STREAM_FLOW_CONTROL_ERROR, desc);
 295       return;
 296     }
 297   }
 298
 299   send_window_size_ += delta_window_size;
 300
 301   net_log_.AddEvent(
 302       NetLog::TYPE_SPDY_STREAM_UPDATE_SEND_WINDOW,
 303       base::Bind(&NetLogSpdyStreamWindowUpdateCallback,
 304                  stream_id_, delta_window_size, send_window_size_));
 305
 306   PossiblyResumeIfSendStalled();
 307 }
 308
 309 void SpdyStream::DecreaseSendWindowSize(int32 delta_window_size) {
 310   DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
 311
 312   if (closed())
 313     return;
 314
 315   // We only call this method when sending a frame. Therefore,
 316   // |delta_window_size| should be within the valid frame size range.
 317   DCHECK_GE(delta_window_size, 1);
 318   DCHECK_LE(delta_window_size, kMaxSpdyFrameChunkSize);
 319
 320   // |send_window_size_| should have been at least |delta_window_size| for
 321   // this call to happen.
 322   DCHECK_GE(send_window_size_, delta_window_size);
 323
 324   send_window_size_ -= delta_window_size;
 325
 326   net_log_.AddEvent(
 327       NetLog::TYPE_SPDY_STREAM_UPDATE_SEND_WINDOW,
 328       base::Bind(&NetLogSpdyStreamWindowUpdateCallback,
 329                  stream_id_, -delta_window_size, send_window_size_));
 330 }
 331
 332 void SpdyStream::OnReadBufferConsumed(
 333     size_t consume_size,
 334     SpdyBuffer::ConsumeSource consume_source) {
 335   DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
 336   DCHECK_GE(consume_size, 1u);
 337   DCHECK_LE(consume_size, static_cast<size_t>(kint32max));
 338   IncreaseRecvWindowSize(static_cast<int32>(consume_size));
 339 }
 340
 341 void SpdyStream::IncreaseRecvWindowSize(int32 delta_window_size) {
 342   DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
 343
 344   // By the time a read is processed by the delegate, this stream may
 345   // already be inactive.
 346   if (!session_->IsStreamActive(stream_id_))
 347     return;
 348
 349   DCHECK_GE(unacked_recv_window_bytes_, 0);
 350   DCHECK_GE(recv_window_size_, unacked_recv_window_bytes_);
 351   DCHECK_GE(delta_window_size, 1);
 352   // Check for overflow.
 353   DCHECK_LE(delta_window_size, kint32max - recv_window_size_);
 354
 355   recv_window_size_ += delta_window_size;
 356   net_log_.AddEvent(
 357       NetLog::TYPE_SPDY_STREAM_UPDATE_RECV_WINDOW,
 358       base::Bind(&NetLogSpdyStreamWindowUpdateCallback,
 359                  stream_id_, delta_window_size, recv_window_size_));
 360
 361   unacked_recv_window_bytes_ += delta_window_size;
 362   if (unacked_recv_window_bytes_ >
 363       session_->stream_initial_recv_window_size() / 2) {
 364     session_->SendStreamWindowUpdate(
 365         stream_id_, static_cast<uint32>(unacked_recv_window_bytes_));
 366     unacked_recv_window_bytes_ = 0;
 367   }
 368 }
 369
 370 void SpdyStream::DecreaseRecvWindowSize(int32 delta_window_size) {
 371   DCHECK(session_->IsStreamActive(stream_id_));
 372   DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
 373   DCHECK_GE(delta_window_size, 1);
 374
 375   // Since we never decrease the initial receive window size,
 376   // |delta_window_size| should never cause |recv_window_size_| to go
 377   // negative. If we do, the receive window isn't being respected.
 378   if (delta_window_size > recv_window_size_) {
 379     session_->ResetStream(
 380         stream_id_, RST_STREAM_PROTOCOL_ERROR,
 381         "delta_window_size is " + base::IntToString(delta_window_size) +
 382             " in DecreaseRecvWindowSize, which is larger than the receive " +
 383             "window size of " + base::IntToString(recv_window_size_));
 384     return;
 385   }
 386
 387   recv_window_size_ -= delta_window_size;
 388   net_log_.AddEvent(
 389       NetLog::TYPE_SPDY_STREAM_UPDATE_RECV_WINDOW,
 390       base::Bind(&NetLogSpdyStreamWindowUpdateCallback,
 391                  stream_id_, -delta_window_size, recv_window_size_));
 392 }
 393
 394 int SpdyStream::GetPeerAddress(IPEndPoint* address) const {
 395   return session_->GetPeerAddress(address);
 396 }
 397
 398 int SpdyStream::GetLocalAddress(IPEndPoint* address) const {
 399   return session_->GetLocalAddress(address);
 400 }
 401
 402 bool SpdyStream::WasEverUsed() const {
 403   return session_->WasEverUsed();
 404 }
 405
 406 base::Time SpdyStream::GetRequestTime() const {
 407   return request_time_;
 408 }
 409
 410 void SpdyStream::SetRequestTime(base::Time t) {
 411   request_time_ = t;
 412 }
 413
 414 int SpdyStream::OnResponseReceived(const SpdyHeaderBlock& response) {
 415   int rv = OK;
 416
 417   metrics_.StartStream();
 418
 419   DCHECK(response_->empty());
 420   *response_ = response;  // TODO(ukai): avoid copy.
 421
 422   recv_first_byte_time_ = base::TimeTicks::Now();
 423   response_time_ = base::Time::Now();
 424
 425   // If we receive a response before we are in STATE_WAITING_FOR_RESPONSE, then
 426   // the server has sent the SYN_REPLY too early.
 427   if (!pushed_ && io_state_ != STATE_WAITING_FOR_RESPONSE)
 428     return ERR_SPDY_PROTOCOL_ERROR;
 429   if (pushed_)
 430     CHECK(io_state_ == STATE_NONE);
 431   io_state_ = STATE_OPEN;
 432
 433   // Append all the headers into the response header block.
 434   for (SpdyHeaderBlock::const_iterator it = response.begin();
 435        it != response.end(); ++it) {
 436     // Disallow uppercase headers.
 437     if (ContainsUpperAscii(it->first)) {
 438       session_->ResetStream(stream_id_, RST_STREAM_PROTOCOL_ERROR,
 439                             "Upper case characters in header: " + it->first);
 440       response_status_ = ERR_SPDY_PROTOCOL_ERROR;
 441       return ERR_SPDY_PROTOCOL_ERROR;
 442     }
 443   }
 444
 445   if ((*response_).find("transfer-encoding") != (*response_).end()) {
 446     session_->ResetStream(stream_id_, RST_STREAM_PROTOCOL_ERROR,
 447                          "Received transfer-encoding header");
 448     return ERR_SPDY_PROTOCOL_ERROR;
 449   }
 450
 451   if (delegate_)
 452     rv = delegate_->OnResponseReceived(*response_, response_time_, rv);
 453   // If delegate_ is not yet attached, we'll call OnResponseReceived after the
 454   // delegate gets attached to the stream.
 455
 456   return rv;
 457 }
 458
 459 int SpdyStream::OnHeaders(const SpdyHeaderBlock& headers) {
 460   DCHECK(!response_->empty());
 461
 462   // Append all the headers into the response header block.
 463   for (SpdyHeaderBlock::const_iterator it = headers.begin();
 464       it != headers.end(); ++it) {
 465     // Disallow duplicate headers.  This is just to be conservative.
 466     if ((*response_).find(it->first) != (*response_).end()) {
 467       LogStreamError(ERR_SPDY_PROTOCOL_ERROR, "HEADERS duplicate header");
 468       response_status_ = ERR_SPDY_PROTOCOL_ERROR;
 469       return ERR_SPDY_PROTOCOL_ERROR;
 470     }
 471
 472     // Disallow uppercase headers.
 473     if (ContainsUpperAscii(it->first)) {
 474       session_->ResetStream(stream_id_, RST_STREAM_PROTOCOL_ERROR,
 475                             "Upper case characters in header: " + it->first);
 476       response_status_ = ERR_SPDY_PROTOCOL_ERROR;
 477       return ERR_SPDY_PROTOCOL_ERROR;
 478     }
 479
 480     (*response_)[it->first] = it->second;
 481   }
 482
 483   if ((*response_).find("transfer-encoding") != (*response_).end()) {
 484     session_->ResetStream(stream_id_, RST_STREAM_PROTOCOL_ERROR,
 485                          "Received transfer-encoding header");
 486     return ERR_SPDY_PROTOCOL_ERROR;
 487   }
 488
 489   int rv = OK;
 490   if (delegate_) {
 491     rv = delegate_->OnResponseReceived(*response_, response_time_, rv);
 492     // ERR_INCOMPLETE_SPDY_HEADERS means that we are waiting for more
 493     // headers before the response header block is complete.
 494     if (rv == ERR_INCOMPLETE_SPDY_HEADERS)
 495       rv = OK;
 496   }
 497   return rv;
 498 }
 499
 500 void SpdyStream::OnDataReceived(scoped_ptr<SpdyBuffer> buffer) {
 501   DCHECK(session_->IsStreamActive(stream_id_));
 502   // If we don't have a response, then the SYN_REPLY did not come through.
 503   // We cannot pass data up to the caller unless the reply headers have been
 504   // received.
 505   if (!response_received()) {
 506     LogStreamError(ERR_SYN_REPLY_NOT_RECEIVED, "Didn't receive a response.");
 507     session_->CloseStream(stream_id_, ERR_SYN_REPLY_NOT_RECEIVED);
 508     return;
 509   }
 510
 511   if (!delegate_ || continue_buffering_data_) {
 512     // It should be valid for this to happen in the server push case.
 513     // We'll return received data when delegate gets attached to the stream.
 514     if (buffer) {
 515       pending_buffers_.push_back(buffer.release());
 516     } else {
 517       pending_buffers_.push_back(NULL);
 518       metrics_.StopStream();
 519       // Note: we leave the stream open in the session until the stream
 520       //       is claimed.
 521     }
 522     return;
 523   }
 524
 525   CHECK(!closed());
 526
 527   if (!buffer) {
 528     metrics_.StopStream();
 529     session_->CloseStream(stream_id_, OK);
 530     // Note: |this| may be deleted after calling CloseStream.
 531     return;
 532   }
 533
 534   size_t length = buffer->GetRemainingSize();
 535   DCHECK_LE(length, session_->GetDataFrameMaximumPayload());
 536   if (session_->flow_control_state() >= SpdySession::FLOW_CONTROL_STREAM) {
 537     DecreaseRecvWindowSize(static_cast<int32>(length));
 538     buffer->AddConsumeCallback(
 539         base::Bind(&SpdyStream::OnReadBufferConsumed,
 540                    weak_ptr_factory_.GetWeakPtr()));
 541   }
 542
 543   // Track our bandwidth.
 544   metrics_.RecordBytes(length);
 545   recv_bytes_ += length;
 546   recv_last_byte_time_ = base::TimeTicks::Now();
 547
 548   if (delegate_->OnDataReceived(buffer.Pass()) != OK) {
 549     // |delegate_| rejected the data.
 550     LogStreamError(ERR_SPDY_PROTOCOL_ERROR, "Delegate rejected the data");
 551     session_->CloseStream(stream_id_, ERR_SPDY_PROTOCOL_ERROR);
 552     return;
 553   }
 554 }
 555
 556 void SpdyStream::OnFrameWriteComplete(SpdyFrameType frame_type,
 557                                       size_t frame_size) {
 558   if (frame_size < session_->GetFrameMinimumSize() ||
 559       frame_size > session_->GetFrameMaximumSize()) {
 560     NOTREACHED();
 561     return;
 562   }
 563   if (cancelled() || closed())
 564     return;
 565   just_completed_frame_type_ = frame_type;
 566   just_completed_frame_size_ = frame_size;
 567   DoLoop(OK);
 568 }
 569
 570 int SpdyStream::GetProtocolVersion() const {
 571   return session_->GetProtocolVersion();
 572 }
 573
 574 void SpdyStream::LogStreamError(int status, const std::string& description) {
 575   net_log_.AddEvent(NetLog::TYPE_SPDY_STREAM_ERROR,
 576                     base::Bind(&NetLogSpdyStreamErrorCallback,
 577                                stream_id_, status, &description));
 578 }
 579
 580 void SpdyStream::OnClose(int status) {
 581   io_state_ = STATE_DONE;
 582   response_status_ = status;
 583   Delegate* delegate = delegate_;
 584   delegate_ = NULL;
 585   if (delegate)
 586     delegate->OnClose(status);
 587 }
 588
 589 void SpdyStream::Cancel() {
 590   if (cancelled())
 591     return;
 592
 593   cancelled_ = true;
 594   if (session_->IsStreamActive(stream_id_))
 595     session_->ResetStream(stream_id_, RST_STREAM_CANCEL, std::string());
 596   else if (stream_id_ == 0)
 597     session_->CloseCreatedStream(this, RST_STREAM_CANCEL);
 598 }
 599
 600 void SpdyStream::Close() {
 601   if (stream_id_ != 0)
 602     session_->CloseStream(stream_id_, OK);
 603   else
 604     session_->CloseCreatedStream(this, OK);
 605 }
 606
 607 int SpdyStream::SendRequest(bool has_upload_data) {
 608   // Pushed streams do not send any data, and should always be
 609   // idle. However, we still want to return IO_PENDING to mimic
 610   // non-push behavior.
 611   has_upload_data_ = has_upload_data;
 612   if (pushed_) {
 613     DCHECK(is_idle());
 614     DCHECK(!has_upload_data_);
 615     DCHECK(response_received());
 616     send_time_ = base::TimeTicks::Now();
 617     return ERR_IO_PENDING;
 618   }
 619   CHECK_EQ(STATE_NONE, io_state_);
 620   io_state_ = STATE_GET_DOMAIN_BOUND_CERT;
 621   return DoLoop(OK);
 622 }
 623
 624 void SpdyStream::QueueHeaders(scoped_ptr<SpdyHeaderBlock> headers) {
 625   // Until the first headers by SYN_STREAM have been completely sent, we can
 626   // not be sure that our stream_id is correct.
 627   DCHECK_GT(io_state_, STATE_SEND_HEADERS_COMPLETE);
 628   CHECK_GT(stream_id_, 0u);
 629
 630   session_->EnqueueStreamWrite(
 631       this, HEADERS,
 632       scoped_ptr<SpdyBufferProducer>(
 633           new HeaderBufferProducer(
 634               weak_ptr_factory_.GetWeakPtr(), headers.Pass())));
 635 }
 636
 637 void SpdyStream::QueueStreamData(IOBuffer* data,
 638                                  int length,
 639                                  SpdyDataFlags flags) {
 640   // Until the headers have been completely sent, we can not be sure
 641   // that our stream_id is correct.
 642   DCHECK_GT(io_state_, STATE_SEND_HEADERS_COMPLETE);
 643   CHECK_GT(stream_id_, 0u);
 644   CHECK(!cancelled());
 645
 646   scoped_ptr<SpdyBuffer> data_buffer(session_->CreateDataBuffer(
 647       stream_id_, data, length, flags));
 648   // We'll get called again by PossiblyResumeIfSendStalled().
 649   if (!data_buffer)
 650     return;
 651
 652   if (session_->flow_control_state() >= SpdySession::FLOW_CONTROL_STREAM) {
 653     DCHECK_GE(data_buffer->GetRemainingSize(),
 654               session_->GetDataFrameMinimumSize());
 655     size_t payload_size =
 656         data_buffer->GetRemainingSize() - session_->GetDataFrameMinimumSize();
 657     DCHECK_LE(payload_size, session_->GetDataFrameMaximumPayload());
 658     DecreaseSendWindowSize(static_cast<int32>(payload_size));
 659     // This currently isn't strictly needed, since write frames are
 660     // discarded only if the stream is about to be closed. But have it
 661     // here anyway just in case this changes.
 662     data_buffer->AddConsumeCallback(
 663         base::Bind(&SpdyStream::OnWriteBufferConsumed,
 664                    weak_ptr_factory_.GetWeakPtr(),
 665                    payload_size));
 666   }
 667
 668   session_->EnqueueStreamWrite(
 669       this, DATA,
 670       scoped_ptr<SpdyBufferProducer>(
 671           new SimpleBufferProducer(data_buffer.Pass())));
 672 }
 673
 674 bool SpdyStream::GetSSLInfo(SSLInfo* ssl_info,
 675                             bool* was_npn_negotiated,
 676                             NextProto* protocol_negotiated) {
 677   return session_->GetSSLInfo(
 678       ssl_info, was_npn_negotiated, protocol_negotiated);
 679 }
 680
 681 bool SpdyStream::GetSSLCertRequestInfo(SSLCertRequestInfo* cert_request_info) {
 682   return session_->GetSSLCertRequestInfo(cert_request_info);
 683 }
 684
 685 void SpdyStream::PossiblyResumeIfSendStalled() {
 686   DCHECK(!closed());
 687
 688   if (send_stalled_by_flow_control_ && !session_->IsSendStalled() &&
 689       send_window_size_ > 0) {
 690     net_log_.AddEvent(
 691         NetLog::TYPE_SPDY_STREAM_FLOW_CONTROL_UNSTALLED,
 692         NetLog::IntegerCallback("stream_id", stream_id_));
 693     send_stalled_by_flow_control_ = false;
 694     io_state_ = STATE_SEND_BODY;
 695     DoLoop(OK);
 696   }
 697 }
 698
 699 bool SpdyStream::HasUrl() const {
 700   if (pushed_)
 701     return response_received();
 702   return request_.get() != NULL;
 703 }
 704
 705 GURL SpdyStream::GetUrl() const {
 706   DCHECK(HasUrl());
 707
 708   const SpdyHeaderBlock& headers = (pushed_) ? *response_ : *request_;
 709   return GetUrlFromHeaderBlock(headers, GetProtocolVersion(), pushed_);
 710 }
 711
 712 void SpdyStream::OnGetDomainBoundCertComplete(int result) {
 713   DCHECK_EQ(STATE_GET_DOMAIN_BOUND_CERT_COMPLETE, io_state_);
 714   DoLoop(result);
 715 }
 716
 717 int SpdyStream::DoLoop(int result) {
 718   do {
 719     State state = io_state_;
 720     io_state_ = STATE_NONE;
 721     switch (state) {
 722       // State machine 1: Send headers and body.
 723       case STATE_GET_DOMAIN_BOUND_CERT:
 724         CHECK_EQ(result, OK);
 725         result = DoGetDomainBoundCert();
 726         break;
 727       case STATE_GET_DOMAIN_BOUND_CERT_COMPLETE:
 728         result = DoGetDomainBoundCertComplete(result);
 729         break;
 730       case STATE_SEND_DOMAIN_BOUND_CERT:
 731         CHECK_EQ(result, OK);
 732         result = DoSendDomainBoundCert();
 733         break;
 734       case STATE_SEND_DOMAIN_BOUND_CERT_COMPLETE:
 735         CHECK_EQ(result, OK);
 736         result = DoSendDomainBoundCertComplete();
 737         break;
 738       case STATE_SEND_HEADERS:
 739         CHECK_EQ(result, OK);
 740         result = DoSendHeaders();
 741         break;
 742       case STATE_SEND_HEADERS_COMPLETE:
 743         CHECK_EQ(result, OK);
 744         result = DoSendHeadersComplete();
 745         break;
 746       case STATE_SEND_BODY:
 747         CHECK_EQ(result, OK);
 748         result = DoSendBody();
 749         break;
 750       case STATE_SEND_BODY_COMPLETE:
 751         CHECK_EQ(result, OK);
 752         result = DoSendBodyComplete();
 753         break;
 754       // This is an intermediary waiting state. This state is reached when all
 755       // data has been sent, but no data has been received.
 756       case STATE_WAITING_FOR_RESPONSE:
 757         io_state_ = STATE_WAITING_FOR_RESPONSE;
 758         result = ERR_IO_PENDING;
 759         break;
 760       // State machine 2: connection is established.
 761       // In STATE_OPEN, OnResponseReceived has already been called.
 762       // OnDataReceived, OnClose and OnFrameWriteComplete can be called.
 763       // Only OnFrameWriteComplete calls DoLoop().
 764       //
 765       // For HTTP streams, no data is sent from the client while in the OPEN
 766       // state, so OnFrameWriteComplete is never called here.  The HTTP body is
 767       // handled in the OnDataReceived callback, which does not call into
 768       // DoLoop.
 769       //
 770       // For WebSocket streams, which are bi-directional, we'll send and
 771       // receive data once the connection is established.  Received data is
 772       // handled in OnDataReceived.  Sent data is handled in
 773       // OnFrameWriteComplete, which calls DoOpen().
 774       case STATE_OPEN:
 775         CHECK_EQ(result, OK);
 776         result = DoOpen();
 777         break;
 778
 779       case STATE_DONE:
 780         DCHECK(result != ERR_IO_PENDING);
 781         break;
 782       default:
 783         NOTREACHED() << io_state_;
 784         break;
 785     }
 786   } while (result != ERR_IO_PENDING && io_state_ != STATE_NONE &&
 787            io_state_ != STATE_OPEN);
 788
 789   return result;
 790 }
 791
 792 int SpdyStream::DoGetDomainBoundCert() {
 793   CHECK(request_.get());
 794   if (!session_->NeedsCredentials()) {
 795     // Proceed directly to sending headers
 796     io_state_ = STATE_SEND_HEADERS;
 797     return OK;
 798   }
 799
 800   slot_ = session_->credential_state()->FindCredentialSlot(GetUrl());
 801   if (slot_ != SpdyCredentialState::kNoEntry) {
 802     // Proceed directly to sending headers
 803     io_state_ = STATE_SEND_HEADERS;
 804     return OK;
 805   }
 806
 807   io_state_ = STATE_GET_DOMAIN_BOUND_CERT_COMPLETE;
 808   ServerBoundCertService* sbc_service = session_->GetServerBoundCertService();
 809   DCHECK(sbc_service != NULL);
 810   std::vector<uint8> requested_cert_types;
 811   requested_cert_types.push_back(CLIENT_CERT_ECDSA_SIGN);
 812   int rv = sbc_service->GetDomainBoundCert(
 813       GetUrl().GetOrigin().host(), requested_cert_types,
 814       &domain_bound_cert_type_, &domain_bound_private_key_, &domain_bound_cert_,
 815       base::Bind(&SpdyStream::OnGetDomainBoundCertComplete,
 816                  weak_ptr_factory_.GetWeakPtr()),
 817       &domain_bound_cert_request_handle_);
 818   return rv;
 819 }
 820
 821 int SpdyStream::DoGetDomainBoundCertComplete(int result) {
 822   if (result != OK)
 823     return result;
 824
 825   io_state_ = STATE_SEND_DOMAIN_BOUND_CERT;
 826   slot_ =  session_->credential_state()->SetHasCredential(GetUrl());
 827   return OK;
 828 }
 829
 830 int SpdyStream::DoSendDomainBoundCert() {
 831   io_state_ = STATE_SEND_DOMAIN_BOUND_CERT_COMPLETE;
 832   CHECK(request_.get());
 833
 834   std::string origin = GetUrl().GetOrigin().spec();
 835   DCHECK(origin[origin.length() - 1] == '/');
 836   origin.erase(origin.length() - 1);  // Trim trailing slash.
 837   scoped_ptr<SpdyFrame> frame;
 838   int rv = session_->CreateCredentialFrame(
 839       origin, domain_bound_cert_type_, domain_bound_private_key_,
 840       domain_bound_cert_, priority_, &frame);
 841   if (rv != OK) {
 842     DCHECK_NE(rv, ERR_IO_PENDING);
 843     return rv;
 844   }
 845
 846   DCHECK(frame);
 847   // TODO(akalin): Fix the following race condition:
 848   //
 849   // Since this is decoupled from sending the SYN_STREAM frame, it is
 850   // possible that other domain-bound cert frames will clobber ours
 851   // before our SYN_STREAM frame gets sent. This can be solved by
 852   // immediately enqueueing the SYN_STREAM frame here and adjusting
 853   // the state machine appropriately.
 854   session_->EnqueueStreamWrite(
 855       this, CREDENTIAL,
 856       scoped_ptr<SpdyBufferProducer>(
 857           new SimpleBufferProducer(
 858               scoped_ptr<SpdyBuffer>(new SpdyBuffer(frame.Pass())))));
 859   return ERR_IO_PENDING;
 860 }
 861
 862 int SpdyStream::DoSendDomainBoundCertComplete() {
 863   DCHECK_EQ(just_completed_frame_type_, CREDENTIAL);
 864   io_state_ = STATE_SEND_HEADERS;
 865   return OK;
 866 }
 867
 868 int SpdyStream::DoSendHeaders() {
 869   CHECK(!cancelled_);
 870   io_state_ = STATE_SEND_HEADERS_COMPLETE;
 871
 872   session_->EnqueueStreamWrite(
 873       this, SYN_STREAM,
 874       scoped_ptr<SpdyBufferProducer>(
 875           new SynStreamBufferProducer(weak_ptr_factory_.GetWeakPtr())));
 876   return ERR_IO_PENDING;
 877 }
 878
 879 int SpdyStream::DoSendHeadersComplete() {
 880   DCHECK_EQ(just_completed_frame_type_, SYN_STREAM);
 881   DCHECK_NE(stream_id_, 0u);
 882   if (!delegate_)
 883     return ERR_UNEXPECTED;
 884
 885   io_state_ =
 886       (delegate_->OnSendHeadersComplete() == MORE_DATA_TO_SEND) ?
 887       STATE_SEND_BODY : STATE_WAITING_FOR_RESPONSE;
 888
 889   return OK;
 890 }
 891
 892 // DoSendBody is called to send the optional body for the request.  This call
 893 // will also be called as each write of a chunk of the body completes.
 894 int SpdyStream::DoSendBody() {
 895   // If we're already in the STATE_SEND_BODY state, then we've already
 896   // sent a portion of the body.  In that case, we need to first consume
 897   // the bytes written in the body stream.  Note that the bytes written is
 898   // the number of bytes in the frame that were written, only consume the
 899   // data portion, of course.
 900   io_state_ = STATE_SEND_BODY_COMPLETE;
 901   if (!delegate_)
 902     return ERR_UNEXPECTED;
 903   return delegate_->OnSendBody();
 904 }
 905
 906 int SpdyStream::DoSendBodyComplete() {
 907   if (just_completed_frame_type_ != DATA) {
 908     NOTREACHED();
 909     return ERR_UNEXPECTED;
 910   }
 911
 912   if (just_completed_frame_size_ < session_->GetDataFrameMinimumSize()) {
 913     NOTREACHED();
 914     return ERR_UNEXPECTED;
 915   }
 916
 917   size_t frame_payload_size =
 918       just_completed_frame_size_ - session_->GetDataFrameMinimumSize();
 919   if (frame_payload_size > session_->GetDataFrameMaximumPayload()) {
 920     NOTREACHED();
 921     return ERR_UNEXPECTED;
 922   }
 923
 924   if (!delegate_) {
 925     NOTREACHED();
 926     return ERR_UNEXPECTED;
 927   }
 928
 929   send_bytes_ += frame_payload_size;
 930
 931   io_state_ =
 932       (delegate_->OnSendBodyComplete(frame_payload_size) == MORE_DATA_TO_SEND) ?
 933       STATE_SEND_BODY : STATE_WAITING_FOR_RESPONSE;
 934
 935   return OK;
 936 }
 937
 938 int SpdyStream::DoOpen() {
 939   io_state_ = STATE_OPEN;
 940
 941   switch (just_completed_frame_type_) {
 942     case DATA: {
 943       if (just_completed_frame_size_ < session_->GetDataFrameMinimumSize()) {
 944         NOTREACHED();
 945         return ERR_UNEXPECTED;
 946       }
 947
 948       size_t frame_payload_size =
 949           just_completed_frame_size_ - session_->GetDataFrameMinimumSize();
 950       if (frame_payload_size > session_->GetDataFrameMaximumPayload()) {
 951         NOTREACHED();
 952         return ERR_UNEXPECTED;
 953       }
 954
 955       send_bytes_ += frame_payload_size;
 956       if (delegate_)
 957         delegate_->OnDataSent(frame_payload_size);
 958       break;
 959     }
 960
 961     case HEADERS:
 962       if (delegate_)
 963         delegate_->OnHeadersSent();
 964       break;
 965
 966     default:
 967       NOTREACHED();
 968       return ERR_UNEXPECTED;
 969   }
 970
 971   return OK;
 972 }
 973
 974 void SpdyStream::UpdateHistograms() {
 975   // We need all timers to be filled in, otherwise metrics can be bogus.
 976   if (send_time_.is_null() || recv_first_byte_time_.is_null() ||
 977       recv_last_byte_time_.is_null())
 978     return;
 979
 980   UMA_HISTOGRAM_TIMES("Net.SpdyStreamTimeToFirstByte",
 981       recv_first_byte_time_ - send_time_);
 982   UMA_HISTOGRAM_TIMES("Net.SpdyStreamDownloadTime",
 983       recv_last_byte_time_ - recv_first_byte_time_);
 984   UMA_HISTOGRAM_TIMES("Net.SpdyStreamTime",
 985       recv_last_byte_time_ - send_time_);
 986
 987   UMA_HISTOGRAM_COUNTS("Net.SpdySendBytes", send_bytes_);
 988   UMA_HISTOGRAM_COUNTS("Net.SpdyRecvBytes", recv_bytes_);
 989 }
 990
 991 }  // namespace net