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 "base/bind.h"
   8 #include "base/compiler_specific.h"
   9 #include "base/logging.h"
  10 #include "base/message_loop/message_loop.h"
  11 #include "base/metrics/histogram_macros.h"
  12 #include "base/strings/string_number_conversions.h"
  13 #include "base/strings/stringprintf.h"
  14 #include "base/values.h"
  15 #include "net/spdy/spdy_buffer_producer.h"
  16 #include "net/spdy/spdy_http_utils.h"
  17 #include "net/spdy/spdy_session.h"
  18
  19 namespace net {
  20
  21 namespace {
  22
  23 base::Value* NetLogSpdyStreamErrorCallback(
  24     SpdyStreamId stream_id,
  25     int status,
  26     const std::string* description,
  27     NetLogCaptureMode /* capture_mode */) {
  28   base::DictionaryValue* dict = new base::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 base::Value* NetLogSpdyStreamWindowUpdateCallback(
  36     SpdyStreamId stream_id,
  37     int32 delta,
  38     int32 window_size,
  39     NetLogCaptureMode /* capture_mode */) {
  40   base::DictionaryValue* dict = new base::DictionaryValue();
  41   dict->SetInteger("stream_id", stream_id);
  42   dict->SetInteger("delta", delta);
  43   dict->SetInteger("window_size", window_size);
  44   return dict;
  45 }
  46
  47 bool ContainsUppercaseAscii(const std::string& str) {
  48   for (std::string::const_iterator i(str.begin()); i != str.end(); ++i) {
  49     if (*i >= 'A' && *i <= 'Z') {
  50       return true;
  51     }
  52   }
  53   return false;
  54 }
  55
  56 }  // namespace
  57
  58 // A wrapper around a stream that calls into ProduceSynStreamFrame().
  59 class SpdyStream::SynStreamBufferProducer : public SpdyBufferProducer {
  60  public:
  61   SynStreamBufferProducer(const base::WeakPtr<SpdyStream>& stream)
  62       : stream_(stream) {
  63     DCHECK(stream_.get());
  64   }
  65
  66   ~SynStreamBufferProducer() override {}
  67
  68   scoped_ptr<SpdyBuffer> ProduceBuffer() override {
  69     if (!stream_.get()) {
  70       NOTREACHED();
  71       return scoped_ptr<SpdyBuffer>();
  72     }
  73     DCHECK_GT(stream_->stream_id(), 0u);
  74     return scoped_ptr<SpdyBuffer>(
  75         new SpdyBuffer(stream_->ProduceSynStreamFrame()));
  76   }
  77
  78  private:
  79   const base::WeakPtr<SpdyStream> stream_;
  80 };
  81
  82 SpdyStream::SpdyStream(SpdyStreamType type,
  83                        const base::WeakPtr<SpdySession>& session,
  84                        const GURL& url,
  85                        RequestPriority priority,
  86                        int32 initial_send_window_size,
  87                        int32 max_recv_window_size,
  88                        const BoundNetLog& net_log)
  89     : type_(type),
  90       stream_id_(0),
  91       url_(url),
  92       priority_(priority),
  93       send_stalled_by_flow_control_(false),
  94       send_window_size_(initial_send_window_size),
  95       max_recv_window_size_(max_recv_window_size),
  96       recv_window_size_(max_recv_window_size),
  97       unacked_recv_window_bytes_(0),
  98       session_(session),
  99       delegate_(NULL),
 100       pending_send_status_(MORE_DATA_TO_SEND),
 101       request_time_(base::Time::Now()),
 102       response_headers_status_(RESPONSE_HEADERS_ARE_INCOMPLETE),
 103       io_state_(STATE_IDLE),
 104       response_status_(OK),
 105       net_log_(net_log),
 106       raw_received_bytes_(0),
 107       send_bytes_(0),
 108       recv_bytes_(0),
 109       write_handler_guard_(false),
 110       weak_ptr_factory_(this) {
 111   CHECK(type_ == SPDY_BIDIRECTIONAL_STREAM ||
 112         type_ == SPDY_REQUEST_RESPONSE_STREAM ||
 113         type_ == SPDY_PUSH_STREAM);
 114   CHECK_GE(priority_, MINIMUM_PRIORITY);
 115   CHECK_LE(priority_, MAXIMUM_PRIORITY);
 116 }
 117
 118 SpdyStream::~SpdyStream() {
 119   CHECK(!write_handler_guard_);
 120   UpdateHistograms();
 121 }
 122
 123 void SpdyStream::SetDelegate(Delegate* delegate) {
 124   CHECK(!delegate_);
 125   CHECK(delegate);
 126   delegate_ = delegate;
 127
 128   CHECK(io_state_ == STATE_IDLE ||
 129         io_state_ == STATE_HALF_CLOSED_LOCAL_UNCLAIMED ||
 130         io_state_ == STATE_RESERVED_REMOTE);
 131
 132   if (io_state_ == STATE_HALF_CLOSED_LOCAL_UNCLAIMED) {
 133     DCHECK_EQ(type_, SPDY_PUSH_STREAM);
 134     base::MessageLoop::current()->PostTask(
 135         FROM_HERE,
 136         base::Bind(&SpdyStream::PushedStreamReplay, GetWeakPtr()));
 137   }
 138 }
 139
 140 void SpdyStream::PushedStreamReplay() {
 141   DCHECK_EQ(type_, SPDY_PUSH_STREAM);
 142   DCHECK_NE(stream_id_, 0u);
 143   CHECK_EQ(stream_id_ % 2, 0u);
 144
 145   CHECK_EQ(io_state_, STATE_HALF_CLOSED_LOCAL_UNCLAIMED);
 146   io_state_ = STATE_HALF_CLOSED_LOCAL;
 147
 148   // The delegate methods called below may delete |this|, so use
 149   // |weak_this| to detect that.
 150   base::WeakPtr<SpdyStream> weak_this = GetWeakPtr();
 151
 152   CHECK(delegate_);
 153   SpdyResponseHeadersStatus status =
 154       delegate_->OnResponseHeadersUpdated(response_headers_);
 155   if (status == RESPONSE_HEADERS_ARE_INCOMPLETE) {
 156     // Since RESPONSE_HEADERS_ARE_INCOMPLETE was returned, we must not
 157     // have been closed. Since we don't have complete headers, assume
 158     // we're waiting for another HEADERS frame, and we had better not
 159     // have any pending data frames.
 160     CHECK(weak_this);
 161     if (!pending_recv_data_.empty()) {
 162       LogStreamError(ERR_SPDY_PROTOCOL_ERROR,
 163                      "Data received with incomplete headers.");
 164       session_->CloseActiveStream(stream_id_, ERR_SPDY_PROTOCOL_ERROR);
 165     }
 166     return;
 167   }
 168
 169   // OnResponseHeadersUpdated() may have closed |this|.
 170   if (!weak_this)
 171     return;
 172
 173   response_headers_status_ = RESPONSE_HEADERS_ARE_COMPLETE;
 174
 175   while (!pending_recv_data_.empty()) {
 176     // Take ownership of the first element of |pending_recv_data_|.
 177     scoped_ptr<SpdyBuffer> buffer(pending_recv_data_.front());
 178     pending_recv_data_.weak_erase(pending_recv_data_.begin());
 179
 180     bool eof = (buffer == NULL);
 181
 182     CHECK(delegate_);
 183     delegate_->OnDataReceived(buffer.Pass());
 184
 185     // OnDataReceived() may have closed |this|.
 186     if (!weak_this)
 187       return;
 188
 189     if (eof) {
 190       DCHECK(pending_recv_data_.empty());
 191       session_->CloseActiveStream(stream_id_, OK);
 192       DCHECK(!weak_this);
 193       // |pending_recv_data_| is invalid at this point.
 194       break;
 195     }
 196   }
 197 }
 198
 199 scoped_ptr<SpdyFrame> SpdyStream::ProduceSynStreamFrame() {
 200   CHECK_EQ(io_state_, STATE_IDLE);
 201   CHECK(request_headers_);
 202   CHECK_GT(stream_id_, 0u);
 203
 204   SpdyControlFlags flags =
 205       (pending_send_status_ == NO_MORE_DATA_TO_SEND) ?
 206       CONTROL_FLAG_FIN : CONTROL_FLAG_NONE;
 207   scoped_ptr<SpdyFrame> frame(session_->CreateSynStream(
 208       stream_id_, priority_, flags, *request_headers_));
 209   send_time_ = base::TimeTicks::Now();
 210   return frame.Pass();
 211 }
 212
 213 void SpdyStream::DetachDelegate() {
 214   DCHECK(!IsClosed());
 215   delegate_ = NULL;
 216   Cancel();
 217 }
 218
 219 void SpdyStream::AdjustSendWindowSize(int32 delta_window_size) {
 220   DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
 221
 222   if (IsClosed())
 223     return;
 224
 225   // Check for wraparound.
 226   if (send_window_size_ > 0) {
 227     DCHECK_LE(delta_window_size, kint32max - send_window_size_);
 228   }
 229   if (send_window_size_ < 0) {
 230     DCHECK_GE(delta_window_size, kint32min - send_window_size_);
 231   }
 232   send_window_size_ += delta_window_size;
 233   PossiblyResumeIfSendStalled();
 234 }
 235
 236 void SpdyStream::OnWriteBufferConsumed(
 237     size_t frame_payload_size,
 238     size_t consume_size,
 239     SpdyBuffer::ConsumeSource consume_source) {
 240   DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
 241   if (consume_source == SpdyBuffer::DISCARD) {
 242     // If we're discarding a frame or part of it, increase the send
 243     // window by the number of discarded bytes. (Although if we're
 244     // discarding part of a frame, it's probably because of a write
 245     // error and we'll be tearing down the stream soon.)
 246     size_t remaining_payload_bytes = std::min(consume_size, frame_payload_size);
 247     DCHECK_GT(remaining_payload_bytes, 0u);
 248     IncreaseSendWindowSize(static_cast<int32>(remaining_payload_bytes));
 249   }
 250   // For consumed bytes, the send window is increased when we receive
 251   // a WINDOW_UPDATE frame.
 252 }
 253
 254 void SpdyStream::IncreaseSendWindowSize(int32 delta_window_size) {
 255   DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
 256   DCHECK_GE(delta_window_size, 1);
 257
 258   // Ignore late WINDOW_UPDATEs.
 259   if (IsClosed())
 260     return;
 261
 262   if (send_window_size_ > 0) {
 263     // Check for overflow.
 264     int32 max_delta_window_size = kint32max - send_window_size_;
 265     if (delta_window_size > max_delta_window_size) {
 266       std::string desc = base::StringPrintf(
 267           "Received WINDOW_UPDATE [delta: %d] for stream %d overflows "
 268           "send_window_size_ [current: %d]", delta_window_size, stream_id_,
 269           send_window_size_);
 270       session_->ResetStream(stream_id_, RST_STREAM_FLOW_CONTROL_ERROR, desc);
 271       return;
 272     }
 273   }
 274
 275   send_window_size_ += delta_window_size;
 276
 277   net_log_.AddEvent(
 278       NetLog::TYPE_HTTP2_STREAM_UPDATE_SEND_WINDOW,
 279       base::Bind(&NetLogSpdyStreamWindowUpdateCallback, stream_id_,
 280                  delta_window_size, send_window_size_));
 281
 282   PossiblyResumeIfSendStalled();
 283 }
 284
 285 void SpdyStream::DecreaseSendWindowSize(int32 delta_window_size) {
 286   DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
 287
 288   if (IsClosed())
 289     return;
 290
 291   // We only call this method when sending a frame. Therefore,
 292   // |delta_window_size| should be within the valid frame size range.
 293   DCHECK_GE(delta_window_size, 1);
 294   DCHECK_LE(delta_window_size, kMaxSpdyFrameChunkSize);
 295
 296   // |send_window_size_| should have been at least |delta_window_size| for
 297   // this call to happen.
 298   DCHECK_GE(send_window_size_, delta_window_size);
 299
 300   send_window_size_ -= delta_window_size;
 301
 302   net_log_.AddEvent(
 303       NetLog::TYPE_HTTP2_STREAM_UPDATE_SEND_WINDOW,
 304       base::Bind(&NetLogSpdyStreamWindowUpdateCallback, stream_id_,
 305                  -delta_window_size, send_window_size_));
 306 }
 307
 308 void SpdyStream::OnReadBufferConsumed(
 309     size_t consume_size,
 310     SpdyBuffer::ConsumeSource consume_source) {
 311   DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
 312   DCHECK_GE(consume_size, 1u);
 313   DCHECK_LE(consume_size, static_cast<size_t>(kint32max));
 314   IncreaseRecvWindowSize(static_cast<int32>(consume_size));
 315 }
 316
 317 void SpdyStream::IncreaseRecvWindowSize(int32 delta_window_size) {
 318   DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
 319
 320   // By the time a read is processed by the delegate, this stream may
 321   // already be inactive.
 322   if (!session_->IsStreamActive(stream_id_))
 323     return;
 324
 325   DCHECK_GE(unacked_recv_window_bytes_, 0);
 326   DCHECK_GE(recv_window_size_, unacked_recv_window_bytes_);
 327   DCHECK_GE(delta_window_size, 1);
 328   // Check for overflow.
 329   DCHECK_LE(delta_window_size, kint32max - recv_window_size_);
 330
 331   recv_window_size_ += delta_window_size;
 332   net_log_.AddEvent(
 333       NetLog::TYPE_HTTP2_STREAM_UPDATE_RECV_WINDOW,
 334       base::Bind(&NetLogSpdyStreamWindowUpdateCallback, stream_id_,
 335                  delta_window_size, recv_window_size_));
 336
 337   unacked_recv_window_bytes_ += delta_window_size;
 338   if (unacked_recv_window_bytes_ > max_recv_window_size_ / 2) {
 339     session_->SendStreamWindowUpdate(
 340         stream_id_, static_cast<uint32>(unacked_recv_window_bytes_));
 341     unacked_recv_window_bytes_ = 0;
 342   }
 343 }
 344
 345 void SpdyStream::DecreaseRecvWindowSize(int32 delta_window_size) {
 346   DCHECK(session_->IsStreamActive(stream_id_));
 347   DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
 348   DCHECK_GE(delta_window_size, 1);
 349
 350   // Since we never decrease the initial receive window size,
 351   // |delta_window_size| should never cause |recv_window_size_| to go
 352   // negative. If we do, the receive window isn't being respected.
 353   if (delta_window_size > recv_window_size_) {
 354     session_->ResetStream(
 355         stream_id_, RST_STREAM_PROTOCOL_ERROR,
 356         "delta_window_size is " + base::IntToString(delta_window_size) +
 357             " in DecreaseRecvWindowSize, which is larger than the receive " +
 358             "window size of " + base::IntToString(recv_window_size_));
 359     return;
 360   }
 361
 362   recv_window_size_ -= delta_window_size;
 363   net_log_.AddEvent(
 364       NetLog::TYPE_HTTP2_STREAM_UPDATE_RECV_WINDOW,
 365       base::Bind(&NetLogSpdyStreamWindowUpdateCallback, stream_id_,
 366                  -delta_window_size, recv_window_size_));
 367 }
 368
 369 int SpdyStream::GetPeerAddress(IPEndPoint* address) const {
 370   return session_->GetPeerAddress(address);
 371 }
 372
 373 int SpdyStream::GetLocalAddress(IPEndPoint* address) const {
 374   return session_->GetLocalAddress(address);
 375 }
 376
 377 bool SpdyStream::WasEverUsed() const {
 378   return session_->WasEverUsed();
 379 }
 380
 381 base::Time SpdyStream::GetRequestTime() const {
 382   return request_time_;
 383 }
 384
 385 void SpdyStream::SetRequestTime(base::Time t) {
 386   request_time_ = t;
 387 }
 388
 389 int SpdyStream::OnInitialResponseHeadersReceived(
 390     const SpdyHeaderBlock& initial_response_headers,
 391     base::Time response_time,
 392     base::TimeTicks recv_first_byte_time) {
 393   // SpdySession guarantees that this is called at most once.
 394   CHECK(response_headers_.empty());
 395
 396   // Check to make sure that we don't receive the response headers
 397   // before we're ready for it.
 398   switch (type_) {
 399     case SPDY_BIDIRECTIONAL_STREAM:
 400       // For a bidirectional stream, we're ready for the response
 401       // headers once we've finished sending the request headers.
 402       if (io_state_ == STATE_IDLE) {
 403         session_->ResetStream(stream_id_, RST_STREAM_PROTOCOL_ERROR,
 404                               "Response received before request sent");
 405         return ERR_SPDY_PROTOCOL_ERROR;
 406       }
 407       break;
 408
 409     case SPDY_REQUEST_RESPONSE_STREAM:
 410       // For a request/response stream, we're ready for the response
 411       // headers once we've finished sending the request headers.
 412       if (io_state_ == STATE_IDLE) {
 413         session_->ResetStream(stream_id_, RST_STREAM_PROTOCOL_ERROR,
 414                               "Response received before request sent");
 415         return ERR_SPDY_PROTOCOL_ERROR;
 416       }
 417       break;
 418
 419     case SPDY_PUSH_STREAM:
 420       // Push streams transition to a locally half-closed state upon headers.
 421       // We must continue to buffer data while waiting for a call to
 422       // SetDelegate() (which may not ever happen).
 423       CHECK_EQ(io_state_, STATE_RESERVED_REMOTE);
 424       if (!delegate_) {
 425         io_state_ = STATE_HALF_CLOSED_LOCAL_UNCLAIMED;
 426       } else {
 427         io_state_ = STATE_HALF_CLOSED_LOCAL;
 428       }
 429       break;
 430   }
 431
 432   DCHECK_NE(io_state_, STATE_IDLE);
 433
 434   response_time_ = response_time;
 435   recv_first_byte_time_ = recv_first_byte_time;
 436   return MergeWithResponseHeaders(initial_response_headers);
 437 }
 438
 439 int SpdyStream::OnAdditionalResponseHeadersReceived(
 440     const SpdyHeaderBlock& additional_response_headers) {
 441   if (type_ == SPDY_REQUEST_RESPONSE_STREAM) {
 442     session_->ResetStream(
 443         stream_id_, RST_STREAM_PROTOCOL_ERROR,
 444         "Additional headers received for request/response stream");
 445     return ERR_SPDY_PROTOCOL_ERROR;
 446   } else if (type_ == SPDY_PUSH_STREAM &&
 447              response_headers_status_ == RESPONSE_HEADERS_ARE_COMPLETE) {
 448     session_->ResetStream(
 449         stream_id_, RST_STREAM_PROTOCOL_ERROR,
 450         "Additional headers received for push stream");
 451     return ERR_SPDY_PROTOCOL_ERROR;
 452   }
 453   return MergeWithResponseHeaders(additional_response_headers);
 454 }
 455
 456 void SpdyStream::OnPushPromiseHeadersReceived(const SpdyHeaderBlock& headers) {
 457   CHECK(!request_headers_.get());
 458   CHECK_EQ(io_state_, STATE_IDLE);
 459   CHECK_EQ(type_, SPDY_PUSH_STREAM);
 460   DCHECK(!delegate_);
 461
 462   io_state_ = STATE_RESERVED_REMOTE;
 463   request_headers_.reset(new SpdyHeaderBlock(headers));
 464 }
 465
 466 void SpdyStream::OnDataReceived(scoped_ptr<SpdyBuffer> buffer) {
 467   DCHECK(session_->IsStreamActive(stream_id_));
 468
 469   // If we're still buffering data for a push stream, we will do the
 470   // check for data received with incomplete headers in
 471   // PushedStreamReplayData().
 472   if (io_state_ == STATE_HALF_CLOSED_LOCAL_UNCLAIMED) {
 473     DCHECK_EQ(type_, SPDY_PUSH_STREAM);
 474     // It should be valid for this to happen in the server push case.
 475     // We'll return received data when delegate gets attached to the stream.
 476     if (buffer) {
 477       pending_recv_data_.push_back(buffer.release());
 478     } else {
 479       pending_recv_data_.push_back(NULL);
 480       // Note: we leave the stream open in the session until the stream
 481       //       is claimed.
 482     }
 483     return;
 484   }
 485
 486   // If we have response headers but the delegate has indicated that
 487   // it's still incomplete, then that's a protocol error.
 488   if (response_headers_status_ == RESPONSE_HEADERS_ARE_INCOMPLETE) {
 489     LogStreamError(ERR_SPDY_PROTOCOL_ERROR,
 490                    "Data received with incomplete headers.");
 491     session_->CloseActiveStream(stream_id_, ERR_SPDY_PROTOCOL_ERROR);
 492     return;
 493   }
 494
 495   CHECK(!IsClosed());
 496
 497   if (!buffer) {
 498     if (io_state_ == STATE_OPEN) {
 499       io_state_ = STATE_HALF_CLOSED_REMOTE;
 500     } else if (io_state_ == STATE_HALF_CLOSED_LOCAL) {
 501       io_state_ = STATE_CLOSED;
 502       // Deletes |this|.
 503       session_->CloseActiveStream(stream_id_, OK);
 504     } else {
 505       NOTREACHED() << io_state_;
 506     }
 507     return;
 508   }
 509
 510   size_t length = buffer->GetRemainingSize();
 511   DCHECK_LE(length, session_->GetDataFrameMaximumPayload());
 512   if (session_->flow_control_state() >= SpdySession::FLOW_CONTROL_STREAM) {
 513     base::WeakPtr<SpdyStream> weak_this = GetWeakPtr();
 514     // May close the stream.
 515     DecreaseRecvWindowSize(static_cast<int32>(length));
 516     if (!weak_this)
 517       return;
 518     buffer->AddConsumeCallback(
 519         base::Bind(&SpdyStream::OnReadBufferConsumed, GetWeakPtr()));
 520   }
 521
 522   // Track our bandwidth.
 523   recv_bytes_ += length;
 524   recv_last_byte_time_ = base::TimeTicks::Now();
 525
 526   // May close |this|.
 527   delegate_->OnDataReceived(buffer.Pass());
 528 }
 529
 530 void SpdyStream::OnPaddingConsumed(size_t len) {
 531   if (session_->flow_control_state() >= SpdySession::FLOW_CONTROL_STREAM) {
 532     // Decrease window size because padding bytes are received.
 533     // Increase window size because padding bytes are consumed (by discarding).
 534     // Net result: |unacked_recv_window_bytes_| increases by |len|,
 535     // |recv_window_size_| does not change.
 536     base::WeakPtr<SpdyStream> weak_this = GetWeakPtr();
 537     // May close the stream.
 538     DecreaseRecvWindowSize(static_cast<int32>(len));
 539     if (!weak_this)
 540       return;
 541     IncreaseRecvWindowSize(static_cast<int32>(len));
 542   }
 543 }
 544
 545 void SpdyStream::OnFrameWriteComplete(SpdyFrameType frame_type,
 546                                       size_t frame_size) {
 547   DCHECK_NE(type_, SPDY_PUSH_STREAM);
 548
 549   if (frame_size < session_->GetFrameMinimumSize() ||
 550       frame_size > session_->GetFrameMaximumSize()) {
 551     NOTREACHED();
 552     return;
 553   }
 554   CHECK(frame_type == SYN_STREAM ||
 555         frame_type == DATA) << frame_type;
 556
 557   int result = (frame_type == SYN_STREAM) ?
 558       OnRequestHeadersSent() : OnDataSent(frame_size);
 559   if (result == ERR_IO_PENDING) {
 560     // The write operation hasn't completed yet.
 561     return;
 562   }
 563
 564   if (pending_send_status_ == NO_MORE_DATA_TO_SEND) {
 565     if(io_state_ == STATE_OPEN) {
 566       io_state_ = STATE_HALF_CLOSED_LOCAL;
 567     } else if(io_state_ == STATE_HALF_CLOSED_REMOTE) {
 568       io_state_ = STATE_CLOSED;
 569     } else {
 570       NOTREACHED() << io_state_;
 571     }
 572   }
 573   // Notify delegate of write completion. Must not destroy |this|.
 574   CHECK(delegate_);
 575   {
 576     base::WeakPtr<SpdyStream> weak_this = GetWeakPtr();
 577     write_handler_guard_ = true;
 578     if (frame_type == SYN_STREAM) {
 579       delegate_->OnRequestHeadersSent();
 580     } else {
 581       delegate_->OnDataSent();
 582     }
 583     CHECK(weak_this);
 584     write_handler_guard_ = false;
 585   }
 586
 587   if (io_state_ == STATE_CLOSED) {
 588     // Deletes |this|.
 589     session_->CloseActiveStream(stream_id_, OK);
 590   }
 591 }
 592
 593 int SpdyStream::OnRequestHeadersSent() {
 594   CHECK_EQ(io_state_, STATE_IDLE);
 595   CHECK_NE(stream_id_, 0u);
 596
 597   io_state_ = STATE_OPEN;
 598   return OK;
 599 }
 600
 601 int SpdyStream::OnDataSent(size_t frame_size) {
 602   CHECK(io_state_ == STATE_OPEN ||
 603         io_state_ == STATE_HALF_CLOSED_REMOTE) << io_state_;
 604
 605   size_t frame_payload_size = frame_size - session_->GetDataFrameMinimumSize();
 606
 607   CHECK_GE(frame_size, session_->GetDataFrameMinimumSize());
 608   CHECK_LE(frame_payload_size, session_->GetDataFrameMaximumPayload());
 609
 610   send_bytes_ += frame_payload_size;
 611
 612   // If more data is available to send, dispatch it and
 613   // return that the write operation is still ongoing.
 614   pending_send_data_->DidConsume(frame_payload_size);
 615   if (pending_send_data_->BytesRemaining() > 0) {
 616     QueueNextDataFrame();
 617     return ERR_IO_PENDING;
 618   } else {
 619     pending_send_data_ = NULL;
 620     return OK;
 621   }
 622 }
 623
 624 SpdyMajorVersion SpdyStream::GetProtocolVersion() const {
 625   return session_->GetProtocolVersion();
 626 }
 627
 628 void SpdyStream::LogStreamError(int status, const std::string& description) {
 629   net_log_.AddEvent(NetLog::TYPE_HTTP2_STREAM_ERROR,
 630                     base::Bind(&NetLogSpdyStreamErrorCallback, stream_id_,
 631                                status, &description));
 632 }
 633
 634 void SpdyStream::OnClose(int status) {
 635   // In most cases, the stream should already be CLOSED. The exception is when a
 636   // SpdySession is shutting down while the stream is in an intermediate state.
 637   io_state_ = STATE_CLOSED;
 638   response_status_ = status;
 639   Delegate* delegate = delegate_;
 640   delegate_ = NULL;
 641   if (delegate)
 642     delegate->OnClose(status);
 643   // Unset |stream_id_| last so that the delegate can look it up.
 644   stream_id_ = 0;
 645 }
 646
 647 void SpdyStream::Cancel() {
 648   // We may be called again from a delegate's OnClose().
 649   if (io_state_ == STATE_CLOSED)
 650     return;
 651
 652   if (stream_id_ != 0) {
 653     session_->ResetStream(stream_id_, RST_STREAM_CANCEL, std::string());
 654   } else {
 655     session_->CloseCreatedStream(GetWeakPtr(), RST_STREAM_CANCEL);
 656   }
 657   // |this| is invalid at this point.
 658 }
 659
 660 void SpdyStream::Close() {
 661   // We may be called again from a delegate's OnClose().
 662   if (io_state_ == STATE_CLOSED)
 663     return;
 664
 665   if (stream_id_ != 0) {
 666     session_->CloseActiveStream(stream_id_, OK);
 667   } else {
 668     session_->CloseCreatedStream(GetWeakPtr(), OK);
 669   }
 670   // |this| is invalid at this point.
 671 }
 672
 673 base::WeakPtr<SpdyStream> SpdyStream::GetWeakPtr() {
 674   return weak_ptr_factory_.GetWeakPtr();
 675 }
 676
 677 int SpdyStream::SendRequestHeaders(scoped_ptr<SpdyHeaderBlock> request_headers,
 678                                    SpdySendStatus send_status) {
 679   CHECK_NE(type_, SPDY_PUSH_STREAM);
 680   CHECK_EQ(pending_send_status_, MORE_DATA_TO_SEND);
 681   CHECK(!request_headers_);
 682   CHECK(!pending_send_data_.get());
 683   CHECK_EQ(io_state_, STATE_IDLE);
 684   request_headers_ = request_headers.Pass();
 685   pending_send_status_ = send_status;
 686   session_->EnqueueStreamWrite(
 687       GetWeakPtr(), SYN_STREAM,
 688       scoped_ptr<SpdyBufferProducer>(
 689           new SynStreamBufferProducer(GetWeakPtr())));
 690   return ERR_IO_PENDING;
 691 }
 692
 693 void SpdyStream::SendData(IOBuffer* data,
 694                           int length,
 695                           SpdySendStatus send_status) {
 696   CHECK_NE(type_, SPDY_PUSH_STREAM);
 697   CHECK_EQ(pending_send_status_, MORE_DATA_TO_SEND);
 698   CHECK(io_state_ == STATE_OPEN ||
 699         io_state_ == STATE_HALF_CLOSED_REMOTE) << io_state_;
 700   CHECK(!pending_send_data_.get());
 701   pending_send_data_ = new DrainableIOBuffer(data, length);
 702   pending_send_status_ = send_status;
 703   QueueNextDataFrame();
 704 }
 705
 706 bool SpdyStream::GetSSLInfo(SSLInfo* ssl_info,
 707                             bool* was_npn_negotiated,
 708                             NextProto* protocol_negotiated) {
 709   return session_->GetSSLInfo(
 710       ssl_info, was_npn_negotiated, protocol_negotiated);
 711 }
 712
 713 bool SpdyStream::GetSSLCertRequestInfo(SSLCertRequestInfo* cert_request_info) {
 714   return session_->GetSSLCertRequestInfo(cert_request_info);
 715 }
 716
 717 void SpdyStream::PossiblyResumeIfSendStalled() {
 718   if (IsLocallyClosed()) {
 719     return;
 720   }
 721   if (send_stalled_by_flow_control_ && !session_->IsSendStalled() &&
 722       send_window_size_ > 0) {
 723     net_log_.AddEvent(NetLog::TYPE_HTTP2_STREAM_FLOW_CONTROL_UNSTALLED,
 724                       NetLog::IntegerCallback("stream_id", stream_id_));
 725     send_stalled_by_flow_control_ = false;
 726     QueueNextDataFrame();
 727   }
 728 }
 729
 730 bool SpdyStream::IsClosed() const {
 731   return io_state_ == STATE_CLOSED;
 732 }
 733
 734 bool SpdyStream::IsLocallyClosed() const {
 735   return io_state_ == STATE_HALF_CLOSED_LOCAL_UNCLAIMED ||
 736       io_state_ == STATE_HALF_CLOSED_LOCAL ||
 737       io_state_ == STATE_CLOSED;
 738 }
 739
 740 bool SpdyStream::IsIdle() const {
 741   return io_state_ == STATE_IDLE;
 742 }
 743
 744 bool SpdyStream::IsOpen() const {
 745   return io_state_ == STATE_OPEN;
 746 }
 747
 748 bool SpdyStream::IsReservedRemote() const {
 749   return io_state_ == STATE_RESERVED_REMOTE;
 750 }
 751
 752 NextProto SpdyStream::GetProtocol() const {
 753   return session_->protocol();
 754 }
 755
 756 bool SpdyStream::GetLoadTimingInfo(LoadTimingInfo* load_timing_info) const {
 757   if (stream_id_ == 0)
 758     return false;
 759
 760   return session_->GetLoadTimingInfo(stream_id_, load_timing_info);
 761 }
 762
 763 GURL SpdyStream::GetUrlFromHeaders() const {
 764   if (!request_headers_)
 765     return GURL();
 766
 767   return GetUrlFromHeaderBlock(
 768       *request_headers_, GetProtocolVersion(), type_ == SPDY_PUSH_STREAM);
 769 }
 770
 771 bool SpdyStream::HasUrlFromHeaders() const {
 772   return !GetUrlFromHeaders().is_empty();
 773 }
 774
 775 void SpdyStream::UpdateHistograms() {
 776   // We need at least the receive timers to be filled in, as otherwise
 777   // metrics can be bogus.
 778   if (recv_first_byte_time_.is_null() || recv_last_byte_time_.is_null())
 779     return;
 780
 781   base::TimeTicks effective_send_time;
 782   if (type_ == SPDY_PUSH_STREAM) {
 783     // Push streams shouldn't have |send_time_| filled in.
 784     DCHECK(send_time_.is_null());
 785     effective_send_time = recv_first_byte_time_;
 786   } else {
 787     // For non-push streams, we also need |send_time_| to be filled
 788     // in.
 789     if (send_time_.is_null())
 790       return;
 791     effective_send_time = send_time_;
 792   }
 793
 794   UMA_HISTOGRAM_TIMES("Net.SpdyStreamTimeToFirstByte",
 795                       recv_first_byte_time_ - effective_send_time);
 796   UMA_HISTOGRAM_TIMES("Net.SpdyStreamDownloadTime",
 797                       recv_last_byte_time_ - recv_first_byte_time_);
 798   UMA_HISTOGRAM_TIMES("Net.SpdyStreamTime",
 799                       recv_last_byte_time_ - effective_send_time);
 800
 801   UMA_HISTOGRAM_COUNTS("Net.SpdySendBytes", send_bytes_);
 802   UMA_HISTOGRAM_COUNTS("Net.SpdyRecvBytes", recv_bytes_);
 803 }
 804
 805 void SpdyStream::QueueNextDataFrame() {
 806   // Until the request has been completely sent, we cannot be sure
 807   // that our stream_id is correct.
 808   CHECK(io_state_ == STATE_OPEN ||
 809         io_state_ == STATE_HALF_CLOSED_REMOTE) << io_state_;
 810   CHECK_GT(stream_id_, 0u);
 811   CHECK(pending_send_data_.get());
 812   // Only the final fame may have a length of 0.
 813   if (pending_send_status_ == NO_MORE_DATA_TO_SEND) {
 814     CHECK_GE(pending_send_data_->BytesRemaining(), 0);
 815   } else {
 816     CHECK_GT(pending_send_data_->BytesRemaining(), 0);
 817   }
 818
 819   SpdyDataFlags flags =
 820       (pending_send_status_ == NO_MORE_DATA_TO_SEND) ?
 821       DATA_FLAG_FIN : DATA_FLAG_NONE;
 822   scoped_ptr<SpdyBuffer> data_buffer(
 823       session_->CreateDataBuffer(stream_id_,
 824                                  pending_send_data_.get(),
 825                                  pending_send_data_->BytesRemaining(),
 826                                  flags));
 827   // We'll get called again by PossiblyResumeIfSendStalled().
 828   if (!data_buffer)
 829     return;
 830
 831   if (session_->flow_control_state() >= SpdySession::FLOW_CONTROL_STREAM) {
 832     DCHECK_GE(data_buffer->GetRemainingSize(),
 833               session_->GetDataFrameMinimumSize());
 834     size_t payload_size =
 835         data_buffer->GetRemainingSize() - session_->GetDataFrameMinimumSize();
 836     DCHECK_LE(payload_size, session_->GetDataFrameMaximumPayload());
 837
 838     // Send window size is based on payload size, so nothing to do if this is
 839     // just a FIN with no payload.
 840     if (payload_size != 0) {
 841       DecreaseSendWindowSize(static_cast<int32>(payload_size));
 842       // This currently isn't strictly needed, since write frames are
 843       // discarded only if the stream is about to be closed. But have it
 844       // here anyway just in case this changes.
 845       data_buffer->AddConsumeCallback(
 846           base::Bind(&SpdyStream::OnWriteBufferConsumed,
 847                      GetWeakPtr(), payload_size));
 848     }
 849   }
 850
 851   session_->EnqueueStreamWrite(
 852       GetWeakPtr(), DATA,
 853       scoped_ptr<SpdyBufferProducer>(
 854           new SimpleBufferProducer(data_buffer.Pass())));
 855 }
 856
 857 int SpdyStream::MergeWithResponseHeaders(
 858     const SpdyHeaderBlock& new_response_headers) {
 859   if (new_response_headers.find("transfer-encoding") !=
 860       new_response_headers.end()) {
 861     session_->ResetStream(stream_id_, RST_STREAM_PROTOCOL_ERROR,
 862                          "Received transfer-encoding header");
 863     return ERR_SPDY_PROTOCOL_ERROR;
 864   }
 865
 866   for (SpdyHeaderBlock::const_iterator it = new_response_headers.begin();
 867       it != new_response_headers.end(); ++it) {
 868     // Disallow uppercase headers.
 869     if (ContainsUppercaseAscii(it->first)) {
 870       session_->ResetStream(stream_id_, RST_STREAM_PROTOCOL_ERROR,
 871                             "Upper case characters in header: " + it->first);
 872       return ERR_SPDY_PROTOCOL_ERROR;
 873     }
 874
 875     SpdyHeaderBlock::iterator it2 = response_headers_.lower_bound(it->first);
 876     // Disallow duplicate headers.  This is just to be conservative.
 877     if (it2 != response_headers_.end() && it2->first == it->first) {
 878       session_->ResetStream(stream_id_, RST_STREAM_PROTOCOL_ERROR,
 879                             "Duplicate header: " + it->first);
 880       return ERR_SPDY_PROTOCOL_ERROR;
 881     }
 882
 883     response_headers_.insert(it2, *it);
 884   }
 885
 886   // If delegate_ is not yet attached, we'll call
 887   // OnResponseHeadersUpdated() after the delegate gets attached to
 888   // the stream.
 889   if (delegate_) {
 890     // The call to OnResponseHeadersUpdated() below may delete |this|,
 891     // so use |weak_this| to detect that.
 892     base::WeakPtr<SpdyStream> weak_this = GetWeakPtr();
 893
 894     SpdyResponseHeadersStatus status =
 895         delegate_->OnResponseHeadersUpdated(response_headers_);
 896     if (status == RESPONSE_HEADERS_ARE_INCOMPLETE) {
 897       // Since RESPONSE_HEADERS_ARE_INCOMPLETE was returned, we must not
 898       // have been closed.
 899       CHECK(weak_this);
 900       // Incomplete headers are OK only for push streams.
 901       if (type_ != SPDY_PUSH_STREAM) {
 902         session_->ResetStream(stream_id_, RST_STREAM_PROTOCOL_ERROR,
 903                               "Incomplete headers");
 904         return ERR_INCOMPLETE_SPDY_HEADERS;
 905       }
 906     } else if (weak_this) {
 907       response_headers_status_ = RESPONSE_HEADERS_ARE_COMPLETE;
 908     }
 909   }
 910
 911   return OK;
 912 }
 913
 914 #define STATE_CASE(s) \
 915   case s: \
 916     description = base::StringPrintf("%s (0x%08X)", #s, s); \
 917     break
 918
 919 std::string SpdyStream::DescribeState(State state) {
 920   std::string description;
 921   switch (state) {
 922     STATE_CASE(STATE_IDLE);
 923     STATE_CASE(STATE_OPEN);
 924     STATE_CASE(STATE_HALF_CLOSED_LOCAL_UNCLAIMED);
 925     STATE_CASE(STATE_HALF_CLOSED_LOCAL);
 926     STATE_CASE(STATE_CLOSED);
 927     default:
 928       description = base::StringPrintf("Unknown state 0x%08X (%u)", state,
 929                                        state);
 930       break;
 931   }
 932   return description;
 933 }
 934
 935 #undef STATE_CASE
 936
 937 }  // namespace net