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