bidirectional_stream_c.cc 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304
  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/include/bidirectional_stream_c.h"
  5. #include <stdbool.h>
  6. #include <memory>
  7. #include <string>
  8. #include <vector>
  9. #include "base/bind.h"
  10. #include "base/location.h"
  11. #include "base/logging.h"
  12. #include "base/memory/raw_ptr.h"
  13. #include "base/memory/ref_counted.h"
  14. #include "base/strings/string_number_conversions.h"
  15. #include "base/strings/string_split.h"
  16. #include "base/task/single_thread_task_runner.h"
  17. #include "components/grpc_support/bidirectional_stream.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/third_party/quiche/src/quiche/spdy/core/http2_header_block.h"
  30. #include "net/url_request/url_request_context.h"
  31. #include "url/gurl.h"
  32. namespace {
  33. class HeadersArray : public bidirectional_stream_header_array {
  34. public:
  35. explicit HeadersArray(const spdy::Http2HeaderBlock& header_block);
  36. HeadersArray(const HeadersArray&) = delete;
  37. HeadersArray& operator=(const HeadersArray&) = delete;
  38. ~HeadersArray();
  39. private:
  40. base::StringPairs headers_strings_;
  41. };
  42. HeadersArray::HeadersArray(const spdy::Http2HeaderBlock& header_block)
  43. : headers_strings_(header_block.size()) {
  44. // Split coalesced headers by '\0' and copy them into |header_strings_|.
  45. for (const auto& it : header_block) {
  46. auto value = std::string(it.second);
  47. size_t start = 0;
  48. size_t end = 0;
  49. do {
  50. end = value.find('\0', start);
  51. std::string split_value;
  52. if (end != value.npos) {
  53. split_value = value.substr(start, end - start);
  54. } else {
  55. split_value = value.substr(start);
  56. }
  57. // |headers_strings_| is initialized to the size of header_block, but
  58. // split headers might take up more space.
  59. headers_strings_.push_back(
  60. std::make_pair(std::string(it.first), split_value));
  61. start = end + 1;
  62. } while (end != value.npos);
  63. }
  64. count = capacity = headers_strings_.size();
  65. headers = new bidirectional_stream_header[count];
  66. size_t i = 0;
  67. for (const auto& it : headers_strings_) {
  68. headers[i].key = it.first.c_str();
  69. headers[i].value = it.second.c_str();
  70. ++i;
  71. }
  72. }
  73. HeadersArray::~HeadersArray() {
  74. delete[] headers;
  75. }
  76. class BidirectionalStreamAdapter
  77. : public grpc_support::BidirectionalStream::Delegate {
  78. public:
  79. BidirectionalStreamAdapter(stream_engine* engine,
  80. void* annotation,
  81. bidirectional_stream_callback* callback);
  82. virtual ~BidirectionalStreamAdapter();
  83. void OnStreamReady() override;
  84. void OnHeadersReceived(const spdy::Http2HeaderBlock& headers_block,
  85. const char* negotiated_protocol) override;
  86. void OnDataRead(char* data, int size) override;
  87. void OnDataSent(const char* data) override;
  88. void OnTrailersReceived(
  89. const spdy::Http2HeaderBlock& trailers_block) override;
  90. void OnSucceeded() override;
  91. void OnFailed(int error) override;
  92. void OnCanceled() override;
  93. bidirectional_stream* c_stream() const { return c_stream_.get(); }
  94. static grpc_support::BidirectionalStream* GetStream(
  95. bidirectional_stream* stream);
  96. static void DestroyAdapterForStream(bidirectional_stream* stream);
  97. private:
  98. void DestroyOnNetworkThread();
  99. // None of these objects are owned by |this|.
  100. raw_ptr<net::URLRequestContextGetter> request_context_getter_;
  101. raw_ptr<grpc_support::BidirectionalStream> bidirectional_stream_;
  102. // C side
  103. std::unique_ptr<bidirectional_stream> c_stream_;
  104. raw_ptr<bidirectional_stream_callback> c_callback_;
  105. };
  106. BidirectionalStreamAdapter::BidirectionalStreamAdapter(
  107. stream_engine* engine,
  108. void* annotation,
  109. bidirectional_stream_callback* callback)
  110. : request_context_getter_(
  111. reinterpret_cast<net::URLRequestContextGetter*>(engine->obj)),
  112. c_stream_(std::make_unique<bidirectional_stream>()),
  113. c_callback_(callback) {
  114. DCHECK(request_context_getter_);
  115. bidirectional_stream_ =
  116. new grpc_support::BidirectionalStream(request_context_getter_, this);
  117. c_stream_->obj = this;
  118. c_stream_->annotation = annotation;
  119. }
  120. BidirectionalStreamAdapter::~BidirectionalStreamAdapter() {}
  121. void BidirectionalStreamAdapter::OnStreamReady() {
  122. DCHECK(c_callback_->on_response_headers_received);
  123. c_callback_->on_stream_ready(c_stream());
  124. }
  125. void BidirectionalStreamAdapter::OnHeadersReceived(
  126. const spdy::Http2HeaderBlock& headers_block,
  127. const char* negotiated_protocol) {
  128. DCHECK(c_callback_->on_response_headers_received);
  129. HeadersArray response_headers(headers_block);
  130. c_callback_->on_response_headers_received(c_stream(), &response_headers,
  131. negotiated_protocol);
  132. }
  133. void BidirectionalStreamAdapter::OnDataRead(char* data, int size) {
  134. DCHECK(c_callback_->on_read_completed);
  135. c_callback_->on_read_completed(c_stream(), data, size);
  136. }
  137. void BidirectionalStreamAdapter::OnDataSent(const char* data) {
  138. DCHECK(c_callback_->on_write_completed);
  139. c_callback_->on_write_completed(c_stream(), data);
  140. }
  141. void BidirectionalStreamAdapter::OnTrailersReceived(
  142. const spdy::Http2HeaderBlock& trailers_block) {
  143. DCHECK(c_callback_->on_response_trailers_received);
  144. HeadersArray response_trailers(trailers_block);
  145. c_callback_->on_response_trailers_received(c_stream(), &response_trailers);
  146. }
  147. void BidirectionalStreamAdapter::OnSucceeded() {
  148. DCHECK(c_callback_->on_succeded);
  149. c_callback_->on_succeded(c_stream());
  150. }
  151. void BidirectionalStreamAdapter::OnFailed(int error) {
  152. DCHECK(c_callback_->on_failed);
  153. c_callback_->on_failed(c_stream(), error);
  154. }
  155. void BidirectionalStreamAdapter::OnCanceled() {
  156. DCHECK(c_callback_->on_canceled);
  157. c_callback_->on_canceled(c_stream());
  158. }
  159. grpc_support::BidirectionalStream* BidirectionalStreamAdapter::GetStream(
  160. bidirectional_stream* stream) {
  161. DCHECK(stream);
  162. BidirectionalStreamAdapter* adapter =
  163. static_cast<BidirectionalStreamAdapter*>(stream->obj);
  164. DCHECK(adapter->c_stream() == stream);
  165. DCHECK(adapter->bidirectional_stream_);
  166. return adapter->bidirectional_stream_;
  167. }
  168. void BidirectionalStreamAdapter::DestroyAdapterForStream(
  169. bidirectional_stream* stream) {
  170. DCHECK(stream);
  171. BidirectionalStreamAdapter* adapter =
  172. static_cast<BidirectionalStreamAdapter*>(stream->obj);
  173. DCHECK(adapter->c_stream() == stream);
  174. // Destroy could be called from any thread, including network thread (if
  175. // posting task to executor throws an exception), but is posted, so |this|
  176. // is valid until calling task is complete.
  177. adapter->bidirectional_stream_->Destroy();
  178. adapter->request_context_getter_->GetNetworkTaskRunner()->PostTask(
  179. FROM_HERE,
  180. base::BindOnce(&BidirectionalStreamAdapter::DestroyOnNetworkThread,
  181. base::Unretained(adapter)));
  182. }
  183. void BidirectionalStreamAdapter::DestroyOnNetworkThread() {
  184. DCHECK(request_context_getter_->GetNetworkTaskRunner()
  185. ->BelongsToCurrentThread());
  186. delete this;
  187. }
  188. } // namespace
  189. bidirectional_stream* bidirectional_stream_create(
  190. stream_engine* engine,
  191. void* annotation,
  192. bidirectional_stream_callback* callback) {
  193. // Allocate new C++ adapter that will invoke |callback|.
  194. BidirectionalStreamAdapter* stream_adapter =
  195. new BidirectionalStreamAdapter(engine, annotation, callback);
  196. return stream_adapter->c_stream();
  197. }
  198. int bidirectional_stream_destroy(bidirectional_stream* stream) {
  199. BidirectionalStreamAdapter::DestroyAdapterForStream(stream);
  200. return 1;
  201. }
  202. void bidirectional_stream_disable_auto_flush(bidirectional_stream* stream,
  203. bool disable_auto_flush) {
  204. BidirectionalStreamAdapter::GetStream(stream)->disable_auto_flush(
  205. disable_auto_flush);
  206. }
  207. void bidirectional_stream_delay_request_headers_until_flush(
  208. bidirectional_stream* stream,
  209. bool delay_headers_until_flush) {
  210. BidirectionalStreamAdapter::GetStream(stream)->delay_headers_until_flush(
  211. delay_headers_until_flush);
  212. }
  213. int bidirectional_stream_start(bidirectional_stream* stream,
  214. const char* url,
  215. int priority,
  216. const char* method,
  217. const bidirectional_stream_header_array* headers,
  218. bool end_of_stream) {
  219. grpc_support::BidirectionalStream* internal_stream =
  220. BidirectionalStreamAdapter::GetStream(stream);
  221. net::HttpRequestHeaders request_headers;
  222. if (headers) {
  223. for (size_t i = 0; i < headers->count; ++i) {
  224. std::string name(headers->headers[i].key);
  225. std::string value(headers->headers[i].value);
  226. if (!net::HttpUtil::IsValidHeaderName(name) ||
  227. !net::HttpUtil::IsValidHeaderValue(value)) {
  228. DLOG(ERROR) << "Invalid Header " << name << "=" << value;
  229. return i + 1;
  230. }
  231. request_headers.SetHeader(name, value);
  232. }
  233. }
  234. return internal_stream->Start(url, priority, method, request_headers,
  235. end_of_stream);
  236. }
  237. int bidirectional_stream_read(bidirectional_stream* stream,
  238. char* buffer,
  239. int capacity) {
  240. return BidirectionalStreamAdapter::GetStream(stream)->ReadData(buffer,
  241. capacity);
  242. }
  243. int bidirectional_stream_write(bidirectional_stream* stream,
  244. const char* buffer,
  245. int count,
  246. bool end_of_stream) {
  247. return BidirectionalStreamAdapter::GetStream(stream)->WriteData(
  248. buffer, count, end_of_stream);
  249. }
  250. void bidirectional_stream_flush(bidirectional_stream* stream) {
  251. return BidirectionalStreamAdapter::GetStream(stream)->Flush();
  252. }
  253. void bidirectional_stream_cancel(bidirectional_stream* stream) {
  254. BidirectionalStreamAdapter::GetStream(stream)->Cancel();
  255. }