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