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"
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"
22 base::Value
* NetLogSpdyStreamErrorCallback(SpdyStreamId stream_id
,
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
);
33 base::Value
* NetLogSpdyStreamWindowUpdateCallback(
34 SpdyStreamId stream_id
,
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
);
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') {
56 // A wrapper around a stream that calls into ProduceSynStreamFrame().
57 class SpdyStream::SynStreamBufferProducer
: public SpdyBufferProducer
{
59 SynStreamBufferProducer(const base::WeakPtr
<SpdyStream
>& stream
)
61 DCHECK(stream_
.get());
64 virtual ~SynStreamBufferProducer() {}
66 virtual scoped_ptr
<SpdyBuffer
> ProduceBuffer() OVERRIDE
{
69 return scoped_ptr
<SpdyBuffer
>();
71 DCHECK_GT(stream_
->stream_id(), 0u);
72 return scoped_ptr
<SpdyBuffer
>(
73 new SpdyBuffer(stream_
->ProduceSynStreamFrame()));
77 const base::WeakPtr
<SpdyStream
> stream_
;
80 SpdyStream::SpdyStream(SpdyStreamType type
,
81 const base::WeakPtr
<SpdySession
>& session
,
83 RequestPriority priority
,
84 int32 initial_send_window_size
,
85 int32 initial_recv_window_size
,
86 const BoundNetLog
& net_log
)
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),
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
),
103 raw_received_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_
);
120 void SpdyStream::SetDelegate(Delegate
* 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(
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();
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.
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
);
166 // OnResponseHeadersUpdated() may have closed |this|.
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
);
180 delegate_
->OnDataReceived(buffer
.Pass());
182 // OnDataReceived() may have closed |this|.
187 DCHECK(pending_recv_data_
.empty());
188 session_
->CloseActiveStream(stream_id_
, OK
);
190 // |pending_recv_data_| is invalid at this point.
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();
210 void SpdyStream::DetachDelegate() {
216 void SpdyStream::AdjustSendWindowSize(int32 delta_window_size
) {
217 DCHECK_GE(session_
->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM
);
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
,
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.
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_
,
267 session_
->ResetStream(stream_id_
, RST_STREAM_FLOW_CONTROL_ERROR
, desc
);
272 send_window_size_
+= delta_window_size
;
275 NetLog::TYPE_SPDY_STREAM_UPDATE_SEND_WINDOW
,
276 base::Bind(&NetLogSpdyStreamWindowUpdateCallback
,
277 stream_id_
, 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
);
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
;
300 NetLog::TYPE_SPDY_STREAM_UPDATE_SEND_WINDOW
,
301 base::Bind(&NetLogSpdyStreamWindowUpdateCallback
,
302 stream_id_
, -delta_window_size
, send_window_size_
));
305 void SpdyStream::OnReadBufferConsumed(
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_
))
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
;
330 NetLog::TYPE_SPDY_STREAM_UPDATE_RECV_WINDOW
,
331 base::Bind(&NetLogSpdyStreamWindowUpdateCallback
,
332 stream_id_
, 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_
));
360 recv_window_size_
-= delta_window_size
;
362 NetLog::TYPE_SPDY_STREAM_UPDATE_RECV_WINDOW
,
363 base::Bind(&NetLogSpdyStreamWindowUpdateCallback
,
364 stream_id_
, -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
) {
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.
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
;
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
;
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
);
423 io_state_
= STATE_HALF_CLOSED_LOCAL_UNCLAIMED
;
425 io_state_
= STATE_HALF_CLOSED_LOCAL
;
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
);
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.
477 pending_recv_data_
.push_back(buffer
.release());
479 pending_recv_data_
.push_back(NULL
);
480 metrics_
.StopStream();
481 // Note: we leave the stream open in the session until the stream
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
);
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
;
505 session_
->CloseActiveStream(stream_id_
, OK
);
507 NOTREACHED() << io_state_
;
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();
526 delegate_
->OnDataReceived(buffer
.Pass());
529 void SpdyStream::OnFrameWriteComplete(SpdyFrameType frame_type
,
531 DCHECK_NE(type_
, SPDY_PUSH_STREAM
);
533 if (frame_size
< session_
->GetFrameMinimumSize() ||
534 frame_size
> session_
->GetFrameMaximumSize()) {
538 CHECK(frame_type
== SYN_STREAM
||
539 frame_type
== DATA
) << frame_type
;
541 int result
= (frame_type
== SYN_STREAM
) ?
542 OnRequestHeadersSent() : OnDataSent(frame_size
);
543 if (result
== ERR_IO_PENDING
) {
544 // The write operation hasn't completed yet.
548 if (pending_send_status_
== NO_MORE_DATA_TO_SEND
) {
549 if(io_state_
== STATE_OPEN
) {
550 io_state_
= STATE_HALF_CLOSED_LOCAL
;
551 } else if(io_state_
== STATE_HALF_CLOSED_REMOTE
) {
552 io_state_
= STATE_CLOSED
;
554 NOTREACHED() << io_state_
;
557 // Notify delegate of write completion. Must not destroy |this|.
560 base::WeakPtr
<SpdyStream
> weak_this
= GetWeakPtr();
561 write_handler_guard_
= true;
562 if (frame_type
== SYN_STREAM
) {
563 delegate_
->OnRequestHeadersSent();
565 delegate_
->OnDataSent();
568 write_handler_guard_
= false;
571 if (io_state_
== STATE_CLOSED
) {
573 session_
->CloseActiveStream(stream_id_
, OK
);
577 int SpdyStream::OnRequestHeadersSent() {
578 CHECK_EQ(io_state_
, STATE_IDLE
);
579 CHECK_NE(stream_id_
, 0u);
581 io_state_
= STATE_OPEN
;
585 int SpdyStream::OnDataSent(size_t frame_size
) {
586 CHECK(io_state_
== STATE_OPEN
||
587 io_state_
== STATE_HALF_CLOSED_REMOTE
) << io_state_
;
589 size_t frame_payload_size
= frame_size
- session_
->GetDataFrameMinimumSize();
591 CHECK_GE(frame_size
, session_
->GetDataFrameMinimumSize());
592 CHECK_LE(frame_payload_size
, session_
->GetDataFrameMaximumPayload());
594 send_bytes_
+= frame_payload_size
;
596 // If more data is available to send, dispatch it and
597 // return that the write operation is still ongoing.
598 pending_send_data_
->DidConsume(frame_payload_size
);
599 if (pending_send_data_
->BytesRemaining() > 0) {
600 QueueNextDataFrame();
601 return ERR_IO_PENDING
;
603 pending_send_data_
= NULL
;
608 SpdyMajorVersion
SpdyStream::GetProtocolVersion() const {
609 return session_
->GetProtocolVersion();
612 void SpdyStream::LogStreamError(int status
, const std::string
& description
) {
613 net_log_
.AddEvent(NetLog::TYPE_SPDY_STREAM_ERROR
,
614 base::Bind(&NetLogSpdyStreamErrorCallback
,
615 stream_id_
, status
, &description
));
618 void SpdyStream::OnClose(int status
) {
619 // In most cases, the stream should already be CLOSED. The exception is when a
620 // SpdySession is shutting down while the stream is in an intermediate state.
621 io_state_
= STATE_CLOSED
;
622 response_status_
= status
;
623 Delegate
* delegate
= delegate_
;
626 delegate
->OnClose(status
);
627 // Unset |stream_id_| last so that the delegate can look it up.
631 void SpdyStream::Cancel() {
632 // We may be called again from a delegate's OnClose().
633 if (io_state_
== STATE_CLOSED
)
636 if (stream_id_
!= 0) {
637 session_
->ResetStream(stream_id_
, RST_STREAM_CANCEL
, std::string());
639 session_
->CloseCreatedStream(GetWeakPtr(), RST_STREAM_CANCEL
);
641 // |this| is invalid at this point.
644 void SpdyStream::Close() {
645 // We may be called again from a delegate's OnClose().
646 if (io_state_
== STATE_CLOSED
)
649 if (stream_id_
!= 0) {
650 session_
->CloseActiveStream(stream_id_
, OK
);
652 session_
->CloseCreatedStream(GetWeakPtr(), OK
);
654 // |this| is invalid at this point.
657 base::WeakPtr
<SpdyStream
> SpdyStream::GetWeakPtr() {
658 return weak_ptr_factory_
.GetWeakPtr();
661 int SpdyStream::SendRequestHeaders(scoped_ptr
<SpdyHeaderBlock
> request_headers
,
662 SpdySendStatus send_status
) {
663 CHECK_NE(type_
, SPDY_PUSH_STREAM
);
664 CHECK_EQ(pending_send_status_
, MORE_DATA_TO_SEND
);
665 CHECK(!request_headers_
);
666 CHECK(!pending_send_data_
.get());
667 CHECK_EQ(io_state_
, STATE_IDLE
);
668 request_headers_
= request_headers
.Pass();
669 pending_send_status_
= send_status
;
670 session_
->EnqueueStreamWrite(
671 GetWeakPtr(), SYN_STREAM
,
672 scoped_ptr
<SpdyBufferProducer
>(
673 new SynStreamBufferProducer(GetWeakPtr())));
674 return ERR_IO_PENDING
;
677 void SpdyStream::SendData(IOBuffer
* data
,
679 SpdySendStatus send_status
) {
680 CHECK_NE(type_
, SPDY_PUSH_STREAM
);
681 CHECK_EQ(pending_send_status_
, MORE_DATA_TO_SEND
);
682 CHECK(io_state_
== STATE_OPEN
||
683 io_state_
== STATE_HALF_CLOSED_REMOTE
) << io_state_
;
684 CHECK(!pending_send_data_
.get());
685 pending_send_data_
= new DrainableIOBuffer(data
, length
);
686 pending_send_status_
= send_status
;
687 QueueNextDataFrame();
690 bool SpdyStream::GetSSLInfo(SSLInfo
* ssl_info
,
691 bool* was_npn_negotiated
,
692 NextProto
* protocol_negotiated
) {
693 return session_
->GetSSLInfo(
694 ssl_info
, was_npn_negotiated
, protocol_negotiated
);
697 bool SpdyStream::GetSSLCertRequestInfo(SSLCertRequestInfo
* cert_request_info
) {
698 return session_
->GetSSLCertRequestInfo(cert_request_info
);
701 void SpdyStream::PossiblyResumeIfSendStalled() {
702 if (IsLocallyClosed()) {
705 if (send_stalled_by_flow_control_
&& !session_
->IsSendStalled() &&
706 send_window_size_
> 0) {
708 NetLog::TYPE_SPDY_STREAM_FLOW_CONTROL_UNSTALLED
,
709 NetLog::IntegerCallback("stream_id", stream_id_
));
710 send_stalled_by_flow_control_
= false;
711 QueueNextDataFrame();
715 bool SpdyStream::IsClosed() const {
716 return io_state_
== STATE_CLOSED
;
719 bool SpdyStream::IsLocallyClosed() const {
720 return io_state_
== STATE_HALF_CLOSED_LOCAL_UNCLAIMED
||
721 io_state_
== STATE_HALF_CLOSED_LOCAL
||
722 io_state_
== STATE_CLOSED
;
725 bool SpdyStream::IsIdle() const {
726 return io_state_
== STATE_IDLE
;
729 bool SpdyStream::IsOpen() const {
730 return io_state_
== STATE_OPEN
;
733 bool SpdyStream::IsReservedRemote() const {
734 return io_state_
== STATE_RESERVED_REMOTE
;
737 NextProto
SpdyStream::GetProtocol() const {
738 return session_
->protocol();
741 bool SpdyStream::GetLoadTimingInfo(LoadTimingInfo
* load_timing_info
) const {
745 return session_
->GetLoadTimingInfo(stream_id_
, load_timing_info
);
748 GURL
SpdyStream::GetUrlFromHeaders() const {
749 if (!request_headers_
)
752 return GetUrlFromHeaderBlock(
753 *request_headers_
, GetProtocolVersion(), type_
== SPDY_PUSH_STREAM
);
756 bool SpdyStream::HasUrlFromHeaders() const {
757 return !GetUrlFromHeaders().is_empty();
760 void SpdyStream::UpdateHistograms() {
761 // We need at least the receive timers to be filled in, as otherwise
762 // metrics can be bogus.
763 if (recv_first_byte_time_
.is_null() || recv_last_byte_time_
.is_null())
766 base::TimeTicks effective_send_time
;
767 if (type_
== SPDY_PUSH_STREAM
) {
768 // Push streams shouldn't have |send_time_| filled in.
769 DCHECK(send_time_
.is_null());
770 effective_send_time
= recv_first_byte_time_
;
772 // For non-push streams, we also need |send_time_| to be filled
774 if (send_time_
.is_null())
776 effective_send_time
= send_time_
;
779 UMA_HISTOGRAM_TIMES("Net.SpdyStreamTimeToFirstByte",
780 recv_first_byte_time_
- effective_send_time
);
781 UMA_HISTOGRAM_TIMES("Net.SpdyStreamDownloadTime",
782 recv_last_byte_time_
- recv_first_byte_time_
);
783 UMA_HISTOGRAM_TIMES("Net.SpdyStreamTime",
784 recv_last_byte_time_
- effective_send_time
);
786 UMA_HISTOGRAM_COUNTS("Net.SpdySendBytes", send_bytes_
);
787 UMA_HISTOGRAM_COUNTS("Net.SpdyRecvBytes", recv_bytes_
);
790 void SpdyStream::QueueNextDataFrame() {
791 // Until the request has been completely sent, we cannot be sure
792 // that our stream_id is correct.
793 CHECK(io_state_
== STATE_OPEN
||
794 io_state_
== STATE_HALF_CLOSED_REMOTE
) << io_state_
;
795 CHECK_GT(stream_id_
, 0u);
796 CHECK(pending_send_data_
.get());
797 // Only the final fame may have a length of 0.
798 if (pending_send_status_
== NO_MORE_DATA_TO_SEND
) {
799 CHECK_GE(pending_send_data_
->BytesRemaining(), 0);
801 CHECK_GT(pending_send_data_
->BytesRemaining(), 0);
804 SpdyDataFlags flags
=
805 (pending_send_status_
== NO_MORE_DATA_TO_SEND
) ?
806 DATA_FLAG_FIN
: DATA_FLAG_NONE
;
807 scoped_ptr
<SpdyBuffer
> data_buffer(
808 session_
->CreateDataBuffer(stream_id_
,
809 pending_send_data_
.get(),
810 pending_send_data_
->BytesRemaining(),
812 // We'll get called again by PossiblyResumeIfSendStalled().
816 if (session_
->flow_control_state() >= SpdySession::FLOW_CONTROL_STREAM
) {
817 DCHECK_GE(data_buffer
->GetRemainingSize(),
818 session_
->GetDataFrameMinimumSize());
819 size_t payload_size
=
820 data_buffer
->GetRemainingSize() - session_
->GetDataFrameMinimumSize();
821 DCHECK_LE(payload_size
, session_
->GetDataFrameMaximumPayload());
823 // Send window size is based on payload size, so nothing to do if this is
824 // just a FIN with no payload.
825 if (payload_size
!= 0) {
826 DecreaseSendWindowSize(static_cast<int32
>(payload_size
));
827 // This currently isn't strictly needed, since write frames are
828 // discarded only if the stream is about to be closed. But have it
829 // here anyway just in case this changes.
830 data_buffer
->AddConsumeCallback(
831 base::Bind(&SpdyStream::OnWriteBufferConsumed
,
832 GetWeakPtr(), payload_size
));
836 session_
->EnqueueStreamWrite(
838 scoped_ptr
<SpdyBufferProducer
>(
839 new SimpleBufferProducer(data_buffer
.Pass())));
842 int SpdyStream::MergeWithResponseHeaders(
843 const SpdyHeaderBlock
& new_response_headers
) {
844 if (new_response_headers
.find("transfer-encoding") !=
845 new_response_headers
.end()) {
846 session_
->ResetStream(stream_id_
, RST_STREAM_PROTOCOL_ERROR
,
847 "Received transfer-encoding header");
848 return ERR_SPDY_PROTOCOL_ERROR
;
851 for (SpdyHeaderBlock::const_iterator it
= new_response_headers
.begin();
852 it
!= new_response_headers
.end(); ++it
) {
853 // Disallow uppercase headers.
854 if (ContainsUppercaseAscii(it
->first
)) {
855 session_
->ResetStream(stream_id_
, RST_STREAM_PROTOCOL_ERROR
,
856 "Upper case characters in header: " + it
->first
);
857 return ERR_SPDY_PROTOCOL_ERROR
;
860 SpdyHeaderBlock::iterator it2
= response_headers_
.lower_bound(it
->first
);
861 // Disallow duplicate headers. This is just to be conservative.
862 if (it2
!= response_headers_
.end() && it2
->first
== it
->first
) {
863 session_
->ResetStream(stream_id_
, RST_STREAM_PROTOCOL_ERROR
,
864 "Duplicate header: " + it
->first
);
865 return ERR_SPDY_PROTOCOL_ERROR
;
868 response_headers_
.insert(it2
, *it
);
871 // If delegate_ is not yet attached, we'll call
872 // OnResponseHeadersUpdated() after the delegate gets attached to
875 // The call to OnResponseHeadersUpdated() below may delete |this|,
876 // so use |weak_this| to detect that.
877 base::WeakPtr
<SpdyStream
> weak_this
= GetWeakPtr();
879 SpdyResponseHeadersStatus status
=
880 delegate_
->OnResponseHeadersUpdated(response_headers_
);
881 if (status
== RESPONSE_HEADERS_ARE_INCOMPLETE
) {
882 // Since RESPONSE_HEADERS_ARE_INCOMPLETE was returned, we must not
885 // Incomplete headers are OK only for push streams.
886 if (type_
!= SPDY_PUSH_STREAM
) {
887 session_
->ResetStream(stream_id_
, RST_STREAM_PROTOCOL_ERROR
,
888 "Incomplete headers");
889 return ERR_INCOMPLETE_SPDY_HEADERS
;
891 } else if (weak_this
) {
892 response_headers_status_
= RESPONSE_HEADERS_ARE_COMPLETE
;
899 #define STATE_CASE(s) \
901 description = base::StringPrintf("%s (0x%08X)", #s, s); \
904 std::string
SpdyStream::DescribeState(State state
) {
905 std::string description
;
907 STATE_CASE(STATE_IDLE
);
908 STATE_CASE(STATE_OPEN
);
909 STATE_CASE(STATE_HALF_CLOSED_LOCAL_UNCLAIMED
);
910 STATE_CASE(STATE_HALF_CLOSED_LOCAL
);
911 STATE_CASE(STATE_CLOSED
);
913 description
= base::StringPrintf("Unknown state 0x%08X (%u)", state
,