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