| // Copyright 2015 The Chromium Authors. All rights reserved. |
| // Use of this source code is governed by a BSD-style license that can be |
| // found in the LICENSE file. |
| |
| #include "net/spdy/bidirectional_stream_spdy_impl.h" |
| |
| #include <utility> |
| |
| #include "base/bind.h" |
| #include "base/location.h" |
| #include "base/logging.h" |
| #include "base/threading/thread_task_runner_handle.h" |
| #include "base/time/time.h" |
| #include "base/timer/timer.h" |
| #include "net/http/bidirectional_stream_request_info.h" |
| #include "net/spdy/spdy_buffer.h" |
| #include "net/spdy/spdy_http_utils.h" |
| #include "net/spdy/spdy_stream.h" |
| #include "net/third_party/quiche/src/spdy/core/spdy_header_block.h" |
| |
| namespace net { |
| |
| namespace { |
| |
| // Time to wait in millisecond to notify |delegate_| of data received. |
| // Handing small chunks of data to the caller creates measurable overhead. |
| // So buffer data in short time-spans and send a single read notification. |
| const int kBufferTimeMs = 1; |
| |
| } // namespace |
| |
| BidirectionalStreamSpdyImpl::BidirectionalStreamSpdyImpl( |
| const base::WeakPtr<SpdySession>& spdy_session, |
| NetLogSource source_dependency) |
| : spdy_session_(spdy_session), |
| request_info_(nullptr), |
| delegate_(nullptr), |
| source_dependency_(source_dependency), |
| negotiated_protocol_(kProtoUnknown), |
| more_read_data_pending_(false), |
| read_buffer_len_(0), |
| written_end_of_stream_(false), |
| write_pending_(false), |
| stream_closed_(false), |
| closed_stream_status_(ERR_FAILED), |
| closed_stream_received_bytes_(0), |
| closed_stream_sent_bytes_(0), |
| closed_has_load_timing_info_(false) {} |
| |
| BidirectionalStreamSpdyImpl::~BidirectionalStreamSpdyImpl() { |
| // Sends a RST to the remote if the stream is destroyed before it completes. |
| ResetStream(); |
| } |
| |
| void BidirectionalStreamSpdyImpl::Start( |
| const BidirectionalStreamRequestInfo* request_info, |
| const NetLogWithSource& net_log, |
| bool /*send_request_headers_automatically*/, |
| BidirectionalStreamImpl::Delegate* delegate, |
| std::unique_ptr<base::OneShotTimer> timer, |
| const NetworkTrafficAnnotationTag& traffic_annotation) { |
| DCHECK(!stream_); |
| DCHECK(timer); |
| |
| delegate_ = delegate; |
| timer_ = std::move(timer); |
| |
| if (!spdy_session_) { |
| base::ThreadTaskRunnerHandle::Get()->PostTask( |
| FROM_HERE, |
| base::BindOnce(&BidirectionalStreamSpdyImpl::NotifyError, |
| weak_factory_.GetWeakPtr(), ERR_CONNECTION_CLOSED)); |
| return; |
| } |
| |
| request_info_ = request_info; |
| |
| int rv = stream_request_.StartRequest( |
| SPDY_BIDIRECTIONAL_STREAM, spdy_session_, request_info_->url, |
| false /* no early data */, request_info_->priority, |
| request_info_->socket_tag, net_log, |
| base::BindOnce(&BidirectionalStreamSpdyImpl::OnStreamInitialized, |
| weak_factory_.GetWeakPtr()), |
| traffic_annotation); |
| if (rv != ERR_IO_PENDING) |
| OnStreamInitialized(rv); |
| } |
| |
| void BidirectionalStreamSpdyImpl::SendRequestHeaders() { |
| // Request headers will be sent automatically. |
| NOTREACHED(); |
| } |
| |
| int BidirectionalStreamSpdyImpl::ReadData(IOBuffer* buf, int buf_len) { |
| if (stream_) |
| DCHECK(!stream_->IsIdle()); |
| |
| DCHECK(buf); |
| DCHECK(buf_len); |
| DCHECK(!timer_->IsRunning()) << "There should be only one ReadData in flight"; |
| |
| // If there is data buffered, complete the IO immediately. |
| if (!read_data_queue_.IsEmpty()) { |
| return read_data_queue_.Dequeue(buf->data(), buf_len); |
| } else if (stream_closed_) { |
| return closed_stream_status_; |
| } |
| // Read will complete asynchronously and Delegate::OnReadCompleted will be |
| // called upon completion. |
| read_buffer_ = buf; |
| read_buffer_len_ = buf_len; |
| return ERR_IO_PENDING; |
| } |
| |
| void BidirectionalStreamSpdyImpl::SendvData( |
| const std::vector<scoped_refptr<IOBuffer>>& buffers, |
| const std::vector<int>& lengths, |
| bool end_stream) { |
| DCHECK_EQ(buffers.size(), lengths.size()); |
| DCHECK(!write_pending_); |
| |
| if (written_end_of_stream_) { |
| LOG(ERROR) << "Writing after end of stream is written."; |
| base::ThreadTaskRunnerHandle::Get()->PostTask( |
| FROM_HERE, base::BindOnce(&BidirectionalStreamSpdyImpl::NotifyError, |
| weak_factory_.GetWeakPtr(), ERR_UNEXPECTED)); |
| return; |
| } |
| |
| write_pending_ = true; |
| written_end_of_stream_ = end_stream; |
| if (MaybeHandleStreamClosedInSendData()) |
| return; |
| |
| DCHECK(!stream_closed_); |
| int total_len = 0; |
| for (int len : lengths) { |
| total_len += len; |
| } |
| |
| if (buffers.size() == 1) { |
| pending_combined_buffer_ = buffers[0]; |
| } else { |
| pending_combined_buffer_ = base::MakeRefCounted<net::IOBuffer>(total_len); |
| int len = 0; |
| // TODO(xunjieli): Get rid of extra copy. Coalesce headers and data frames. |
| for (size_t i = 0; i < buffers.size(); ++i) { |
| memcpy(pending_combined_buffer_->data() + len, buffers[i]->data(), |
| lengths[i]); |
| len += lengths[i]; |
| } |
| } |
| stream_->SendData(pending_combined_buffer_.get(), total_len, |
| end_stream ? NO_MORE_DATA_TO_SEND : MORE_DATA_TO_SEND); |
| } |
| |
| NextProto BidirectionalStreamSpdyImpl::GetProtocol() const { |
| return negotiated_protocol_; |
| } |
| |
| int64_t BidirectionalStreamSpdyImpl::GetTotalReceivedBytes() const { |
| if (stream_closed_) |
| return closed_stream_received_bytes_; |
| |
| if (!stream_) |
| return 0; |
| |
| return stream_->raw_received_bytes(); |
| } |
| |
| int64_t BidirectionalStreamSpdyImpl::GetTotalSentBytes() const { |
| if (stream_closed_) |
| return closed_stream_sent_bytes_; |
| |
| if (!stream_) |
| return 0; |
| |
| return stream_->raw_sent_bytes(); |
| } |
| |
| bool BidirectionalStreamSpdyImpl::GetLoadTimingInfo( |
| LoadTimingInfo* load_timing_info) const { |
| if (stream_closed_) { |
| if (!closed_has_load_timing_info_) |
| return false; |
| *load_timing_info = closed_load_timing_info_; |
| return true; |
| } |
| |
| // If |stream_| isn't created or has ID 0, return false. This is to match |
| // the implementation in SpdyHttpStream. |
| if (!stream_ || stream_->stream_id() == 0) |
| return false; |
| |
| return stream_->GetLoadTimingInfo(load_timing_info); |
| } |
| |
| void BidirectionalStreamSpdyImpl::PopulateNetErrorDetails( |
| NetErrorDetails* details) {} |
| |
| void BidirectionalStreamSpdyImpl::OnHeadersSent() { |
| DCHECK(stream_); |
| |
| negotiated_protocol_ = kProtoHTTP2; |
| if (delegate_) |
| delegate_->OnStreamReady(/*request_headers_sent=*/true); |
| } |
| |
| void BidirectionalStreamSpdyImpl::OnHeadersReceived( |
| const spdy::SpdyHeaderBlock& response_headers, |
| const spdy::SpdyHeaderBlock* pushed_request_headers) { |
| DCHECK(stream_); |
| |
| if (delegate_) |
| delegate_->OnHeadersReceived(response_headers); |
| } |
| |
| void BidirectionalStreamSpdyImpl::OnDataReceived( |
| std::unique_ptr<SpdyBuffer> buffer) { |
| DCHECK(stream_); |
| DCHECK(!stream_closed_); |
| |
| // If |buffer| is null, BidirectionalStreamSpdyImpl::OnClose will be invoked |
| // by SpdyStream to indicate the end of stream. |
| if (!buffer) |
| return; |
| |
| // When buffer is consumed, SpdyStream::OnReadBufferConsumed will adjust |
| // recv window size accordingly. |
| read_data_queue_.Enqueue(std::move(buffer)); |
| if (read_buffer_) { |
| // Handing small chunks of data to the caller creates measurable overhead. |
| // So buffer data in short time-spans and send a single read notification. |
| ScheduleBufferedRead(); |
| } |
| } |
| |
| void BidirectionalStreamSpdyImpl::OnDataSent() { |
| DCHECK(write_pending_); |
| |
| pending_combined_buffer_ = nullptr; |
| write_pending_ = false; |
| |
| if (delegate_) |
| delegate_->OnDataSent(); |
| } |
| |
| void BidirectionalStreamSpdyImpl::OnTrailers( |
| const spdy::SpdyHeaderBlock& trailers) { |
| DCHECK(stream_); |
| DCHECK(!stream_closed_); |
| |
| if (delegate_) |
| delegate_->OnTrailersReceived(trailers); |
| } |
| |
| void BidirectionalStreamSpdyImpl::OnClose(int status) { |
| DCHECK(stream_); |
| |
| stream_closed_ = true; |
| closed_stream_status_ = status; |
| closed_stream_received_bytes_ = stream_->raw_received_bytes(); |
| closed_stream_sent_bytes_ = stream_->raw_sent_bytes(); |
| closed_has_load_timing_info_ = |
| stream_->GetLoadTimingInfo(&closed_load_timing_info_); |
| |
| if (status != OK) { |
| NotifyError(status); |
| return; |
| } |
| ResetStream(); |
| // Complete any remaining read, as all data has been buffered. |
| // If user has not called ReadData (i.e |read_buffer_| is nullptr), this will |
| // do nothing. |
| timer_->Stop(); |
| |
| // |this| might get destroyed after calling into |delegate_| in |
| // DoBufferedRead(). |
| auto weak_this = weak_factory_.GetWeakPtr(); |
| DoBufferedRead(); |
| if (weak_this.get() && write_pending_) |
| OnDataSent(); |
| } |
| |
| bool BidirectionalStreamSpdyImpl::CanGreaseFrameType() const { |
| return false; |
| } |
| |
| NetLogSource BidirectionalStreamSpdyImpl::source_dependency() const { |
| return source_dependency_; |
| } |
| |
| int BidirectionalStreamSpdyImpl::SendRequestHeadersHelper() { |
| spdy::SpdyHeaderBlock headers; |
| HttpRequestInfo http_request_info; |
| http_request_info.url = request_info_->url; |
| http_request_info.method = request_info_->method; |
| http_request_info.extra_headers = request_info_->extra_headers; |
| |
| CreateSpdyHeadersFromHttpRequest(http_request_info, |
| http_request_info.extra_headers, &headers); |
| written_end_of_stream_ = request_info_->end_stream_on_headers; |
| return stream_->SendRequestHeaders(std::move(headers), |
| request_info_->end_stream_on_headers |
| ? NO_MORE_DATA_TO_SEND |
| : MORE_DATA_TO_SEND); |
| } |
| |
| void BidirectionalStreamSpdyImpl::OnStreamInitialized(int rv) { |
| DCHECK_NE(ERR_IO_PENDING, rv); |
| if (rv == OK) { |
| stream_ = stream_request_.ReleaseStream(); |
| stream_->SetDelegate(this); |
| rv = SendRequestHeadersHelper(); |
| if (rv == OK) { |
| OnHeadersSent(); |
| return; |
| } else if (rv == ERR_IO_PENDING) { |
| return; |
| } |
| } |
| NotifyError(rv); |
| } |
| |
| void BidirectionalStreamSpdyImpl::NotifyError(int rv) { |
| ResetStream(); |
| write_pending_ = false; |
| if (delegate_) { |
| BidirectionalStreamImpl::Delegate* delegate = delegate_; |
| delegate_ = nullptr; |
| // Cancel any pending callback. |
| weak_factory_.InvalidateWeakPtrs(); |
| delegate->OnFailed(rv); |
| // |this| can be null when returned from delegate. |
| } |
| } |
| |
| void BidirectionalStreamSpdyImpl::ResetStream() { |
| if (!stream_) |
| return; |
| if (!stream_->IsClosed()) { |
| // This sends a RST to the remote. |
| stream_->DetachDelegate(); |
| DCHECK(!stream_); |
| } else { |
| // Stream is already closed, so it is not legal to call DetachDelegate. |
| stream_.reset(); |
| } |
| } |
| |
| void BidirectionalStreamSpdyImpl::ScheduleBufferedRead() { |
| // If there is already a scheduled DoBufferedRead, don't issue |
| // another one. Mark that we have received more data and return. |
| if (timer_->IsRunning()) { |
| more_read_data_pending_ = true; |
| return; |
| } |
| |
| more_read_data_pending_ = false; |
| timer_->Start(FROM_HERE, base::TimeDelta::FromMilliseconds(kBufferTimeMs), |
| base::BindOnce(&BidirectionalStreamSpdyImpl::DoBufferedRead, |
| weak_factory_.GetWeakPtr())); |
| } |
| |
| void BidirectionalStreamSpdyImpl::DoBufferedRead() { |
| DCHECK(!timer_->IsRunning()); |
| // Check to see that the stream has not errored out. |
| DCHECK(stream_ || stream_closed_); |
| DCHECK(!stream_closed_ || closed_stream_status_ == OK); |
| |
| // When |more_read_data_pending_| is true, it means that more data has arrived |
| // since started waiting. Wait a little longer and continue to buffer. |
| if (more_read_data_pending_ && ShouldWaitForMoreBufferedData()) { |
| ScheduleBufferedRead(); |
| return; |
| } |
| |
| int rv = 0; |
| if (read_buffer_) { |
| rv = ReadData(read_buffer_.get(), read_buffer_len_); |
| DCHECK_NE(ERR_IO_PENDING, rv); |
| read_buffer_ = nullptr; |
| read_buffer_len_ = 0; |
| if (delegate_) |
| delegate_->OnDataRead(rv); |
| } |
| } |
| |
| bool BidirectionalStreamSpdyImpl::ShouldWaitForMoreBufferedData() const { |
| if (stream_closed_) |
| return false; |
| DCHECK_GT(read_buffer_len_, 0); |
| return read_data_queue_.GetTotalSize() < |
| static_cast<size_t>(read_buffer_len_); |
| } |
| |
| bool BidirectionalStreamSpdyImpl::MaybeHandleStreamClosedInSendData() { |
| if (stream_) |
| return false; |
| // If |stream_| is closed without an error before client half closes, |
| // blackhole any pending write data. crbug.com/650438. |
| if (stream_closed_ && closed_stream_status_ == OK) { |
| base::ThreadTaskRunnerHandle::Get()->PostTask( |
| FROM_HERE, base::BindOnce(&BidirectionalStreamSpdyImpl::OnDataSent, |
| weak_factory_.GetWeakPtr())); |
| return true; |
| } |
| LOG(ERROR) << "Trying to send data after stream has been destroyed."; |
| base::ThreadTaskRunnerHandle::Get()->PostTask( |
| FROM_HERE, base::BindOnce(&BidirectionalStreamSpdyImpl::NotifyError, |
| weak_factory_.GetWeakPtr(), ERR_UNEXPECTED)); |
| return true; |
| } |
| |
| } // namespace net |