| // Protocol Buffers - Google's data interchange format |
| // Copyright 2008 Google Inc. All rights reserved. |
| // |
| // Use of this source code is governed by a BSD-style |
| // license that can be found in the LICENSE file or at |
| // https://developers.google.com/open-source/licenses/bsd |
| |
| // Author: kenton@google.com (Kenton Varda) |
| // Based on original Protocol Buffers design by |
| // Sanjay Ghemawat, Jeff Dean, and others. |
| // |
| // This file contains common implementations of the interfaces defined in |
| // zero_copy_stream.h which are included in the "lite" protobuf library. |
| // These implementations cover I/O on raw arrays and strings, as well as |
| // adaptors which make it easy to implement streams based on traditional |
| // streams. Of course, many users will probably want to write their own |
| // implementations of these interfaces specific to the particular I/O |
| // abstractions they prefer to use, but these should cover the most common |
| // cases. |
| |
| #ifndef GOOGLE_PROTOBUF_IO_ZERO_COPY_STREAM_IMPL_LITE_H__ |
| #define GOOGLE_PROTOBUF_IO_ZERO_COPY_STREAM_IMPL_LITE_H__ |
| |
| #include <iosfwd> |
| #include <memory> |
| #include <string> |
| #include <utility> |
| |
| #include "google/protobuf/stubs/callback.h" |
| #include "google/protobuf/stubs/common.h" |
| #include "absl/base/attributes.h" |
| #include "absl/base/macros.h" |
| #include "absl/strings/cord.h" |
| #include "absl/strings/cord_buffer.h" |
| #include "google/protobuf/io/zero_copy_stream.h" |
| #include "google/protobuf/port.h" |
| |
| |
| // Must be included last. |
| #include "google/protobuf/port_def.inc" |
| |
| namespace google { |
| namespace protobuf { |
| namespace io { |
| |
| // =================================================================== |
| |
| // A ZeroCopyInputStream backed by an in-memory array of bytes. |
| class PROTOBUF_EXPORT ArrayInputStream final : public ZeroCopyInputStream { |
| public: |
| // Create an InputStream that returns the bytes pointed to by "data". |
| // "data" remains the property of the caller but must remain valid until |
| // the stream is destroyed. If a block_size is given, calls to Next() |
| // will return data blocks no larger than the given size. Otherwise, the |
| // first call to Next() returns the entire array. block_size is mainly |
| // useful for testing; in production you would probably never want to set |
| // it. |
| ArrayInputStream(const void* data, int size, int block_size = -1); |
| ~ArrayInputStream() override = default; |
| |
| // `ArrayInputStream` is neither copiable nor assignable |
| ArrayInputStream(const ArrayInputStream&) = delete; |
| ArrayInputStream& operator=(const ArrayInputStream&) = delete; |
| |
| // implements ZeroCopyInputStream ---------------------------------- |
| bool Next(const void** data, int* size) override; |
| void BackUp(int count) override; |
| bool Skip(int count) override; |
| int64_t ByteCount() const override; |
| |
| |
| private: |
| const uint8_t* const data_; // The byte array. |
| const int size_; // Total size of the array. |
| const int block_size_; // How many bytes to return at a time. |
| |
| int position_; |
| int last_returned_size_; // How many bytes we returned last time Next() |
| // was called (used for error checking only). |
| }; |
| |
| // =================================================================== |
| |
| // A ZeroCopyOutputStream backed by an in-memory array of bytes. |
| class PROTOBUF_EXPORT ArrayOutputStream final : public ZeroCopyOutputStream { |
| public: |
| // Create an OutputStream that writes to the bytes pointed to by "data". |
| // "data" remains the property of the caller but must remain valid until |
| // the stream is destroyed. If a block_size is given, calls to Next() |
| // will return data blocks no larger than the given size. Otherwise, the |
| // first call to Next() returns the entire array. block_size is mainly |
| // useful for testing; in production you would probably never want to set |
| // it. |
| ArrayOutputStream(void* data, int size, int block_size = -1); |
| ~ArrayOutputStream() override = default; |
| |
| // `ArrayOutputStream` is neither copiable nor assignable |
| ArrayOutputStream(const ArrayOutputStream&) = delete; |
| ArrayOutputStream& operator=(const ArrayOutputStream&) = delete; |
| |
| // implements ZeroCopyOutputStream --------------------------------- |
| bool Next(void** data, int* size) override; |
| void BackUp(int count) override; |
| int64_t ByteCount() const override; |
| |
| private: |
| uint8_t* const data_; // The byte array. |
| const int size_; // Total size of the array. |
| const int block_size_; // How many bytes to return at a time. |
| |
| int position_; |
| int last_returned_size_; // How many bytes we returned last time Next() |
| // was called (used for error checking only). |
| }; |
| |
| // =================================================================== |
| |
| // A ZeroCopyOutputStream which appends bytes to a string. |
| class PROTOBUF_EXPORT StringOutputStream final : public ZeroCopyOutputStream { |
| public: |
| // Create a StringOutputStream which appends bytes to the given string. |
| // The string remains property of the caller, but it is mutated in arbitrary |
| // ways and MUST NOT be accessed in any way until you're done with the |
| // stream. Either be sure there's no further usage, or (safest) destroy the |
| // stream before using the contents. |
| // |
| // Hint: If you call target->reserve(n) before creating the stream, |
| // the first call to Next() will return at least n bytes of buffer |
| // space. |
| explicit StringOutputStream(std::string* target); |
| ~StringOutputStream() override = default; |
| |
| // `StringOutputStream` is neither copiable nor assignable |
| StringOutputStream(const StringOutputStream&) = delete; |
| StringOutputStream& operator=(const StringOutputStream&) = delete; |
| |
| // implements ZeroCopyOutputStream --------------------------------- |
| bool Next(void** data, int* size) override; |
| void BackUp(int count) override; |
| int64_t ByteCount() const override; |
| |
| private: |
| static constexpr size_t kMinimumSize = 16; |
| |
| std::string* target_; |
| }; |
| |
| // Note: There is no StringInputStream. Instead, just create an |
| // ArrayInputStream as follows: |
| // ArrayInputStream input(str.data(), str.size()); |
| |
| // =================================================================== |
| |
| // A generic traditional input stream interface. |
| // |
| // Lots of traditional input streams (e.g. file descriptors, C stdio |
| // streams, and C++ iostreams) expose an interface where every read |
| // involves copying bytes into a buffer. If you want to take such an |
| // interface and make a ZeroCopyInputStream based on it, simply implement |
| // CopyingInputStream and then use CopyingInputStreamAdaptor. |
| // |
| // CopyingInputStream implementations should avoid buffering if possible. |
| // CopyingInputStreamAdaptor does its own buffering and will read data |
| // in large blocks. |
| class PROTOBUF_EXPORT CopyingInputStream { |
| public: |
| virtual ~CopyingInputStream() {} |
| |
| // Reads up to "size" bytes into the given buffer. Returns the number of |
| // bytes read. Read() waits until at least one byte is available, or |
| // returns zero if no bytes will ever become available (EOF), or -1 if a |
| // permanent read error occurred. |
| virtual int Read(void* buffer, int size) = 0; |
| |
| // Skips the next "count" bytes of input. Returns the number of bytes |
| // actually skipped. This will always be exactly equal to "count" unless |
| // EOF was reached or a permanent read error occurred. |
| // |
| // The default implementation just repeatedly calls Read() into a scratch |
| // buffer. |
| virtual int Skip(int count); |
| }; |
| |
| // A ZeroCopyInputStream which reads from a CopyingInputStream. This is |
| // useful for implementing ZeroCopyInputStreams that read from traditional |
| // streams. Note that this class is not really zero-copy. |
| // |
| // If you want to read from file descriptors or C++ istreams, this is |
| // already implemented for you: use FileInputStream or IstreamInputStream |
| // respectively. |
| class PROTOBUF_EXPORT CopyingInputStreamAdaptor : public ZeroCopyInputStream { |
| public: |
| // Creates a stream that reads from the given CopyingInputStream. |
| // If a block_size is given, it specifies the number of bytes that |
| // should be read and returned with each call to Next(). Otherwise, |
| // a reasonable default is used. The caller retains ownership of |
| // copying_stream unless SetOwnsCopyingStream(true) is called. |
| explicit CopyingInputStreamAdaptor(CopyingInputStream* copying_stream, |
| int block_size = -1); |
| ~CopyingInputStreamAdaptor() override; |
| |
| // `CopyingInputStreamAdaptor` is neither copiable nor assignable |
| CopyingInputStreamAdaptor(const CopyingInputStreamAdaptor&) = delete; |
| CopyingInputStreamAdaptor& operator=(const CopyingInputStreamAdaptor&) = delete; |
| |
| // Call SetOwnsCopyingStream(true) to tell the CopyingInputStreamAdaptor to |
| // delete the underlying CopyingInputStream when it is destroyed. |
| void SetOwnsCopyingStream(bool value) { owns_copying_stream_ = value; } |
| |
| // implements ZeroCopyInputStream ---------------------------------- |
| bool Next(const void** data, int* size) override; |
| void BackUp(int count) override; |
| bool Skip(int count) override; |
| int64_t ByteCount() const override; |
| |
| private: |
| // Insures that buffer_ is not NULL. |
| void AllocateBufferIfNeeded(); |
| // Frees the buffer and resets buffer_used_. |
| void FreeBuffer(); |
| |
| // The underlying copying stream. |
| CopyingInputStream* copying_stream_; |
| bool owns_copying_stream_; |
| |
| // True if we have seen a permanent error from the underlying stream. |
| bool failed_; |
| |
| // The current position of copying_stream_, relative to the point where |
| // we started reading. |
| int64_t position_; |
| |
| // Data is read into this buffer. It may be NULL if no buffer is currently |
| // in use. Otherwise, it points to an array of size buffer_size_. |
| std::unique_ptr<uint8_t[]> buffer_; |
| const int buffer_size_; |
| |
| // Number of valid bytes currently in the buffer (i.e. the size last |
| // returned by Next()). 0 <= buffer_used_ <= buffer_size_. |
| int buffer_used_; |
| |
| // Number of bytes in the buffer which were backed up over by a call to |
| // BackUp(). These need to be returned again. |
| // 0 <= backup_bytes_ <= buffer_used_ |
| int backup_bytes_; |
| }; |
| |
| // =================================================================== |
| |
| // A generic traditional output stream interface. |
| // |
| // Lots of traditional output streams (e.g. file descriptors, C stdio |
| // streams, and C++ iostreams) expose an interface where every write |
| // involves copying bytes from a buffer. If you want to take such an |
| // interface and make a ZeroCopyOutputStream based on it, simply implement |
| // CopyingOutputStream and then use CopyingOutputStreamAdaptor. |
| // |
| // CopyingOutputStream implementations should avoid buffering if possible. |
| // CopyingOutputStreamAdaptor does its own buffering and will write data |
| // in large blocks. |
| class PROTOBUF_EXPORT CopyingOutputStream { |
| public: |
| virtual ~CopyingOutputStream() {} |
| |
| // Writes "size" bytes from the given buffer to the output. Returns true |
| // if successful, false on a write error. |
| virtual bool Write(const void* buffer, int size) = 0; |
| }; |
| |
| // A ZeroCopyOutputStream which writes to a CopyingOutputStream. This is |
| // useful for implementing ZeroCopyOutputStreams that write to traditional |
| // streams. Note that this class is not really zero-copy. |
| // |
| // If you want to write to file descriptors or C++ ostreams, this is |
| // already implemented for you: use FileOutputStream or OstreamOutputStream |
| // respectively. |
| class PROTOBUF_EXPORT CopyingOutputStreamAdaptor : public ZeroCopyOutputStream { |
| public: |
| // Creates a stream that writes to the given Unix file descriptor. |
| // If a block_size is given, it specifies the size of the buffers |
| // that should be returned by Next(). Otherwise, a reasonable default |
| // is used. |
| explicit CopyingOutputStreamAdaptor(CopyingOutputStream* copying_stream, |
| int block_size = -1); |
| ~CopyingOutputStreamAdaptor() override; |
| |
| // `CopyingOutputStreamAdaptor` is neither copiable nor assignable |
| CopyingOutputStreamAdaptor(const CopyingOutputStreamAdaptor&) = delete; |
| CopyingOutputStreamAdaptor& operator=(const CopyingOutputStreamAdaptor&) = delete; |
| |
| // Writes all pending data to the underlying stream. Returns false if a |
| // write error occurred on the underlying stream. (The underlying |
| // stream itself is not necessarily flushed.) |
| bool Flush(); |
| |
| // Call SetOwnsCopyingStream(true) to tell the CopyingOutputStreamAdaptor to |
| // delete the underlying CopyingOutputStream when it is destroyed. |
| void SetOwnsCopyingStream(bool value) { owns_copying_stream_ = value; } |
| |
| // implements ZeroCopyOutputStream --------------------------------- |
| bool Next(void** data, int* size) override; |
| void BackUp(int count) override; |
| int64_t ByteCount() const override; |
| bool WriteAliasedRaw(const void* data, int size) override; |
| bool AllowsAliasing() const override { return true; } |
| bool WriteCord(const absl::Cord& cord) override; |
| |
| private: |
| // Write the current buffer, if it is present. |
| bool WriteBuffer(); |
| // Insures that buffer_ is not NULL. |
| void AllocateBufferIfNeeded(); |
| // Frees the buffer. |
| void FreeBuffer(); |
| |
| // The underlying copying stream. |
| CopyingOutputStream* copying_stream_; |
| bool owns_copying_stream_; |
| |
| // True if we have seen a permanent error from the underlying stream. |
| bool failed_; |
| |
| // The current position of copying_stream_, relative to the point where |
| // we started writing. |
| int64_t position_; |
| |
| // Data is written from this buffer. It may be NULL if no buffer is |
| // currently in use. Otherwise, it points to an array of size buffer_size_. |
| std::unique_ptr<uint8_t[]> buffer_; |
| const int buffer_size_; |
| |
| // Number of valid bytes currently in the buffer (i.e. the size last |
| // returned by Next()). When BackUp() is called, we just reduce this. |
| // 0 <= buffer_used_ <= buffer_size_. |
| int buffer_used_; |
| }; |
| |
| // =================================================================== |
| |
| // A ZeroCopyInputStream which wraps some other stream and limits it to |
| // a particular byte count. |
| class PROTOBUF_EXPORT LimitingInputStream final : public ZeroCopyInputStream { |
| public: |
| LimitingInputStream(ZeroCopyInputStream* input, int64_t limit); |
| ~LimitingInputStream() override; |
| |
| // `LimitingInputStream` is neither copiable nor assignable |
| LimitingInputStream(const LimitingInputStream&) = delete; |
| LimitingInputStream& operator=(const LimitingInputStream&) = delete; |
| |
| // implements ZeroCopyInputStream ---------------------------------- |
| bool Next(const void** data, int* size) override; |
| void BackUp(int count) override; |
| bool Skip(int count) override; |
| int64_t ByteCount() const override; |
| bool ReadCord(absl::Cord* cord, int count) override; |
| |
| |
| private: |
| ZeroCopyInputStream* input_; |
| int64_t limit_; // Decreases as we go, becomes negative if we overshoot. |
| int64_t prior_bytes_read_; // Bytes read on underlying stream at construction |
| }; |
| |
| // =================================================================== |
| |
| // A ZeroCopyInputStream backed by a Cord. This stream implements ReadCord() |
| // in a way that can share memory between the source and destination cords |
| // rather than copying. |
| class PROTOBUF_EXPORT CordInputStream final : public ZeroCopyInputStream { |
| public: |
| // Creates an InputStream that reads from the given Cord. `cord` must |
| // not be null and must outlive this CordInputStream instance. `cord` must |
| // not be modified while this instance is actively being used: any change |
| // to `cord` will lead to undefined behavior on any subsequent call into |
| // this instance. |
| explicit CordInputStream( |
| const absl::Cord* cord ABSL_ATTRIBUTE_LIFETIME_BOUND); |
| |
| |
| // `CordInputStream` is neither copiable nor assignable |
| CordInputStream(const CordInputStream&) = delete; |
| CordInputStream& operator=(const CordInputStream&) = delete; |
| |
| // implements ZeroCopyInputStream ---------------------------------- |
| bool Next(const void** data, int* size) override; |
| void BackUp(int count) override; |
| bool Skip(int count) override; |
| int64_t ByteCount() const override; |
| bool ReadCord(absl::Cord* cord, int count) override; |
| |
| |
| private: |
| // Moves `it_` to the next available chunk skipping `skip` extra bytes |
| // and updates the chunk data pointers. |
| bool NextChunk(size_t skip); |
| |
| // Updates the current chunk data context `data_`, `size_` and `available_`. |
| // If `bytes_remaining_` is zero, sets `size_` and `available_` to zero. |
| // Returns true if more data is available, false otherwise. |
| bool LoadChunkData(); |
| |
| absl::Cord::CharIterator it_; |
| size_t length_; |
| size_t bytes_remaining_; |
| const char* data_; |
| size_t size_; |
| size_t available_; |
| }; |
| |
| // =================================================================== |
| |
| // A ZeroCopyOutputStream that writes to a Cord. This stream implements |
| // WriteCord() in a way that can share memory between the source and |
| // destination cords rather than copying. |
| class PROTOBUF_EXPORT CordOutputStream final : public ZeroCopyOutputStream { |
| public: |
| // Creates an OutputStream streaming serialized data into a Cord. `size_hint`, |
| // if given, is the expected total size of the resulting Cord. This is a hint |
| // only, used for optimization. Callers can obtain the generated Cord value by |
| // invoking `Consume()`. |
| explicit CordOutputStream(size_t size_hint = 0); |
| |
| // Creates an OutputStream with an initial Cord value. This constructor can be |
| // used by applications wanting to directly append serialization data to a |
| // given cord. In such cases, donating the existing value as in: |
| // |
| // CordOutputStream stream(std::move(cord)); |
| // message.SerializeToZeroCopyStream(&stream); |
| // cord = std::move(stream.Consume()); |
| // |
| // is more efficient then appending the serialized cord in application code: |
| // |
| // CordOutputStream stream; |
| // message.SerializeToZeroCopyStream(&stream); |
| // cord.Append(stream.Consume()); |
| // |
| // The former allows `CordOutputStream` to utilize pre-existing privately |
| // owned Cord buffers from the donated cord where the latter does not, which |
| // may lead to more memory usage when serialuzing data into existing cords. |
| explicit CordOutputStream(absl::Cord cord, size_t size_hint = 0); |
| |
| // Creates an OutputStream with an initial Cord value and initial buffer. |
| // This donates both the preexisting cord in `cord`, as well as any |
| // pre-existing data and additional capacity in `buffer`. |
| // This function is mainly intended to be used in internal serialization logic |
| // using eager buffer initialization in EpsCopyOutputStream. |
| // The donated buffer can be empty, partially empty or full: the outputstream |
| // will DTRT in all cases and preserve any pre-existing data. |
| explicit CordOutputStream(absl::Cord cord, absl::CordBuffer buffer, |
| size_t size_hint = 0); |
| |
| // Creates an OutputStream with an initial buffer. |
| // This method is logically identical to, but more efficient than: |
| // `CordOutputStream(absl::Cord(), std::move(buffer), size_hint)` |
| explicit CordOutputStream(absl::CordBuffer buffer, size_t size_hint = 0); |
| |
| // `CordOutputStream` is neither copiable nor assignable |
| CordOutputStream(const CordOutputStream&) = delete; |
| CordOutputStream& operator=(const CordOutputStream&) = delete; |
| |
| // implements `ZeroCopyOutputStream` --------------------------------- |
| bool Next(void** data, int* size) final; |
| void BackUp(int count) final; |
| int64_t ByteCount() const final; |
| bool WriteCord(const absl::Cord& cord) final; |
| |
| // Consumes the serialized data as a cord value. `Consume()` internally |
| // flushes any pending state 'as if' BackUp(0) was called. While a final call |
| // to BackUp() is generally required by the `ZeroCopyOutputStream` contract, |
| // applications using `CordOutputStream` directly can call `Consume()` without |
| // a preceding call to `BackUp()`. |
| // |
| // While it will rarely be useful in practice (and especially in the presence |
| // of size hints) an instance is safe to be used after a call to `Consume()`. |
| // The only logical change in state is that all serialized data is extracted, |
| // and any new serialization calls will serialize into new cord data. |
| absl::Cord Consume(); |
| |
| private: |
| // State of `buffer_` and 'cord_. As a default CordBuffer instance always has |
| // inlined capacity, we track state explicitly to avoid returning 'existing |
| // capacity' from the default or 'moved from' CordBuffer. 'kSteal' indicates |
| // we should (attempt to) steal the next buffer from the cord. |
| enum class State { kEmpty, kFull, kPartial, kSteal }; |
| |
| absl::Cord cord_; |
| size_t size_hint_; |
| State state_ = State::kEmpty; |
| absl::CordBuffer buffer_; |
| }; |
| |
| |
| // =================================================================== |
| |
| // Return a pointer to mutable characters underlying the given string. The |
| // return value is valid until the next time the string is resized. We |
| // trust the caller to treat the return value as an array of length s->size(). |
| inline char* mutable_string_data(std::string* s) { |
| return &(*s)[0]; |
| } |
| |
| // as_string_data(s) is equivalent to |
| // ({ char* p = mutable_string_data(s); make_pair(p, p != NULL); }) |
| // Sometimes it's faster: in some scenarios p cannot be NULL, and then the |
| // code can avoid that check. |
| inline std::pair<char*, bool> as_string_data(std::string* s) { |
| char* p = mutable_string_data(s); |
| return std::make_pair(p, true); |
| } |
| |
| } // namespace io |
| } // namespace protobuf |
| } // namespace google |
| |
| #include "google/protobuf/port_undef.inc" |
| |
| #endif // GOOGLE_PROTOBUF_IO_ZERO_COPY_STREAM_IMPL_LITE_H__ |