| // Copyright 2013 The Chromium Authors |
| // Use of this source code is governed by a BSD-style license that can be |
| // found in the LICENSE file. |
| // |
| // Protobuf ZeroCopy[Input/Output]Stream implementations capable of using |
| // mojo data pipes. Built to work with Protobuf CodedStreams. |
| |
| #ifndef GOOGLE_APIS_GCM_BASE_SOCKET_STREAM_H_ |
| #define GOOGLE_APIS_GCM_BASE_SOCKET_STREAM_H_ |
| |
| #include <stdint.h> |
| |
| #include "base/compiler_specific.h" |
| #include "base/functional/callback.h" |
| #include "base/functional/callback_forward.h" |
| #include "base/memory/ref_counted.h" |
| #include "base/memory/weak_ptr.h" |
| #include "google/protobuf/io/zero_copy_stream.h" |
| #include "google_apis/gcm/base/gcm_export.h" |
| #include "mojo/public/cpp/system/data_pipe.h" |
| #include "mojo/public/cpp/system/simple_watcher.h" |
| #include "net/base/net_errors.h" |
| |
| namespace net { |
| class DrainableIOBuffer; |
| class IOBuffer; |
| class IOBufferWithSize; |
| } // namespace net |
| |
| namespace gcm { |
| |
| // A helper class for interacting with a mojo consumer pipe that is receiving |
| // protobuf encoded messages. If an error is encounters, the input stream will |
| // store the error in |last_error_|, and GetState() will be set to CLOSED. |
| // Typical usage: |
| // 1. Check the GetState() of the input stream before using it. If CLOSED, the |
| // input stream must be rebuilt (and the socket likely needs to be |
| // reconnected as an error was encountered). |
| // 2. If GetState() is EMPTY, call Refresh(..), passing the maximum byte size |
| // for a message, and wait until completion. It is invalid to attempt to |
| // Refresh an input stream or read data from the stream while a Refresh is |
| // pending. |
| // 3. Check GetState() again to ensure the Refresh was successful. |
| // 4. Use a CodedInputStream to read from the ZeroCopyInputStream interface of |
| // the SocketInputStream. Next(..) will return true until there is no data |
| // remaining. |
| // 5. Call RebuildBuffer when done reading, to shift any unread data to the |
| // start of the buffer. |
| // 6. Repeat as necessary. |
| class GCM_EXPORT SocketInputStream |
| : public google::protobuf::io::ZeroCopyInputStream { |
| public: |
| enum State { |
| // No valid data to read. This means the buffer is either empty or all data |
| // in the buffer has already been consumed. |
| EMPTY, |
| // Valid data to read. |
| READY, |
| // In the process of reading new data from the socket. |
| READING, |
| // An permanent error occurred and the stream is now closed. |
| CLOSED, |
| }; |
| |
| // |socket| should already be connected. |
| explicit SocketInputStream(mojo::ScopedDataPipeConsumerHandle stream); |
| |
| SocketInputStream(const SocketInputStream&) = delete; |
| SocketInputStream& operator=(const SocketInputStream&) = delete; |
| |
| ~SocketInputStream() override; |
| |
| // ZeroCopyInputStream implementation. |
| bool Next(const void** data, int* size) override; |
| void BackUp(int count) override; |
| bool Skip(int count) override; // Not implemented. |
| int64_t ByteCount() const override; |
| |
| // The remaining amount of valid data available to be read. |
| int UnreadByteCount() const; |
| |
| // Reads from the socket, appending a max of |byte_limit| bytes onto the read |
| // buffer. net::ERR_IO_PENDING is returned if the refresh can't complete |
| // synchronously, in which case the callback is invoked upon completion. If |
| // the refresh can complete synchronously, even in case of an error, returns |
| // net::OK without invoking callback. |
| // Note: GetState() (and possibly last_error()) should be checked upon |
| // completion to determine whether the Refresh encountered an error. |
| net::Error Refresh(base::OnceClosure callback, int byte_limit); |
| |
| // Rebuilds the buffer state by copying over any unread data to the beginning |
| // of the buffer and resetting the buffer read/write positions. |
| // Note: it is not valid to call Rebuild() if GetState() == CLOSED. The stream |
| // must be recreated from scratch in such a scenario. |
| void RebuildBuffer(); |
| |
| // Returns the last fatal error encountered. Only valid if GetState() == |
| // CLOSED. Note that all network read errors will be reported as |
| // net::ERR_FAILED, because mojo data pipe doesn't allow surfacing a more |
| // specific error code. |
| net::Error last_error() const; |
| |
| // Returns the current state. |
| State GetState() const; |
| |
| private: |
| // Clears the local state. |
| void ResetInternal(); |
| |
| void ReadMore(MojoResult result, const mojo::HandleSignalsState& state); |
| |
| // Permanently closes the stream. |
| void CloseStream(net::Error error); |
| |
| // Internal net components. |
| mojo::ScopedDataPipeConsumerHandle stream_; |
| mojo::SimpleWatcher stream_watcher_; |
| uint32_t read_size_; |
| const scoped_refptr<net::IOBuffer> io_buffer_; |
| base::OnceClosure read_callback_; |
| // IOBuffer implementation that wraps the data within |io_buffer_| that hasn't |
| // been written to yet by Socket::Read calls. |
| const scoped_refptr<net::DrainableIOBuffer> read_buffer_; |
| |
| // Starting position of the data within |io_buffer_| to consume on subsequent |
| // Next(..) call. 0 <= next_pos_ <= read_buffer_.BytesConsumed() |
| // Note: next_pos == read_buffer_.BytesConsumed() implies GetState() == EMPTY. |
| int next_pos_; |
| |
| // If < net::ERR_IO_PENDING, the last net error received. |
| // Note: last_error_ == net::ERR_IO_PENDING implies GetState() == READING. |
| net::Error last_error_; |
| |
| base::WeakPtrFactory<SocketInputStream> weak_ptr_factory_{this}; |
| }; |
| |
| // A helper class for writing to a mojo producer handle with protobuf encoded |
| // data. Typical usage: |
| // 1. Check the GetState() of the output stream before using it. If CLOSED, the |
| // output stream must be rebuilt (and the socket likely needs to be |
| // reconnected, as an error was encountered). |
| // 2. If EMPTY, the output stream can be written via a CodedOutputStream using |
| // the ZeroCopyOutputStream interface. |
| // 3. Once done writing, GetState() should be READY, so call Flush(..) to write |
| // the buffer into the mojo producer handle. Wait for the callback to be |
| // invoked (it's invalid to write to an output stream while it's flushing). |
| // 4. Check the GetState() again to ensure the Flush was successful. GetState() |
| // should be EMPTY again. |
| // 5. Repeat. |
| class GCM_EXPORT SocketOutputStream |
| : public google::protobuf::io::ZeroCopyOutputStream { |
| public: |
| enum State { |
| // No valid data yet. |
| EMPTY, |
| // Ready for flushing (some data is present). |
| READY, |
| // In the process of flushing into the socket. |
| FLUSHING, |
| // A permanent error occurred, and the stream is now closed. |
| CLOSED, |
| }; |
| |
| explicit SocketOutputStream(mojo::ScopedDataPipeProducerHandle stream); |
| |
| SocketOutputStream(const SocketOutputStream&) = delete; |
| SocketOutputStream& operator=(const SocketOutputStream&) = delete; |
| |
| ~SocketOutputStream() override; |
| |
| // ZeroCopyOutputStream implementation. |
| bool Next(void** data, int* size) override; |
| void BackUp(int count) override; |
| int64_t ByteCount() const override; |
| |
| // Writes the buffer into the Socket. |
| net::Error Flush(base::OnceClosure callback); |
| |
| // Returns the last fatal error encountered. Only valid if GetState() == |
| // CLOSED. Note that All network read errors will be reported as |
| // net::ERR_FAILED, because mojo data pipe doesn't allow surfacing a more |
| // specific error code. |
| net::Error last_error() const; |
| |
| // Returns the current state. |
| State GetState() const; |
| |
| private: |
| void WriteMore(MojoResult result, const mojo::HandleSignalsState& state); |
| |
| // Internal net components. |
| mojo::ScopedDataPipeProducerHandle stream_; |
| mojo::SimpleWatcher stream_watcher_; |
| base::OnceClosure write_callback_; |
| const scoped_refptr<net::IOBufferWithSize> io_buffer_; |
| // IOBuffer implementation that wraps the data within |io_buffer_| that hasn't |
| // been written to the socket yet. |
| scoped_refptr<net::DrainableIOBuffer> write_buffer_; |
| |
| // Starting position of the data within |io_buffer_| to consume on subsequent |
| // Next(..) call. 0 <= write_buffer_.BytesConsumed() <= next_pos_ |
| // Note: next_pos == 0 implies GetState() == EMPTY. |
| int next_pos_; |
| |
| // If < net::ERR_IO_PENDING, the last net error received. |
| // Note: last_error_ == net::ERR_IO_PENDING implies GetState() == FLUSHING. |
| net::Error last_error_; |
| |
| base::WeakPtrFactory<SocketOutputStream> weak_ptr_factory_{this}; |
| }; |
| |
| } // namespace gcm |
| |
| #endif // GOOGLE_APIS_GCM_BASE_SOCKET_STREAM_H_ |