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