bidirectional_stream.h 8.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247
  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. #ifndef COMPONENTS_GRPC_SUPPORT_BIDIRECTIONAL_STREAM_H_
  5. #define COMPONENTS_GRPC_SUPPORT_BIDIRECTIONAL_STREAM_H_
  6. #include <memory>
  7. #include <vector>
  8. #include "base/memory/raw_ptr.h"
  9. #include "base/memory/ref_counted.h"
  10. #include "base/memory/weak_ptr.h"
  11. #include "base/synchronization/lock.h"
  12. #include "net/http/bidirectional_stream.h"
  13. #include "net/third_party/quiche/src/quiche/spdy/core/http2_header_block.h"
  14. #include "net/url_request/url_request_context_getter.h"
  15. namespace base {
  16. class Location;
  17. } // namespace base
  18. namespace net {
  19. class HttpRequestHeaders;
  20. class WrappedIOBuffer;
  21. } // namespace net
  22. namespace grpc_support {
  23. // An adapter to net::BidirectionalStream.
  24. // Created and configured from any thread. Start, ReadData, WriteData and
  25. // Destroy can be called on any thread (including network thread), and post
  26. // calls to corresponding {Start|ReadData|WriteData|Destroy}OnNetworkThread to
  27. // the network thread. The object is always deleted on network thread. All
  28. // callbacks into the Delegate are done on the network thread.
  29. // The app is expected to initiate the next step like ReadData or Destroy.
  30. // Public methods can be called on any thread.
  31. class BidirectionalStream : public net::BidirectionalStream::Delegate {
  32. public:
  33. class Delegate {
  34. public:
  35. virtual void OnStreamReady() = 0;
  36. virtual void OnHeadersReceived(
  37. const spdy::Http2HeaderBlock& response_headers,
  38. const char* negotiated_protocol) = 0;
  39. virtual void OnDataRead(char* data, int size) = 0;
  40. virtual void OnDataSent(const char* data) = 0;
  41. virtual void OnTrailersReceived(const spdy::Http2HeaderBlock& trailers) = 0;
  42. virtual void OnSucceeded() = 0;
  43. virtual void OnFailed(int error) = 0;
  44. virtual void OnCanceled() = 0;
  45. };
  46. BidirectionalStream(net::URLRequestContextGetter* request_context_getter,
  47. Delegate* delegate);
  48. BidirectionalStream(const BidirectionalStream&) = delete;
  49. BidirectionalStream& operator=(const BidirectionalStream&) = delete;
  50. ~BidirectionalStream() override;
  51. // Disables automatic flushing of each buffer passed to WriteData().
  52. void disable_auto_flush(bool disable_auto_flush) {
  53. disable_auto_flush_ = disable_auto_flush;
  54. }
  55. // Delays sending request headers until first call to Flush().
  56. void delay_headers_until_flush(bool delay_headers_until_flush) {
  57. delay_headers_until_flush_ = delay_headers_until_flush;
  58. }
  59. // Validates method and headers, initializes and starts the request. If
  60. // |end_of_stream| is true, then stream is half-closed after sending header
  61. // frame and no data is expected to be written.
  62. // Returns 0 if request is valid and started successfully,
  63. // Returns -1 if |method| is not valid HTTP method name.
  64. // Returns position of invalid header value in |headers| if header name is
  65. // not valid.
  66. int Start(const char* url,
  67. int priority,
  68. const char* method,
  69. const net::HttpRequestHeaders& headers,
  70. bool end_of_stream);
  71. // Reads more data into |buffer| up to |capacity| bytes.
  72. bool ReadData(char* buffer, int capacity);
  73. // Writes |count| bytes of data from |buffer|. The |end_of_stream| is
  74. // passed to remote to indicate end of stream.
  75. bool WriteData(const char* buffer, int count, bool end_of_stream);
  76. // Sends buffers passed to WriteData().
  77. void Flush();
  78. // Cancels the request. The OnCanceled callback is invoked when request is
  79. // caneceled, and not other callbacks are invoked afterwards..
  80. void Cancel();
  81. // Releases all resources for the request and deletes the object itself.
  82. void Destroy();
  83. private:
  84. // States of BidirectionalStream are tracked in |read_state_| and
  85. // |write_state_|.
  86. // The write state is separated as it changes independently of the read state.
  87. // There is one initial state: NOT_STARTED. There is one normal final state:
  88. // SUCCESS, reached after READING_DONE and WRITING_DONE. There are two
  89. // exceptional final states: CANCELED and ERROR, which can be reached from
  90. // any other non-final state.
  91. enum State {
  92. // Initial state, stream not started.
  93. NOT_STARTED,
  94. // Stream started, request headers are being sent.
  95. STARTED,
  96. // Waiting for ReadData() to be called.
  97. WAITING_FOR_READ,
  98. // Reading from the remote, OnDataRead callback will be invoked when done.
  99. READING,
  100. // There is no more data to read and stream is half-closed by the remote
  101. // side.
  102. READING_DONE,
  103. // Stream is canceled.
  104. CANCELED,
  105. // Error has occured, stream is closed.
  106. ERR,
  107. // Reading and writing are done, and the stream is closed successfully.
  108. SUCCESS,
  109. // Waiting for Flush() to be called.
  110. WAITING_FOR_FLUSH,
  111. // Writing to the remote, callback will be invoked when done.
  112. WRITING,
  113. // There is no more data to write and stream is half-closed by the local
  114. // side.
  115. WRITING_DONE,
  116. };
  117. // Container to hold buffers and sizes of the pending data to be written.
  118. class WriteBuffers {
  119. public:
  120. WriteBuffers();
  121. WriteBuffers(const WriteBuffers&) = delete;
  122. WriteBuffers& operator=(const WriteBuffers&) = delete;
  123. ~WriteBuffers();
  124. // Clears Write Buffers list.
  125. void Clear();
  126. // Appends |buffer| of |buffer_size| length to the end of buffer list.
  127. void AppendBuffer(const scoped_refptr<net::IOBuffer>& buffer,
  128. int buffer_size);
  129. void MoveTo(WriteBuffers* target);
  130. // Returns true of Write Buffers list is empty.
  131. bool Empty() const;
  132. const std::vector<scoped_refptr<net::IOBuffer>>& buffers() const {
  133. return write_buffer_list;
  134. }
  135. const std::vector<int>& lengths() const { return write_buffer_len_list; }
  136. private:
  137. // Every IOBuffer in |write_buffer_list| points to the memory owned by the
  138. // application.
  139. std::vector<scoped_refptr<net::IOBuffer>> write_buffer_list;
  140. // A list of the length of each IOBuffer in |write_buffer_list|.
  141. std::vector<int> write_buffer_len_list;
  142. };
  143. // net::BidirectionalStream::Delegate implementations:
  144. void OnStreamReady(bool request_headers_sent) override;
  145. void OnHeadersReceived(
  146. const spdy::Http2HeaderBlock& response_headers) override;
  147. void OnDataRead(int bytes_read) override;
  148. void OnDataSent() override;
  149. void OnTrailersReceived(const spdy::Http2HeaderBlock& trailers) override;
  150. void OnFailed(int error) override;
  151. // Helper method to derive OnSucceeded.
  152. void MaybeOnSucceded();
  153. void StartOnNetworkThread(
  154. std::unique_ptr<net::BidirectionalStreamRequestInfo> request_info);
  155. void ReadDataOnNetworkThread(scoped_refptr<net::WrappedIOBuffer> read_buffer,
  156. int buffer_size);
  157. void WriteDataOnNetworkThread(scoped_refptr<net::WrappedIOBuffer> read_buffer,
  158. int buffer_size,
  159. bool end_of_stream);
  160. void FlushOnNetworkThread();
  161. void SendFlushingWriteData();
  162. void CancelOnNetworkThread();
  163. void DestroyOnNetworkThread();
  164. bool IsOnNetworkThread();
  165. void PostToNetworkThread(const base::Location& from_here,
  166. base::OnceClosure task);
  167. // Read state is tracking reading flow. Only accessed on network thread.
  168. // | <--- READING <--- |
  169. // | |
  170. // | |
  171. // NOT_STARTED -> STARTED --> WAITING_FOR_READ -> READING_DONE -> SUCCESS
  172. State read_state_;
  173. // Write state is tracking writing flow. Only accessed on network thread.
  174. // | <--- WRITING <--- |
  175. // | |
  176. // | |
  177. // NOT_STARTED -> STARTED --> WAITING_FOR_FLUSH -> WRITING_DONE -> SUCCESS
  178. State write_state_;
  179. bool write_end_of_stream_;
  180. bool request_headers_sent_;
  181. bool disable_auto_flush_;
  182. bool delay_headers_until_flush_;
  183. const raw_ptr<net::URLRequestContextGetter> request_context_getter_;
  184. scoped_refptr<net::WrappedIOBuffer> read_buffer_;
  185. // Write data that is pending the flush.
  186. std::unique_ptr<WriteBuffers> pending_write_data_;
  187. // Write data that is flushed, but not sending yet.
  188. std::unique_ptr<WriteBuffers> flushing_write_data_;
  189. // Write data that is sending.
  190. std::unique_ptr<WriteBuffers> sending_write_data_;
  191. std::unique_ptr<net::BidirectionalStream> bidi_stream_;
  192. raw_ptr<Delegate> delegate_;
  193. base::WeakPtr<BidirectionalStream> weak_this_;
  194. base::WeakPtrFactory<BidirectionalStream> weak_factory_{this};
  195. };
  196. } // namespace grpc_support
  197. #endif // COMPONENTS_GRPC_SUPPORT_BIDIRECTIONAL_STREAM_H_