Port Android relocation packer to chromium build
[chromium-blink-merge.git] / net / spdy / spdy_stream.cc
blob85b20165a2ddeed39728008c288dfef67498f473
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.
5 #include "net/spdy/spdy_stream.h"
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"
18 namespace net {
20 namespace {
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;
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;
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;
51 return false;
54 } // namespace
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());
64 ~SynStreamBufferProducer() override {}
66 scoped_ptr<SpdyBuffer> ProduceBuffer() override {
67 if (!stream_.get()) {
68 NOTREACHED();
69 return scoped_ptr<SpdyBuffer>();
71 DCHECK_GT(stream_->stream_id(), 0u);
72 return scoped_ptr<SpdyBuffer>(
73 new SpdyBuffer(stream_->ProduceSynStreamFrame()));
76 private:
77 const base::WeakPtr<SpdyStream> stream_;
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);
115 SpdyStream::~SpdyStream() {
116 CHECK(!write_handler_guard_);
117 UpdateHistograms();
120 void SpdyStream::SetDelegate(Delegate* delegate) {
121 CHECK(!delegate_);
122 CHECK(delegate);
123 delegate_ = delegate;
125 CHECK(io_state_ == STATE_IDLE ||
126 io_state_ == STATE_HALF_CLOSED_LOCAL_UNCLAIMED ||
127 io_state_ == STATE_RESERVED_REMOTE);
129 if (io_state_ == STATE_HALF_CLOSED_LOCAL_UNCLAIMED) {
130 DCHECK_EQ(type_, SPDY_PUSH_STREAM);
131 base::MessageLoop::current()->PostTask(
132 FROM_HERE,
133 base::Bind(&SpdyStream::PushedStreamReplay, GetWeakPtr()));
137 void SpdyStream::PushedStreamReplay() {
138 DCHECK_EQ(type_, SPDY_PUSH_STREAM);
139 DCHECK_NE(stream_id_, 0u);
140 CHECK_EQ(stream_id_ % 2, 0u);
142 CHECK_EQ(io_state_, STATE_HALF_CLOSED_LOCAL_UNCLAIMED);
143 io_state_ = STATE_HALF_CLOSED_LOCAL;
145 // The delegate methods called below may delete |this|, so use
146 // |weak_this| to detect that.
147 base::WeakPtr<SpdyStream> weak_this = GetWeakPtr();
149 CHECK(delegate_);
150 SpdyResponseHeadersStatus status =
151 delegate_->OnResponseHeadersUpdated(response_headers_);
152 if (status == RESPONSE_HEADERS_ARE_INCOMPLETE) {
153 // Since RESPONSE_HEADERS_ARE_INCOMPLETE was returned, we must not
154 // have been closed. Since we don't have complete headers, assume
155 // we're waiting for another HEADERS frame, and we had better not
156 // have any pending data frames.
157 CHECK(weak_this);
158 if (!pending_recv_data_.empty()) {
159 LogStreamError(ERR_SPDY_PROTOCOL_ERROR,
160 "Data received with incomplete headers.");
161 session_->CloseActiveStream(stream_id_, ERR_SPDY_PROTOCOL_ERROR);
163 return;
166 // OnResponseHeadersUpdated() may have closed |this|.
167 if (!weak_this)
168 return;
170 response_headers_status_ = RESPONSE_HEADERS_ARE_COMPLETE;
172 while (!pending_recv_data_.empty()) {
173 // Take ownership of the first element of |pending_recv_data_|.
174 scoped_ptr<SpdyBuffer> buffer(pending_recv_data_.front());
175 pending_recv_data_.weak_erase(pending_recv_data_.begin());
177 bool eof = (buffer == NULL);
179 CHECK(delegate_);
180 delegate_->OnDataReceived(buffer.Pass());
182 // OnDataReceived() may have closed |this|.
183 if (!weak_this)
184 return;
186 if (eof) {
187 DCHECK(pending_recv_data_.empty());
188 session_->CloseActiveStream(stream_id_, OK);
189 DCHECK(!weak_this);
190 // |pending_recv_data_| is invalid at this point.
191 break;
196 scoped_ptr<SpdyFrame> SpdyStream::ProduceSynStreamFrame() {
197 CHECK_EQ(io_state_, STATE_IDLE);
198 CHECK(request_headers_);
199 CHECK_GT(stream_id_, 0u);
201 SpdyControlFlags flags =
202 (pending_send_status_ == NO_MORE_DATA_TO_SEND) ?
203 CONTROL_FLAG_FIN : CONTROL_FLAG_NONE;
204 scoped_ptr<SpdyFrame> frame(session_->CreateSynStream(
205 stream_id_, priority_, flags, *request_headers_));
206 send_time_ = base::TimeTicks::Now();
207 return frame.Pass();
210 void SpdyStream::DetachDelegate() {
211 DCHECK(!IsClosed());
212 delegate_ = NULL;
213 Cancel();
216 void SpdyStream::AdjustSendWindowSize(int32 delta_window_size) {
217 DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
219 if (IsClosed())
220 return;
222 // Check for wraparound.
223 if (send_window_size_ > 0) {
224 DCHECK_LE(delta_window_size, kint32max - send_window_size_);
226 if (send_window_size_ < 0) {
227 DCHECK_GE(delta_window_size, kint32min - send_window_size_);
229 send_window_size_ += delta_window_size;
230 PossiblyResumeIfSendStalled();
233 void SpdyStream::OnWriteBufferConsumed(
234 size_t frame_payload_size,
235 size_t consume_size,
236 SpdyBuffer::ConsumeSource consume_source) {
237 DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
238 if (consume_source == SpdyBuffer::DISCARD) {
239 // If we're discarding a frame or part of it, increase the send
240 // window by the number of discarded bytes. (Although if we're
241 // discarding part of a frame, it's probably because of a write
242 // error and we'll be tearing down the stream soon.)
243 size_t remaining_payload_bytes = std::min(consume_size, frame_payload_size);
244 DCHECK_GT(remaining_payload_bytes, 0u);
245 IncreaseSendWindowSize(static_cast<int32>(remaining_payload_bytes));
247 // For consumed bytes, the send window is increased when we receive
248 // a WINDOW_UPDATE frame.
251 void SpdyStream::IncreaseSendWindowSize(int32 delta_window_size) {
252 DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
253 DCHECK_GE(delta_window_size, 1);
255 // Ignore late WINDOW_UPDATEs.
256 if (IsClosed())
257 return;
259 if (send_window_size_ > 0) {
260 // Check for overflow.
261 int32 max_delta_window_size = kint32max - send_window_size_;
262 if (delta_window_size > max_delta_window_size) {
263 std::string desc = base::StringPrintf(
264 "Received WINDOW_UPDATE [delta: %d] for stream %d overflows "
265 "send_window_size_ [current: %d]", delta_window_size, stream_id_,
266 send_window_size_);
267 session_->ResetStream(stream_id_, RST_STREAM_FLOW_CONTROL_ERROR, desc);
268 return;
272 send_window_size_ += delta_window_size;
274 net_log_.AddEvent(
275 NetLog::TYPE_HTTP2_STREAM_UPDATE_SEND_WINDOW,
276 base::Bind(&NetLogSpdyStreamWindowUpdateCallback, stream_id_,
277 delta_window_size, send_window_size_));
279 PossiblyResumeIfSendStalled();
282 void SpdyStream::DecreaseSendWindowSize(int32 delta_window_size) {
283 DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
285 if (IsClosed())
286 return;
288 // We only call this method when sending a frame. Therefore,
289 // |delta_window_size| should be within the valid frame size range.
290 DCHECK_GE(delta_window_size, 1);
291 DCHECK_LE(delta_window_size, kMaxSpdyFrameChunkSize);
293 // |send_window_size_| should have been at least |delta_window_size| for
294 // this call to happen.
295 DCHECK_GE(send_window_size_, delta_window_size);
297 send_window_size_ -= delta_window_size;
299 net_log_.AddEvent(
300 NetLog::TYPE_HTTP2_STREAM_UPDATE_SEND_WINDOW,
301 base::Bind(&NetLogSpdyStreamWindowUpdateCallback, stream_id_,
302 -delta_window_size, send_window_size_));
305 void SpdyStream::OnReadBufferConsumed(
306 size_t consume_size,
307 SpdyBuffer::ConsumeSource consume_source) {
308 DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
309 DCHECK_GE(consume_size, 1u);
310 DCHECK_LE(consume_size, static_cast<size_t>(kint32max));
311 IncreaseRecvWindowSize(static_cast<int32>(consume_size));
314 void SpdyStream::IncreaseRecvWindowSize(int32 delta_window_size) {
315 DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
317 // By the time a read is processed by the delegate, this stream may
318 // already be inactive.
319 if (!session_->IsStreamActive(stream_id_))
320 return;
322 DCHECK_GE(unacked_recv_window_bytes_, 0);
323 DCHECK_GE(recv_window_size_, unacked_recv_window_bytes_);
324 DCHECK_GE(delta_window_size, 1);
325 // Check for overflow.
326 DCHECK_LE(delta_window_size, kint32max - recv_window_size_);
328 recv_window_size_ += delta_window_size;
329 net_log_.AddEvent(
330 NetLog::TYPE_HTTP2_STREAM_UPDATE_RECV_WINDOW,
331 base::Bind(&NetLogSpdyStreamWindowUpdateCallback, stream_id_,
332 delta_window_size, recv_window_size_));
334 unacked_recv_window_bytes_ += delta_window_size;
335 if (unacked_recv_window_bytes_ >
336 session_->stream_initial_recv_window_size() / 2) {
337 session_->SendStreamWindowUpdate(
338 stream_id_, static_cast<uint32>(unacked_recv_window_bytes_));
339 unacked_recv_window_bytes_ = 0;
343 void SpdyStream::DecreaseRecvWindowSize(int32 delta_window_size) {
344 DCHECK(session_->IsStreamActive(stream_id_));
345 DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
346 DCHECK_GE(delta_window_size, 1);
348 // Since we never decrease the initial receive window size,
349 // |delta_window_size| should never cause |recv_window_size_| to go
350 // negative. If we do, the receive window isn't being respected.
351 if (delta_window_size > recv_window_size_) {
352 session_->ResetStream(
353 stream_id_, RST_STREAM_PROTOCOL_ERROR,
354 "delta_window_size is " + base::IntToString(delta_window_size) +
355 " in DecreaseRecvWindowSize, which is larger than the receive " +
356 "window size of " + base::IntToString(recv_window_size_));
357 return;
360 recv_window_size_ -= delta_window_size;
361 net_log_.AddEvent(
362 NetLog::TYPE_HTTP2_STREAM_UPDATE_RECV_WINDOW,
363 base::Bind(&NetLogSpdyStreamWindowUpdateCallback, stream_id_,
364 -delta_window_size, recv_window_size_));
367 int SpdyStream::GetPeerAddress(IPEndPoint* address) const {
368 return session_->GetPeerAddress(address);
371 int SpdyStream::GetLocalAddress(IPEndPoint* address) const {
372 return session_->GetLocalAddress(address);
375 bool SpdyStream::WasEverUsed() const {
376 return session_->WasEverUsed();
379 base::Time SpdyStream::GetRequestTime() const {
380 return request_time_;
383 void SpdyStream::SetRequestTime(base::Time t) {
384 request_time_ = t;
387 int SpdyStream::OnInitialResponseHeadersReceived(
388 const SpdyHeaderBlock& initial_response_headers,
389 base::Time response_time,
390 base::TimeTicks recv_first_byte_time) {
391 // SpdySession guarantees that this is called at most once.
392 CHECK(response_headers_.empty());
394 // Check to make sure that we don't receive the response headers
395 // before we're ready for it.
396 switch (type_) {
397 case SPDY_BIDIRECTIONAL_STREAM:
398 // For a bidirectional stream, we're ready for the response
399 // headers once we've finished sending the request headers.
400 if (io_state_ == STATE_IDLE) {
401 session_->ResetStream(stream_id_, RST_STREAM_PROTOCOL_ERROR,
402 "Response received before request sent");
403 return ERR_SPDY_PROTOCOL_ERROR;
405 break;
407 case SPDY_REQUEST_RESPONSE_STREAM:
408 // For a request/response stream, we're ready for the response
409 // headers once we've finished sending the request headers.
410 if (io_state_ == STATE_IDLE) {
411 session_->ResetStream(stream_id_, RST_STREAM_PROTOCOL_ERROR,
412 "Response received before request sent");
413 return ERR_SPDY_PROTOCOL_ERROR;
415 break;
417 case SPDY_PUSH_STREAM:
418 // Push streams transition to a locally half-closed state upon headers.
419 // We must continue to buffer data while waiting for a call to
420 // SetDelegate() (which may not ever happen).
421 CHECK_EQ(io_state_, STATE_RESERVED_REMOTE);
422 if (!delegate_) {
423 io_state_ = STATE_HALF_CLOSED_LOCAL_UNCLAIMED;
424 } else {
425 io_state_ = STATE_HALF_CLOSED_LOCAL;
427 break;
430 metrics_.StartStream();
432 DCHECK_NE(io_state_, STATE_IDLE);
434 response_time_ = response_time;
435 recv_first_byte_time_ = recv_first_byte_time;
436 return MergeWithResponseHeaders(initial_response_headers);
439 int SpdyStream::OnAdditionalResponseHeadersReceived(
440 const SpdyHeaderBlock& additional_response_headers) {
441 if (type_ == SPDY_REQUEST_RESPONSE_STREAM) {
442 session_->ResetStream(
443 stream_id_, RST_STREAM_PROTOCOL_ERROR,
444 "Additional headers received for request/response stream");
445 return ERR_SPDY_PROTOCOL_ERROR;
446 } else if (type_ == SPDY_PUSH_STREAM &&
447 response_headers_status_ == RESPONSE_HEADERS_ARE_COMPLETE) {
448 session_->ResetStream(
449 stream_id_, RST_STREAM_PROTOCOL_ERROR,
450 "Additional headers received for push stream");
451 return ERR_SPDY_PROTOCOL_ERROR;
453 return MergeWithResponseHeaders(additional_response_headers);
456 void SpdyStream::OnPushPromiseHeadersReceived(const SpdyHeaderBlock& headers) {
457 CHECK(!request_headers_.get());
458 CHECK_EQ(io_state_, STATE_IDLE);
459 CHECK_EQ(type_, SPDY_PUSH_STREAM);
460 DCHECK(!delegate_);
462 io_state_ = STATE_RESERVED_REMOTE;
463 request_headers_.reset(new SpdyHeaderBlock(headers));
466 void SpdyStream::OnDataReceived(scoped_ptr<SpdyBuffer> buffer) {
467 DCHECK(session_->IsStreamActive(stream_id_));
469 // If we're still buffering data for a push stream, we will do the
470 // check for data received with incomplete headers in
471 // PushedStreamReplayData().
472 if (io_state_ == STATE_HALF_CLOSED_LOCAL_UNCLAIMED) {
473 DCHECK_EQ(type_, SPDY_PUSH_STREAM);
474 // It should be valid for this to happen in the server push case.
475 // We'll return received data when delegate gets attached to the stream.
476 if (buffer) {
477 pending_recv_data_.push_back(buffer.release());
478 } else {
479 pending_recv_data_.push_back(NULL);
480 metrics_.StopStream();
481 // Note: we leave the stream open in the session until the stream
482 // is claimed.
484 return;
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;
496 CHECK(!IsClosed());
498 if (!buffer) {
499 metrics_.StopStream();
500 if (io_state_ == STATE_OPEN) {
501 io_state_ = STATE_HALF_CLOSED_REMOTE;
502 } else if (io_state_ == STATE_HALF_CLOSED_LOCAL) {
503 io_state_ = STATE_CLOSED;
504 // Deletes |this|.
505 session_->CloseActiveStream(stream_id_, OK);
506 } else {
507 NOTREACHED() << io_state_;
509 return;
512 size_t length = buffer->GetRemainingSize();
513 DCHECK_LE(length, session_->GetDataFrameMaximumPayload());
514 if (session_->flow_control_state() >= SpdySession::FLOW_CONTROL_STREAM) {
515 DecreaseRecvWindowSize(static_cast<int32>(length));
516 buffer->AddConsumeCallback(
517 base::Bind(&SpdyStream::OnReadBufferConsumed, GetWeakPtr()));
520 // Track our bandwidth.
521 metrics_.RecordBytes(length);
522 recv_bytes_ += length;
523 recv_last_byte_time_ = base::TimeTicks::Now();
525 // May close |this|.
526 delegate_->OnDataReceived(buffer.Pass());
529 void SpdyStream::OnPaddingConsumed(size_t len) {
530 if (session_->flow_control_state() >= SpdySession::FLOW_CONTROL_STREAM) {
531 // Decrease window size because padding bytes are received.
532 // Increase window size because padding bytes are consumed (by discarding).
533 // Net result: |session_unacked_recv_window_bytes_| increases by |len|,
534 // |session_recv_window_size_| does not change.
535 DecreaseRecvWindowSize(static_cast<int32>(len));
536 IncreaseRecvWindowSize(static_cast<int32>(len));
540 void SpdyStream::OnFrameWriteComplete(SpdyFrameType frame_type,
541 size_t frame_size) {
542 DCHECK_NE(type_, SPDY_PUSH_STREAM);
544 if (frame_size < session_->GetFrameMinimumSize() ||
545 frame_size > session_->GetFrameMaximumSize()) {
546 NOTREACHED();
547 return;
549 CHECK(frame_type == SYN_STREAM ||
550 frame_type == DATA) << frame_type;
552 int result = (frame_type == SYN_STREAM) ?
553 OnRequestHeadersSent() : OnDataSent(frame_size);
554 if (result == ERR_IO_PENDING) {
555 // The write operation hasn't completed yet.
556 return;
559 if (pending_send_status_ == NO_MORE_DATA_TO_SEND) {
560 if(io_state_ == STATE_OPEN) {
561 io_state_ = STATE_HALF_CLOSED_LOCAL;
562 } else if(io_state_ == STATE_HALF_CLOSED_REMOTE) {
563 io_state_ = STATE_CLOSED;
564 } else {
565 NOTREACHED() << io_state_;
568 // Notify delegate of write completion. Must not destroy |this|.
569 CHECK(delegate_);
571 base::WeakPtr<SpdyStream> weak_this = GetWeakPtr();
572 write_handler_guard_ = true;
573 if (frame_type == SYN_STREAM) {
574 delegate_->OnRequestHeadersSent();
575 } else {
576 delegate_->OnDataSent();
578 CHECK(weak_this);
579 write_handler_guard_ = false;
582 if (io_state_ == STATE_CLOSED) {
583 // Deletes |this|.
584 session_->CloseActiveStream(stream_id_, OK);
588 int SpdyStream::OnRequestHeadersSent() {
589 CHECK_EQ(io_state_, STATE_IDLE);
590 CHECK_NE(stream_id_, 0u);
592 io_state_ = STATE_OPEN;
593 return OK;
596 int SpdyStream::OnDataSent(size_t frame_size) {
597 CHECK(io_state_ == STATE_OPEN ||
598 io_state_ == STATE_HALF_CLOSED_REMOTE) << io_state_;
600 size_t frame_payload_size = frame_size - session_->GetDataFrameMinimumSize();
602 CHECK_GE(frame_size, session_->GetDataFrameMinimumSize());
603 CHECK_LE(frame_payload_size, session_->GetDataFrameMaximumPayload());
605 send_bytes_ += frame_payload_size;
607 // If more data is available to send, dispatch it and
608 // return that the write operation is still ongoing.
609 pending_send_data_->DidConsume(frame_payload_size);
610 if (pending_send_data_->BytesRemaining() > 0) {
611 QueueNextDataFrame();
612 return ERR_IO_PENDING;
613 } else {
614 pending_send_data_ = NULL;
615 return OK;
619 SpdyMajorVersion SpdyStream::GetProtocolVersion() const {
620 return session_->GetProtocolVersion();
623 void SpdyStream::LogStreamError(int status, const std::string& description) {
624 net_log_.AddEvent(NetLog::TYPE_HTTP2_STREAM_ERROR,
625 base::Bind(&NetLogSpdyStreamErrorCallback, stream_id_,
626 status, &description));
629 void SpdyStream::OnClose(int status) {
630 // In most cases, the stream should already be CLOSED. The exception is when a
631 // SpdySession is shutting down while the stream is in an intermediate state.
632 io_state_ = STATE_CLOSED;
633 response_status_ = status;
634 Delegate* delegate = delegate_;
635 delegate_ = NULL;
636 if (delegate)
637 delegate->OnClose(status);
638 // Unset |stream_id_| last so that the delegate can look it up.
639 stream_id_ = 0;
642 void SpdyStream::Cancel() {
643 // We may be called again from a delegate's OnClose().
644 if (io_state_ == STATE_CLOSED)
645 return;
647 if (stream_id_ != 0) {
648 session_->ResetStream(stream_id_, RST_STREAM_CANCEL, std::string());
649 } else {
650 session_->CloseCreatedStream(GetWeakPtr(), RST_STREAM_CANCEL);
652 // |this| is invalid at this point.
655 void SpdyStream::Close() {
656 // We may be called again from a delegate's OnClose().
657 if (io_state_ == STATE_CLOSED)
658 return;
660 if (stream_id_ != 0) {
661 session_->CloseActiveStream(stream_id_, OK);
662 } else {
663 session_->CloseCreatedStream(GetWeakPtr(), OK);
665 // |this| is invalid at this point.
668 base::WeakPtr<SpdyStream> SpdyStream::GetWeakPtr() {
669 return weak_ptr_factory_.GetWeakPtr();
672 int SpdyStream::SendRequestHeaders(scoped_ptr<SpdyHeaderBlock> request_headers,
673 SpdySendStatus send_status) {
674 CHECK_NE(type_, SPDY_PUSH_STREAM);
675 CHECK_EQ(pending_send_status_, MORE_DATA_TO_SEND);
676 CHECK(!request_headers_);
677 CHECK(!pending_send_data_.get());
678 CHECK_EQ(io_state_, STATE_IDLE);
679 request_headers_ = request_headers.Pass();
680 pending_send_status_ = send_status;
681 session_->EnqueueStreamWrite(
682 GetWeakPtr(), SYN_STREAM,
683 scoped_ptr<SpdyBufferProducer>(
684 new SynStreamBufferProducer(GetWeakPtr())));
685 return ERR_IO_PENDING;
688 void SpdyStream::SendData(IOBuffer* data,
689 int length,
690 SpdySendStatus send_status) {
691 CHECK_NE(type_, SPDY_PUSH_STREAM);
692 CHECK_EQ(pending_send_status_, MORE_DATA_TO_SEND);
693 CHECK(io_state_ == STATE_OPEN ||
694 io_state_ == STATE_HALF_CLOSED_REMOTE) << io_state_;
695 CHECK(!pending_send_data_.get());
696 pending_send_data_ = new DrainableIOBuffer(data, length);
697 pending_send_status_ = send_status;
698 QueueNextDataFrame();
701 bool SpdyStream::GetSSLInfo(SSLInfo* ssl_info,
702 bool* was_npn_negotiated,
703 NextProto* protocol_negotiated) {
704 return session_->GetSSLInfo(
705 ssl_info, was_npn_negotiated, protocol_negotiated);
708 bool SpdyStream::GetSSLCertRequestInfo(SSLCertRequestInfo* cert_request_info) {
709 return session_->GetSSLCertRequestInfo(cert_request_info);
712 void SpdyStream::PossiblyResumeIfSendStalled() {
713 if (IsLocallyClosed()) {
714 return;
716 if (send_stalled_by_flow_control_ && !session_->IsSendStalled() &&
717 send_window_size_ > 0) {
718 net_log_.AddEvent(NetLog::TYPE_HTTP2_STREAM_FLOW_CONTROL_UNSTALLED,
719 NetLog::IntegerCallback("stream_id", stream_id_));
720 send_stalled_by_flow_control_ = false;
721 QueueNextDataFrame();
725 bool SpdyStream::IsClosed() const {
726 return io_state_ == STATE_CLOSED;
729 bool SpdyStream::IsLocallyClosed() const {
730 return io_state_ == STATE_HALF_CLOSED_LOCAL_UNCLAIMED ||
731 io_state_ == STATE_HALF_CLOSED_LOCAL ||
732 io_state_ == STATE_CLOSED;
735 bool SpdyStream::IsIdle() const {
736 return io_state_ == STATE_IDLE;
739 bool SpdyStream::IsOpen() const {
740 return io_state_ == STATE_OPEN;
743 bool SpdyStream::IsReservedRemote() const {
744 return io_state_ == STATE_RESERVED_REMOTE;
747 NextProto SpdyStream::GetProtocol() const {
748 return session_->protocol();
751 bool SpdyStream::GetLoadTimingInfo(LoadTimingInfo* load_timing_info) const {
752 if (stream_id_ == 0)
753 return false;
755 return session_->GetLoadTimingInfo(stream_id_, load_timing_info);
758 GURL SpdyStream::GetUrlFromHeaders() const {
759 if (!request_headers_)
760 return GURL();
762 return GetUrlFromHeaderBlock(
763 *request_headers_, GetProtocolVersion(), type_ == SPDY_PUSH_STREAM);
766 bool SpdyStream::HasUrlFromHeaders() const {
767 return !GetUrlFromHeaders().is_empty();
770 void SpdyStream::UpdateHistograms() {
771 // We need at least the receive timers to be filled in, as otherwise
772 // metrics can be bogus.
773 if (recv_first_byte_time_.is_null() || recv_last_byte_time_.is_null())
774 return;
776 base::TimeTicks effective_send_time;
777 if (type_ == SPDY_PUSH_STREAM) {
778 // Push streams shouldn't have |send_time_| filled in.
779 DCHECK(send_time_.is_null());
780 effective_send_time = recv_first_byte_time_;
781 } else {
782 // For non-push streams, we also need |send_time_| to be filled
783 // in.
784 if (send_time_.is_null())
785 return;
786 effective_send_time = send_time_;
789 UMA_HISTOGRAM_TIMES("Net.SpdyStreamTimeToFirstByte",
790 recv_first_byte_time_ - effective_send_time);
791 UMA_HISTOGRAM_TIMES("Net.SpdyStreamDownloadTime",
792 recv_last_byte_time_ - recv_first_byte_time_);
793 UMA_HISTOGRAM_TIMES("Net.SpdyStreamTime",
794 recv_last_byte_time_ - effective_send_time);
796 UMA_HISTOGRAM_COUNTS("Net.SpdySendBytes", send_bytes_);
797 UMA_HISTOGRAM_COUNTS("Net.SpdyRecvBytes", recv_bytes_);
800 void SpdyStream::QueueNextDataFrame() {
801 // Until the request has been completely sent, we cannot be sure
802 // that our stream_id is correct.
803 CHECK(io_state_ == STATE_OPEN ||
804 io_state_ == STATE_HALF_CLOSED_REMOTE) << io_state_;
805 CHECK_GT(stream_id_, 0u);
806 CHECK(pending_send_data_.get());
807 // Only the final fame may have a length of 0.
808 if (pending_send_status_ == NO_MORE_DATA_TO_SEND) {
809 CHECK_GE(pending_send_data_->BytesRemaining(), 0);
810 } else {
811 CHECK_GT(pending_send_data_->BytesRemaining(), 0);
814 SpdyDataFlags flags =
815 (pending_send_status_ == NO_MORE_DATA_TO_SEND) ?
816 DATA_FLAG_FIN : DATA_FLAG_NONE;
817 scoped_ptr<SpdyBuffer> data_buffer(
818 session_->CreateDataBuffer(stream_id_,
819 pending_send_data_.get(),
820 pending_send_data_->BytesRemaining(),
821 flags));
822 // We'll get called again by PossiblyResumeIfSendStalled().
823 if (!data_buffer)
824 return;
826 if (session_->flow_control_state() >= SpdySession::FLOW_CONTROL_STREAM) {
827 DCHECK_GE(data_buffer->GetRemainingSize(),
828 session_->GetDataFrameMinimumSize());
829 size_t payload_size =
830 data_buffer->GetRemainingSize() - session_->GetDataFrameMinimumSize();
831 DCHECK_LE(payload_size, session_->GetDataFrameMaximumPayload());
833 // Send window size is based on payload size, so nothing to do if this is
834 // just a FIN with no payload.
835 if (payload_size != 0) {
836 DecreaseSendWindowSize(static_cast<int32>(payload_size));
837 // This currently isn't strictly needed, since write frames are
838 // discarded only if the stream is about to be closed. But have it
839 // here anyway just in case this changes.
840 data_buffer->AddConsumeCallback(
841 base::Bind(&SpdyStream::OnWriteBufferConsumed,
842 GetWeakPtr(), payload_size));
846 session_->EnqueueStreamWrite(
847 GetWeakPtr(), DATA,
848 scoped_ptr<SpdyBufferProducer>(
849 new SimpleBufferProducer(data_buffer.Pass())));
852 int SpdyStream::MergeWithResponseHeaders(
853 const SpdyHeaderBlock& new_response_headers) {
854 if (new_response_headers.find("transfer-encoding") !=
855 new_response_headers.end()) {
856 session_->ResetStream(stream_id_, RST_STREAM_PROTOCOL_ERROR,
857 "Received transfer-encoding header");
858 return ERR_SPDY_PROTOCOL_ERROR;
861 for (SpdyHeaderBlock::const_iterator it = new_response_headers.begin();
862 it != new_response_headers.end(); ++it) {
863 // Disallow uppercase headers.
864 if (ContainsUppercaseAscii(it->first)) {
865 session_->ResetStream(stream_id_, RST_STREAM_PROTOCOL_ERROR,
866 "Upper case characters in header: " + it->first);
867 return ERR_SPDY_PROTOCOL_ERROR;
870 SpdyHeaderBlock::iterator it2 = response_headers_.lower_bound(it->first);
871 // Disallow duplicate headers. This is just to be conservative.
872 if (it2 != response_headers_.end() && it2->first == it->first) {
873 session_->ResetStream(stream_id_, RST_STREAM_PROTOCOL_ERROR,
874 "Duplicate header: " + it->first);
875 return ERR_SPDY_PROTOCOL_ERROR;
878 response_headers_.insert(it2, *it);
881 // If delegate_ is not yet attached, we'll call
882 // OnResponseHeadersUpdated() after the delegate gets attached to
883 // the stream.
884 if (delegate_) {
885 // The call to OnResponseHeadersUpdated() below may delete |this|,
886 // so use |weak_this| to detect that.
887 base::WeakPtr<SpdyStream> weak_this = GetWeakPtr();
889 SpdyResponseHeadersStatus status =
890 delegate_->OnResponseHeadersUpdated(response_headers_);
891 if (status == RESPONSE_HEADERS_ARE_INCOMPLETE) {
892 // Since RESPONSE_HEADERS_ARE_INCOMPLETE was returned, we must not
893 // have been closed.
894 CHECK(weak_this);
895 // Incomplete headers are OK only for push streams.
896 if (type_ != SPDY_PUSH_STREAM) {
897 session_->ResetStream(stream_id_, RST_STREAM_PROTOCOL_ERROR,
898 "Incomplete headers");
899 return ERR_INCOMPLETE_SPDY_HEADERS;
901 } else if (weak_this) {
902 response_headers_status_ = RESPONSE_HEADERS_ARE_COMPLETE;
906 return OK;
909 #define STATE_CASE(s) \
910 case s: \
911 description = base::StringPrintf("%s (0x%08X)", #s, s); \
912 break
914 std::string SpdyStream::DescribeState(State state) {
915 std::string description;
916 switch (state) {
917 STATE_CASE(STATE_IDLE);
918 STATE_CASE(STATE_OPEN);
919 STATE_CASE(STATE_HALF_CLOSED_LOCAL_UNCLAIMED);
920 STATE_CASE(STATE_HALF_CLOSED_LOCAL);
921 STATE_CASE(STATE_CLOSED);
922 default:
923 description = base::StringPrintf("Unknown state 0x%08X (%u)", state,
924 state);
925 break;
927 return description;
930 #undef STATE_CASE
932 } // namespace net