123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407 |
- // Copyright 2016 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 "components/grpc_support/bidirectional_stream.h"
- #include <memory>
- #include <string>
- #include <utility>
- #include <vector>
- #include "base/bind.h"
- #include "base/callback_helpers.h"
- #include "base/location.h"
- #include "base/logging.h"
- #include "base/memory/ref_counted.h"
- #include "base/strings/abseil_string_conversions.h"
- #include "base/strings/string_number_conversions.h"
- #include "base/task/single_thread_task_runner.h"
- #include "net/base/http_user_agent_settings.h"
- #include "net/base/io_buffer.h"
- #include "net/base/net_errors.h"
- #include "net/base/request_priority.h"
- #include "net/http/bidirectional_stream.h"
- #include "net/http/bidirectional_stream_request_info.h"
- #include "net/http/http_network_session.h"
- #include "net/http/http_response_headers.h"
- #include "net/http/http_status_code.h"
- #include "net/http/http_transaction_factory.h"
- #include "net/http/http_util.h"
- #include "net/ssl/ssl_info.h"
- #include "net/url_request/url_request_context.h"
- #include "net/url_request/url_request_context_getter.h"
- #include "url/gurl.h"
- namespace grpc_support {
- BidirectionalStream::WriteBuffers::WriteBuffers() {}
- BidirectionalStream::WriteBuffers::~WriteBuffers() {}
- void BidirectionalStream::WriteBuffers::Clear() {
- write_buffer_list.clear();
- write_buffer_len_list.clear();
- }
- void BidirectionalStream::WriteBuffers::AppendBuffer(
- const scoped_refptr<net::IOBuffer>& buffer,
- int buffer_size) {
- write_buffer_list.push_back(buffer);
- write_buffer_len_list.push_back(buffer_size);
- }
- void BidirectionalStream::WriteBuffers::MoveTo(WriteBuffers* target) {
- std::move(write_buffer_list.begin(), write_buffer_list.end(),
- std::back_inserter(target->write_buffer_list));
- std::move(write_buffer_len_list.begin(), write_buffer_len_list.end(),
- std::back_inserter(target->write_buffer_len_list));
- Clear();
- }
- bool BidirectionalStream::WriteBuffers::Empty() const {
- return write_buffer_list.empty();
- }
- BidirectionalStream::BidirectionalStream(
- net::URLRequestContextGetter* request_context_getter,
- Delegate* delegate)
- : read_state_(NOT_STARTED),
- write_state_(NOT_STARTED),
- write_end_of_stream_(false),
- request_headers_sent_(false),
- disable_auto_flush_(false),
- delay_headers_until_flush_(false),
- request_context_getter_(request_context_getter),
- pending_write_data_(new WriteBuffers()),
- flushing_write_data_(new WriteBuffers()),
- sending_write_data_(new WriteBuffers()),
- delegate_(delegate) {
- weak_this_ = weak_factory_.GetWeakPtr();
- }
- BidirectionalStream::~BidirectionalStream() {
- DCHECK(IsOnNetworkThread());
- }
- int BidirectionalStream::Start(const char* url,
- int priority,
- const char* method,
- const net::HttpRequestHeaders& headers,
- bool end_of_stream) {
- // Prepare request info here to be able to return the error.
- std::unique_ptr<net::BidirectionalStreamRequestInfo> request_info(
- new net::BidirectionalStreamRequestInfo());
- request_info->url = GURL(url);
- request_info->priority = static_cast<net::RequestPriority>(priority);
- // Http method is a token, just as header name.
- request_info->method = method;
- if (!net::HttpUtil::IsValidHeaderName(request_info->method))
- return -1;
- request_info->extra_headers.CopyFrom(headers);
- request_info->end_stream_on_headers = end_of_stream;
- write_end_of_stream_ = end_of_stream;
- PostToNetworkThread(FROM_HERE,
- base::BindOnce(&BidirectionalStream::StartOnNetworkThread,
- weak_this_, std::move(request_info)));
- return 0;
- }
- bool BidirectionalStream::ReadData(char* buffer, int capacity) {
- if (!buffer)
- return false;
- scoped_refptr<net::WrappedIOBuffer> read_buffer =
- base::MakeRefCounted<net::WrappedIOBuffer>(buffer);
- PostToNetworkThread(
- FROM_HERE, base::BindOnce(&BidirectionalStream::ReadDataOnNetworkThread,
- weak_this_, std::move(read_buffer), capacity));
- return true;
- }
- bool BidirectionalStream::WriteData(const char* buffer,
- int count,
- bool end_of_stream) {
- if (!buffer)
- return false;
- scoped_refptr<net::WrappedIOBuffer> write_buffer =
- base::MakeRefCounted<net::WrappedIOBuffer>(buffer);
- PostToNetworkThread(
- FROM_HERE,
- base::BindOnce(&BidirectionalStream::WriteDataOnNetworkThread, weak_this_,
- std::move(write_buffer), count, end_of_stream));
- return true;
- }
- void BidirectionalStream::Flush() {
- PostToNetworkThread(
- FROM_HERE,
- base::BindOnce(&BidirectionalStream::FlushOnNetworkThread, weak_this_));
- }
- void BidirectionalStream::Cancel() {
- PostToNetworkThread(
- FROM_HERE,
- base::BindOnce(&BidirectionalStream::CancelOnNetworkThread, weak_this_));
- }
- void BidirectionalStream::Destroy() {
- // Destroy could be called from any thread, including network thread (if
- // posting task to executor throws an exception), but is posted, so |this|
- // is valid until calling task is complete.
- PostToNetworkThread(
- FROM_HERE, base::BindOnce(&BidirectionalStream::DestroyOnNetworkThread,
- base::Unretained(this)));
- }
- void BidirectionalStream::OnStreamReady(bool request_headers_sent) {
- DCHECK(IsOnNetworkThread());
- DCHECK_EQ(STARTED, write_state_);
- if (!bidi_stream_)
- return;
- request_headers_sent_ = request_headers_sent;
- write_state_ = WAITING_FOR_FLUSH;
- if (write_end_of_stream_) {
- if (!request_headers_sent) {
- // If there is no data to write, then just send headers explicitly.
- bidi_stream_->SendRequestHeaders();
- request_headers_sent_ = true;
- }
- write_state_ = WRITING_DONE;
- }
- delegate_->OnStreamReady();
- }
- void BidirectionalStream::OnHeadersReceived(
- const spdy::Http2HeaderBlock& response_headers) {
- DCHECK(IsOnNetworkThread());
- DCHECK_EQ(STARTED, read_state_);
- if (!bidi_stream_)
- return;
- read_state_ = WAITING_FOR_READ;
- // Get http status code from response headers.
- int http_status_code = 0;
- const auto http_status_header = response_headers.find(":status");
- if (http_status_header != response_headers.end())
- base::StringToInt(base::StringViewToStringPiece(http_status_header->second),
- &http_status_code);
- const char* protocol = "unknown";
- switch (bidi_stream_->GetProtocol()) {
- case net::kProtoHTTP2:
- protocol = "h2";
- break;
- case net::kProtoQUIC:
- protocol = "quic/1+spdy/3";
- break;
- default:
- break;
- }
- delegate_->OnHeadersReceived(response_headers, protocol);
- }
- void BidirectionalStream::OnDataRead(int bytes_read) {
- DCHECK(IsOnNetworkThread());
- DCHECK_EQ(READING, read_state_);
- if (!bidi_stream_)
- return;
- read_state_ = WAITING_FOR_READ;
- delegate_->OnDataRead(read_buffer_->data(), bytes_read);
- // Free the read buffer.
- read_buffer_ = nullptr;
- if (bytes_read == 0)
- read_state_ = READING_DONE;
- MaybeOnSucceded();
- }
- void BidirectionalStream::OnDataSent() {
- DCHECK(IsOnNetworkThread());
- if (!bidi_stream_)
- return;
- DCHECK_EQ(WRITING, write_state_);
- write_state_ = WAITING_FOR_FLUSH;
- for (const scoped_refptr<net::IOBuffer>& buffer :
- sending_write_data_->buffers()) {
- delegate_->OnDataSent(buffer->data());
- }
- sending_write_data_->Clear();
- // Send data flushed while other data was sending.
- if (!flushing_write_data_->Empty()) {
- SendFlushingWriteData();
- return;
- }
- if (write_end_of_stream_ && pending_write_data_->Empty()) {
- write_state_ = WRITING_DONE;
- MaybeOnSucceded();
- }
- }
- void BidirectionalStream::OnTrailersReceived(
- const spdy::Http2HeaderBlock& response_trailers) {
- DCHECK(IsOnNetworkThread());
- if (!bidi_stream_)
- return;
- delegate_->OnTrailersReceived(response_trailers);
- }
- void BidirectionalStream::OnFailed(int error) {
- DCHECK(IsOnNetworkThread());
- if (!bidi_stream_ && read_state_ != NOT_STARTED)
- return;
- read_state_ = write_state_ = ERR;
- weak_factory_.InvalidateWeakPtrs();
- // Delete underlying |bidi_stream_| asynchronously as it may still be used.
- PostToNetworkThread(
- FROM_HERE, base::BindOnce(&base::DeletePointer<net::BidirectionalStream>,
- bidi_stream_.release()));
- delegate_->OnFailed(error);
- }
- void BidirectionalStream::StartOnNetworkThread(
- std::unique_ptr<net::BidirectionalStreamRequestInfo> request_info) {
- DCHECK(IsOnNetworkThread());
- DCHECK(!bidi_stream_);
- DCHECK(request_context_getter_->GetURLRequestContext());
- net::URLRequestContext* request_context =
- request_context_getter_->GetURLRequestContext();
- request_info->extra_headers.SetHeaderIfMissing(
- net::HttpRequestHeaders::kUserAgent,
- request_context->http_user_agent_settings()->GetUserAgent());
- bidi_stream_ = std::make_unique<net::BidirectionalStream>(
- std::move(request_info),
- request_context->http_transaction_factory()->GetSession(),
- !delay_headers_until_flush_, this);
- DCHECK(read_state_ == NOT_STARTED && write_state_ == NOT_STARTED);
- read_state_ = write_state_ = STARTED;
- }
- void BidirectionalStream::ReadDataOnNetworkThread(
- scoped_refptr<net::WrappedIOBuffer> read_buffer,
- int buffer_size) {
- DCHECK(IsOnNetworkThread());
- DCHECK(read_buffer);
- DCHECK(!read_buffer_);
- if (read_state_ != WAITING_FOR_READ) {
- DLOG(ERROR) << "Unexpected Read Data in read_state " << read_state_;
- // Invoke OnFailed unless it is already invoked.
- if (read_state_ != ERR)
- OnFailed(net::ERR_UNEXPECTED);
- return;
- }
- read_state_ = READING;
- read_buffer_ = read_buffer;
- int bytes_read = bidi_stream_->ReadData(read_buffer_.get(), buffer_size);
- // If IO is pending, wait for the BidirectionalStream to call OnDataRead.
- if (bytes_read == net::ERR_IO_PENDING)
- return;
- if (bytes_read < 0) {
- OnFailed(bytes_read);
- return;
- }
- OnDataRead(bytes_read);
- }
- void BidirectionalStream::WriteDataOnNetworkThread(
- scoped_refptr<net::WrappedIOBuffer> write_buffer,
- int buffer_size,
- bool end_of_stream) {
- DCHECK(IsOnNetworkThread());
- DCHECK(write_buffer);
- DCHECK(!write_end_of_stream_);
- if (!bidi_stream_ || write_end_of_stream_) {
- DLOG(ERROR) << "Unexpected Flush Data in write_state " << write_state_;
- // Invoke OnFailed unless it is already invoked.
- if (write_state_ != ERR)
- OnFailed(net::ERR_UNEXPECTED);
- return;
- }
- pending_write_data_->AppendBuffer(write_buffer, buffer_size);
- write_end_of_stream_ = end_of_stream;
- if (!disable_auto_flush_)
- FlushOnNetworkThread();
- }
- void BidirectionalStream::FlushOnNetworkThread() {
- DCHECK(IsOnNetworkThread());
- if (!bidi_stream_)
- return;
- // If there is no data to flush, may need to send headers.
- if (pending_write_data_->Empty()) {
- if (!request_headers_sent_) {
- request_headers_sent_ = true;
- bidi_stream_->SendRequestHeaders();
- }
- return;
- }
- // If request headers are not sent yet, they will be sent with the data.
- if (!request_headers_sent_)
- request_headers_sent_ = true;
- // Move pending data to the flushing list.
- pending_write_data_->MoveTo(flushing_write_data_.get());
- DCHECK(pending_write_data_->Empty());
- if (write_state_ != WRITING)
- SendFlushingWriteData();
- }
- void BidirectionalStream::SendFlushingWriteData() {
- DCHECK(bidi_stream_);
- // If previous send is not done, or there is nothing to flush, then exit.
- if (write_state_ == WRITING || flushing_write_data_->Empty())
- return;
- DCHECK(sending_write_data_->Empty());
- write_state_ = WRITING;
- flushing_write_data_->MoveTo(sending_write_data_.get());
- bidi_stream_->SendvData(sending_write_data_->buffers(),
- sending_write_data_->lengths(),
- write_end_of_stream_ && pending_write_data_->Empty());
- }
- void BidirectionalStream::CancelOnNetworkThread() {
- DCHECK(IsOnNetworkThread());
- if (!bidi_stream_)
- return;
- read_state_ = write_state_ = CANCELED;
- bidi_stream_.reset();
- weak_factory_.InvalidateWeakPtrs();
- delegate_->OnCanceled();
- }
- void BidirectionalStream::DestroyOnNetworkThread() {
- DCHECK(IsOnNetworkThread());
- delete this;
- }
- void BidirectionalStream::MaybeOnSucceded() {
- DCHECK(IsOnNetworkThread());
- if (!bidi_stream_)
- return;
- if (read_state_ == READING_DONE && write_state_ == WRITING_DONE) {
- read_state_ = write_state_ = SUCCESS;
- weak_factory_.InvalidateWeakPtrs();
- // Delete underlying |bidi_stream_| asynchronously as it may still be used.
- PostToNetworkThread(
- FROM_HERE,
- base::BindOnce(&base::DeletePointer<net::BidirectionalStream>,
- bidi_stream_.release()));
- delegate_->OnSucceeded();
- }
- }
- bool BidirectionalStream::IsOnNetworkThread() {
- return request_context_getter_->GetNetworkTaskRunner()
- ->BelongsToCurrentThread();
- }
- void BidirectionalStream::PostToNetworkThread(const base::Location& from_here,
- base::OnceClosure task) {
- request_context_getter_->GetNetworkTaskRunner()->PostTask(from_here,
- std::move(task));
- }
- } // namespace grpc_support
|