socket_stream.h 8.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214
  1. // Copyright 2013 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. //
  5. // Protobuf ZeroCopy[Input/Output]Stream implementations capable of using
  6. // mojo data pipes. Built to work with Protobuf CodedStreams.
  7. #ifndef GOOGLE_APIS_GCM_BASE_SOCKET_STREAM_H_
  8. #define GOOGLE_APIS_GCM_BASE_SOCKET_STREAM_H_
  9. #include <stdint.h>
  10. #include "base/callback.h"
  11. #include "base/callback_forward.h"
  12. #include "base/compiler_specific.h"
  13. #include "base/memory/ref_counted.h"
  14. #include "base/memory/weak_ptr.h"
  15. #include "google/protobuf/io/zero_copy_stream.h"
  16. #include "google_apis/gcm/base/gcm_export.h"
  17. #include "mojo/public/cpp/system/data_pipe.h"
  18. #include "mojo/public/cpp/system/simple_watcher.h"
  19. #include "net/base/net_errors.h"
  20. namespace net {
  21. class DrainableIOBuffer;
  22. class IOBuffer;
  23. class IOBufferWithSize;
  24. } // namespace net
  25. namespace gcm {
  26. // A helper class for interacting with a mojo consumer pipe that is receiving
  27. // protobuf encoded messages. If an error is encounters, the input stream will
  28. // store the error in |last_error_|, and GetState() will be set to CLOSED.
  29. // Typical usage:
  30. // 1. Check the GetState() of the input stream before using it. If CLOSED, the
  31. // input stream must be rebuilt (and the socket likely needs to be
  32. // reconnected as an error was encountered).
  33. // 2. If GetState() is EMPTY, call Refresh(..), passing the maximum byte size
  34. // for a message, and wait until completion. It is invalid to attempt to
  35. // Refresh an input stream or read data from the stream while a Refresh is
  36. // pending.
  37. // 3. Check GetState() again to ensure the Refresh was successful.
  38. // 4. Use a CodedInputStream to read from the ZeroCopyInputStream interface of
  39. // the SocketInputStream. Next(..) will return true until there is no data
  40. // remaining.
  41. // 5. Call RebuildBuffer when done reading, to shift any unread data to the
  42. // start of the buffer.
  43. // 6. Repeat as necessary.
  44. class GCM_EXPORT SocketInputStream
  45. : public google::protobuf::io::ZeroCopyInputStream {
  46. public:
  47. enum State {
  48. // No valid data to read. This means the buffer is either empty or all data
  49. // in the buffer has already been consumed.
  50. EMPTY,
  51. // Valid data to read.
  52. READY,
  53. // In the process of reading new data from the socket.
  54. READING,
  55. // An permanent error occurred and the stream is now closed.
  56. CLOSED,
  57. };
  58. // |socket| should already be connected.
  59. explicit SocketInputStream(mojo::ScopedDataPipeConsumerHandle stream);
  60. SocketInputStream(const SocketInputStream&) = delete;
  61. SocketInputStream& operator=(const SocketInputStream&) = delete;
  62. ~SocketInputStream() override;
  63. // ZeroCopyInputStream implementation.
  64. bool Next(const void** data, int* size) override;
  65. void BackUp(int count) override;
  66. bool Skip(int count) override; // Not implemented.
  67. int64_t ByteCount() const override;
  68. // The remaining amount of valid data available to be read.
  69. int UnreadByteCount() const;
  70. // Reads from the socket, appending a max of |byte_limit| bytes onto the read
  71. // buffer. net::ERR_IO_PENDING is returned if the refresh can't complete
  72. // synchronously, in which case the callback is invoked upon completion. If
  73. // the refresh can complete synchronously, even in case of an error, returns
  74. // net::OK without invoking callback.
  75. // Note: GetState() (and possibly last_error()) should be checked upon
  76. // completion to determine whether the Refresh encountered an error.
  77. net::Error Refresh(base::OnceClosure callback, int byte_limit);
  78. // Rebuilds the buffer state by copying over any unread data to the beginning
  79. // of the buffer and resetting the buffer read/write positions.
  80. // Note: it is not valid to call Rebuild() if GetState() == CLOSED. The stream
  81. // must be recreated from scratch in such a scenario.
  82. void RebuildBuffer();
  83. // Returns the last fatal error encountered. Only valid if GetState() ==
  84. // CLOSED. Note that all network read errors will be reported as
  85. // net::ERR_FAILED, because mojo data pipe doesn't allow surfacing a more
  86. // specific error code.
  87. net::Error last_error() const;
  88. // Returns the current state.
  89. State GetState() const;
  90. private:
  91. // Clears the local state.
  92. void ResetInternal();
  93. void ReadMore(MojoResult result, const mojo::HandleSignalsState& state);
  94. // Permanently closes the stream.
  95. void CloseStream(net::Error error);
  96. // Internal net components.
  97. mojo::ScopedDataPipeConsumerHandle stream_;
  98. mojo::SimpleWatcher stream_watcher_;
  99. uint32_t read_size_;
  100. const scoped_refptr<net::IOBuffer> io_buffer_;
  101. base::OnceClosure read_callback_;
  102. // IOBuffer implementation that wraps the data within |io_buffer_| that hasn't
  103. // been written to yet by Socket::Read calls.
  104. const scoped_refptr<net::DrainableIOBuffer> read_buffer_;
  105. // Starting position of the data within |io_buffer_| to consume on subsequent
  106. // Next(..) call. 0 <= next_pos_ <= read_buffer_.BytesConsumed()
  107. // Note: next_pos == read_buffer_.BytesConsumed() implies GetState() == EMPTY.
  108. int next_pos_;
  109. // If < net::ERR_IO_PENDING, the last net error received.
  110. // Note: last_error_ == net::ERR_IO_PENDING implies GetState() == READING.
  111. net::Error last_error_;
  112. base::WeakPtrFactory<SocketInputStream> weak_ptr_factory_{this};
  113. };
  114. // A helper class for writing to a mojo producer handle with protobuf encoded
  115. // data. Typical usage:
  116. // 1. Check the GetState() of the output stream before using it. If CLOSED, the
  117. // output stream must be rebuilt (and the socket likely needs to be
  118. // reconnected, as an error was encountered).
  119. // 2. If EMPTY, the output stream can be written via a CodedOutputStream using
  120. // the ZeroCopyOutputStream interface.
  121. // 3. Once done writing, GetState() should be READY, so call Flush(..) to write
  122. // the buffer into the mojo producer handle. Wait for the callback to be
  123. // invoked (it's invalid to write to an output stream while it's flushing).
  124. // 4. Check the GetState() again to ensure the Flush was successful. GetState()
  125. // should be EMPTY again.
  126. // 5. Repeat.
  127. class GCM_EXPORT SocketOutputStream
  128. : public google::protobuf::io::ZeroCopyOutputStream {
  129. public:
  130. enum State {
  131. // No valid data yet.
  132. EMPTY,
  133. // Ready for flushing (some data is present).
  134. READY,
  135. // In the process of flushing into the socket.
  136. FLUSHING,
  137. // A permanent error occurred, and the stream is now closed.
  138. CLOSED,
  139. };
  140. explicit SocketOutputStream(mojo::ScopedDataPipeProducerHandle stream);
  141. SocketOutputStream(const SocketOutputStream&) = delete;
  142. SocketOutputStream& operator=(const SocketOutputStream&) = delete;
  143. ~SocketOutputStream() override;
  144. // ZeroCopyOutputStream implementation.
  145. bool Next(void** data, int* size) override;
  146. void BackUp(int count) override;
  147. int64_t ByteCount() const override;
  148. // Writes the buffer into the Socket.
  149. net::Error Flush(base::OnceClosure callback);
  150. // Returns the last fatal error encountered. Only valid if GetState() ==
  151. // CLOSED. Note that All network read errors will be reported as
  152. // net::ERR_FAILED, because mojo data pipe doesn't allow surfacing a more
  153. // specific error code.
  154. net::Error last_error() const;
  155. // Returns the current state.
  156. State GetState() const;
  157. private:
  158. void WriteMore(MojoResult result, const mojo::HandleSignalsState& state);
  159. // Internal net components.
  160. mojo::ScopedDataPipeProducerHandle stream_;
  161. mojo::SimpleWatcher stream_watcher_;
  162. base::OnceClosure write_callback_;
  163. const scoped_refptr<net::IOBufferWithSize> io_buffer_;
  164. // IOBuffer implementation that wraps the data within |io_buffer_| that hasn't
  165. // been written to the socket yet.
  166. scoped_refptr<net::DrainableIOBuffer> write_buffer_;
  167. // Starting position of the data within |io_buffer_| to consume on subsequent
  168. // Next(..) call. 0 <= write_buffer_.BytesConsumed() <= next_pos_
  169. // Note: next_pos == 0 implies GetState() == EMPTY.
  170. int next_pos_;
  171. // If < net::ERR_IO_PENDING, the last net error received.
  172. // Note: last_error_ == net::ERR_IO_PENDING implies GetState() == FLUSHING.
  173. net::Error last_error_;
  174. base::WeakPtrFactory<SocketOutputStream> weak_ptr_factory_{this};
  175. };
  176. } // namespace gcm
  177. #endif // GOOGLE_APIS_GCM_BASE_SOCKET_STREAM_H_