bidirectional_stream.cc 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407
  1. // Copyright 2016 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. #include "components/grpc_support/bidirectional_stream.h"
  5. #include <memory>
  6. #include <string>
  7. #include <utility>
  8. #include <vector>
  9. #include "base/bind.h"
  10. #include "base/callback_helpers.h"
  11. #include "base/location.h"
  12. #include "base/logging.h"
  13. #include "base/memory/ref_counted.h"
  14. #include "base/strings/abseil_string_conversions.h"
  15. #include "base/strings/string_number_conversions.h"
  16. #include "base/task/single_thread_task_runner.h"
  17. #include "net/base/http_user_agent_settings.h"
  18. #include "net/base/io_buffer.h"
  19. #include "net/base/net_errors.h"
  20. #include "net/base/request_priority.h"
  21. #include "net/http/bidirectional_stream.h"
  22. #include "net/http/bidirectional_stream_request_info.h"
  23. #include "net/http/http_network_session.h"
  24. #include "net/http/http_response_headers.h"
  25. #include "net/http/http_status_code.h"
  26. #include "net/http/http_transaction_factory.h"
  27. #include "net/http/http_util.h"
  28. #include "net/ssl/ssl_info.h"
  29. #include "net/url_request/url_request_context.h"
  30. #include "net/url_request/url_request_context_getter.h"
  31. #include "url/gurl.h"
  32. namespace grpc_support {
  33. BidirectionalStream::WriteBuffers::WriteBuffers() {}
  34. BidirectionalStream::WriteBuffers::~WriteBuffers() {}
  35. void BidirectionalStream::WriteBuffers::Clear() {
  36. write_buffer_list.clear();
  37. write_buffer_len_list.clear();
  38. }
  39. void BidirectionalStream::WriteBuffers::AppendBuffer(
  40. const scoped_refptr<net::IOBuffer>& buffer,
  41. int buffer_size) {
  42. write_buffer_list.push_back(buffer);
  43. write_buffer_len_list.push_back(buffer_size);
  44. }
  45. void BidirectionalStream::WriteBuffers::MoveTo(WriteBuffers* target) {
  46. std::move(write_buffer_list.begin(), write_buffer_list.end(),
  47. std::back_inserter(target->write_buffer_list));
  48. std::move(write_buffer_len_list.begin(), write_buffer_len_list.end(),
  49. std::back_inserter(target->write_buffer_len_list));
  50. Clear();
  51. }
  52. bool BidirectionalStream::WriteBuffers::Empty() const {
  53. return write_buffer_list.empty();
  54. }
  55. BidirectionalStream::BidirectionalStream(
  56. net::URLRequestContextGetter* request_context_getter,
  57. Delegate* delegate)
  58. : read_state_(NOT_STARTED),
  59. write_state_(NOT_STARTED),
  60. write_end_of_stream_(false),
  61. request_headers_sent_(false),
  62. disable_auto_flush_(false),
  63. delay_headers_until_flush_(false),
  64. request_context_getter_(request_context_getter),
  65. pending_write_data_(new WriteBuffers()),
  66. flushing_write_data_(new WriteBuffers()),
  67. sending_write_data_(new WriteBuffers()),
  68. delegate_(delegate) {
  69. weak_this_ = weak_factory_.GetWeakPtr();
  70. }
  71. BidirectionalStream::~BidirectionalStream() {
  72. DCHECK(IsOnNetworkThread());
  73. }
  74. int BidirectionalStream::Start(const char* url,
  75. int priority,
  76. const char* method,
  77. const net::HttpRequestHeaders& headers,
  78. bool end_of_stream) {
  79. // Prepare request info here to be able to return the error.
  80. std::unique_ptr<net::BidirectionalStreamRequestInfo> request_info(
  81. new net::BidirectionalStreamRequestInfo());
  82. request_info->url = GURL(url);
  83. request_info->priority = static_cast<net::RequestPriority>(priority);
  84. // Http method is a token, just as header name.
  85. request_info->method = method;
  86. if (!net::HttpUtil::IsValidHeaderName(request_info->method))
  87. return -1;
  88. request_info->extra_headers.CopyFrom(headers);
  89. request_info->end_stream_on_headers = end_of_stream;
  90. write_end_of_stream_ = end_of_stream;
  91. PostToNetworkThread(FROM_HERE,
  92. base::BindOnce(&BidirectionalStream::StartOnNetworkThread,
  93. weak_this_, std::move(request_info)));
  94. return 0;
  95. }
  96. bool BidirectionalStream::ReadData(char* buffer, int capacity) {
  97. if (!buffer)
  98. return false;
  99. scoped_refptr<net::WrappedIOBuffer> read_buffer =
  100. base::MakeRefCounted<net::WrappedIOBuffer>(buffer);
  101. PostToNetworkThread(
  102. FROM_HERE, base::BindOnce(&BidirectionalStream::ReadDataOnNetworkThread,
  103. weak_this_, std::move(read_buffer), capacity));
  104. return true;
  105. }
  106. bool BidirectionalStream::WriteData(const char* buffer,
  107. int count,
  108. bool end_of_stream) {
  109. if (!buffer)
  110. return false;
  111. scoped_refptr<net::WrappedIOBuffer> write_buffer =
  112. base::MakeRefCounted<net::WrappedIOBuffer>(buffer);
  113. PostToNetworkThread(
  114. FROM_HERE,
  115. base::BindOnce(&BidirectionalStream::WriteDataOnNetworkThread, weak_this_,
  116. std::move(write_buffer), count, end_of_stream));
  117. return true;
  118. }
  119. void BidirectionalStream::Flush() {
  120. PostToNetworkThread(
  121. FROM_HERE,
  122. base::BindOnce(&BidirectionalStream::FlushOnNetworkThread, weak_this_));
  123. }
  124. void BidirectionalStream::Cancel() {
  125. PostToNetworkThread(
  126. FROM_HERE,
  127. base::BindOnce(&BidirectionalStream::CancelOnNetworkThread, weak_this_));
  128. }
  129. void BidirectionalStream::Destroy() {
  130. // Destroy could be called from any thread, including network thread (if
  131. // posting task to executor throws an exception), but is posted, so |this|
  132. // is valid until calling task is complete.
  133. PostToNetworkThread(
  134. FROM_HERE, base::BindOnce(&BidirectionalStream::DestroyOnNetworkThread,
  135. base::Unretained(this)));
  136. }
  137. void BidirectionalStream::OnStreamReady(bool request_headers_sent) {
  138. DCHECK(IsOnNetworkThread());
  139. DCHECK_EQ(STARTED, write_state_);
  140. if (!bidi_stream_)
  141. return;
  142. request_headers_sent_ = request_headers_sent;
  143. write_state_ = WAITING_FOR_FLUSH;
  144. if (write_end_of_stream_) {
  145. if (!request_headers_sent) {
  146. // If there is no data to write, then just send headers explicitly.
  147. bidi_stream_->SendRequestHeaders();
  148. request_headers_sent_ = true;
  149. }
  150. write_state_ = WRITING_DONE;
  151. }
  152. delegate_->OnStreamReady();
  153. }
  154. void BidirectionalStream::OnHeadersReceived(
  155. const spdy::Http2HeaderBlock& response_headers) {
  156. DCHECK(IsOnNetworkThread());
  157. DCHECK_EQ(STARTED, read_state_);
  158. if (!bidi_stream_)
  159. return;
  160. read_state_ = WAITING_FOR_READ;
  161. // Get http status code from response headers.
  162. int http_status_code = 0;
  163. const auto http_status_header = response_headers.find(":status");
  164. if (http_status_header != response_headers.end())
  165. base::StringToInt(base::StringViewToStringPiece(http_status_header->second),
  166. &http_status_code);
  167. const char* protocol = "unknown";
  168. switch (bidi_stream_->GetProtocol()) {
  169. case net::kProtoHTTP2:
  170. protocol = "h2";
  171. break;
  172. case net::kProtoQUIC:
  173. protocol = "quic/1+spdy/3";
  174. break;
  175. default:
  176. break;
  177. }
  178. delegate_->OnHeadersReceived(response_headers, protocol);
  179. }
  180. void BidirectionalStream::OnDataRead(int bytes_read) {
  181. DCHECK(IsOnNetworkThread());
  182. DCHECK_EQ(READING, read_state_);
  183. if (!bidi_stream_)
  184. return;
  185. read_state_ = WAITING_FOR_READ;
  186. delegate_->OnDataRead(read_buffer_->data(), bytes_read);
  187. // Free the read buffer.
  188. read_buffer_ = nullptr;
  189. if (bytes_read == 0)
  190. read_state_ = READING_DONE;
  191. MaybeOnSucceded();
  192. }
  193. void BidirectionalStream::OnDataSent() {
  194. DCHECK(IsOnNetworkThread());
  195. if (!bidi_stream_)
  196. return;
  197. DCHECK_EQ(WRITING, write_state_);
  198. write_state_ = WAITING_FOR_FLUSH;
  199. for (const scoped_refptr<net::IOBuffer>& buffer :
  200. sending_write_data_->buffers()) {
  201. delegate_->OnDataSent(buffer->data());
  202. }
  203. sending_write_data_->Clear();
  204. // Send data flushed while other data was sending.
  205. if (!flushing_write_data_->Empty()) {
  206. SendFlushingWriteData();
  207. return;
  208. }
  209. if (write_end_of_stream_ && pending_write_data_->Empty()) {
  210. write_state_ = WRITING_DONE;
  211. MaybeOnSucceded();
  212. }
  213. }
  214. void BidirectionalStream::OnTrailersReceived(
  215. const spdy::Http2HeaderBlock& response_trailers) {
  216. DCHECK(IsOnNetworkThread());
  217. if (!bidi_stream_)
  218. return;
  219. delegate_->OnTrailersReceived(response_trailers);
  220. }
  221. void BidirectionalStream::OnFailed(int error) {
  222. DCHECK(IsOnNetworkThread());
  223. if (!bidi_stream_ && read_state_ != NOT_STARTED)
  224. return;
  225. read_state_ = write_state_ = ERR;
  226. weak_factory_.InvalidateWeakPtrs();
  227. // Delete underlying |bidi_stream_| asynchronously as it may still be used.
  228. PostToNetworkThread(
  229. FROM_HERE, base::BindOnce(&base::DeletePointer<net::BidirectionalStream>,
  230. bidi_stream_.release()));
  231. delegate_->OnFailed(error);
  232. }
  233. void BidirectionalStream::StartOnNetworkThread(
  234. std::unique_ptr<net::BidirectionalStreamRequestInfo> request_info) {
  235. DCHECK(IsOnNetworkThread());
  236. DCHECK(!bidi_stream_);
  237. DCHECK(request_context_getter_->GetURLRequestContext());
  238. net::URLRequestContext* request_context =
  239. request_context_getter_->GetURLRequestContext();
  240. request_info->extra_headers.SetHeaderIfMissing(
  241. net::HttpRequestHeaders::kUserAgent,
  242. request_context->http_user_agent_settings()->GetUserAgent());
  243. bidi_stream_ = std::make_unique<net::BidirectionalStream>(
  244. std::move(request_info),
  245. request_context->http_transaction_factory()->GetSession(),
  246. !delay_headers_until_flush_, this);
  247. DCHECK(read_state_ == NOT_STARTED && write_state_ == NOT_STARTED);
  248. read_state_ = write_state_ = STARTED;
  249. }
  250. void BidirectionalStream::ReadDataOnNetworkThread(
  251. scoped_refptr<net::WrappedIOBuffer> read_buffer,
  252. int buffer_size) {
  253. DCHECK(IsOnNetworkThread());
  254. DCHECK(read_buffer);
  255. DCHECK(!read_buffer_);
  256. if (read_state_ != WAITING_FOR_READ) {
  257. DLOG(ERROR) << "Unexpected Read Data in read_state " << read_state_;
  258. // Invoke OnFailed unless it is already invoked.
  259. if (read_state_ != ERR)
  260. OnFailed(net::ERR_UNEXPECTED);
  261. return;
  262. }
  263. read_state_ = READING;
  264. read_buffer_ = read_buffer;
  265. int bytes_read = bidi_stream_->ReadData(read_buffer_.get(), buffer_size);
  266. // If IO is pending, wait for the BidirectionalStream to call OnDataRead.
  267. if (bytes_read == net::ERR_IO_PENDING)
  268. return;
  269. if (bytes_read < 0) {
  270. OnFailed(bytes_read);
  271. return;
  272. }
  273. OnDataRead(bytes_read);
  274. }
  275. void BidirectionalStream::WriteDataOnNetworkThread(
  276. scoped_refptr<net::WrappedIOBuffer> write_buffer,
  277. int buffer_size,
  278. bool end_of_stream) {
  279. DCHECK(IsOnNetworkThread());
  280. DCHECK(write_buffer);
  281. DCHECK(!write_end_of_stream_);
  282. if (!bidi_stream_ || write_end_of_stream_) {
  283. DLOG(ERROR) << "Unexpected Flush Data in write_state " << write_state_;
  284. // Invoke OnFailed unless it is already invoked.
  285. if (write_state_ != ERR)
  286. OnFailed(net::ERR_UNEXPECTED);
  287. return;
  288. }
  289. pending_write_data_->AppendBuffer(write_buffer, buffer_size);
  290. write_end_of_stream_ = end_of_stream;
  291. if (!disable_auto_flush_)
  292. FlushOnNetworkThread();
  293. }
  294. void BidirectionalStream::FlushOnNetworkThread() {
  295. DCHECK(IsOnNetworkThread());
  296. if (!bidi_stream_)
  297. return;
  298. // If there is no data to flush, may need to send headers.
  299. if (pending_write_data_->Empty()) {
  300. if (!request_headers_sent_) {
  301. request_headers_sent_ = true;
  302. bidi_stream_->SendRequestHeaders();
  303. }
  304. return;
  305. }
  306. // If request headers are not sent yet, they will be sent with the data.
  307. if (!request_headers_sent_)
  308. request_headers_sent_ = true;
  309. // Move pending data to the flushing list.
  310. pending_write_data_->MoveTo(flushing_write_data_.get());
  311. DCHECK(pending_write_data_->Empty());
  312. if (write_state_ != WRITING)
  313. SendFlushingWriteData();
  314. }
  315. void BidirectionalStream::SendFlushingWriteData() {
  316. DCHECK(bidi_stream_);
  317. // If previous send is not done, or there is nothing to flush, then exit.
  318. if (write_state_ == WRITING || flushing_write_data_->Empty())
  319. return;
  320. DCHECK(sending_write_data_->Empty());
  321. write_state_ = WRITING;
  322. flushing_write_data_->MoveTo(sending_write_data_.get());
  323. bidi_stream_->SendvData(sending_write_data_->buffers(),
  324. sending_write_data_->lengths(),
  325. write_end_of_stream_ && pending_write_data_->Empty());
  326. }
  327. void BidirectionalStream::CancelOnNetworkThread() {
  328. DCHECK(IsOnNetworkThread());
  329. if (!bidi_stream_)
  330. return;
  331. read_state_ = write_state_ = CANCELED;
  332. bidi_stream_.reset();
  333. weak_factory_.InvalidateWeakPtrs();
  334. delegate_->OnCanceled();
  335. }
  336. void BidirectionalStream::DestroyOnNetworkThread() {
  337. DCHECK(IsOnNetworkThread());
  338. delete this;
  339. }
  340. void BidirectionalStream::MaybeOnSucceded() {
  341. DCHECK(IsOnNetworkThread());
  342. if (!bidi_stream_)
  343. return;
  344. if (read_state_ == READING_DONE && write_state_ == WRITING_DONE) {
  345. read_state_ = write_state_ = SUCCESS;
  346. weak_factory_.InvalidateWeakPtrs();
  347. // Delete underlying |bidi_stream_| asynchronously as it may still be used.
  348. PostToNetworkThread(
  349. FROM_HERE,
  350. base::BindOnce(&base::DeletePointer<net::BidirectionalStream>,
  351. bidi_stream_.release()));
  352. delegate_->OnSucceeded();
  353. }
  354. }
  355. bool BidirectionalStream::IsOnNetworkThread() {
  356. return request_context_getter_->GetNetworkTaskRunner()
  357. ->BelongsToCurrentThread();
  358. }
  359. void BidirectionalStream::PostToNetworkThread(const base::Location& from_here,
  360. base::OnceClosure task) {
  361. request_context_getter_->GetNetworkTaskRunner()->PostTask(from_here,
  362. std::move(task));
  363. }
  364. } // namespace grpc_support