diff options
Diffstat (limited to 'include/grpcpp/impl/codegen/proto_utils.h')
| -rw-r--r-- | include/grpcpp/impl/codegen/proto_utils.h | 219 |
1 files changed, 36 insertions, 183 deletions
diff --git a/include/grpcpp/impl/codegen/proto_utils.h b/include/grpcpp/impl/codegen/proto_utils.h index 81438ee1d5..d9db6de05c 100644 --- a/include/grpcpp/impl/codegen/proto_utils.h +++ b/include/grpcpp/impl/codegen/proto_utils.h @@ -24,203 +24,61 @@ #include <grpc/impl/codegen/byte_buffer_reader.h> #include <grpc/impl/codegen/grpc_types.h> #include <grpc/impl/codegen/slice.h> +#include <grpcpp/impl/codegen/byte_buffer.h> #include <grpcpp/impl/codegen/config_protobuf.h> #include <grpcpp/impl/codegen/core_codegen_interface.h> +#include <grpcpp/impl/codegen/proto_buffer_reader.h> +#include <grpcpp/impl/codegen/proto_buffer_writer.h> #include <grpcpp/impl/codegen/serialization_traits.h> +#include <grpcpp/impl/codegen/slice.h> #include <grpcpp/impl/codegen/status.h> +/// This header provides serialization and deserialization between gRPC +/// messages serialized using protobuf and the C++ objects they represent. + namespace grpc { extern CoreCodegenInterface* g_core_codegen_interface; -namespace internal { - -class GrpcBufferWriterPeer; - -const int kGrpcBufferWriterMaxBufferLength = 1024 * 1024; - -class GrpcBufferWriter : public ::grpc::protobuf::io::ZeroCopyOutputStream { - public: - GrpcBufferWriter(grpc_byte_buffer** bp, int block_size, int total_size) - : block_size_(block_size), - total_size_(total_size), - byte_count_(0), - have_backup_(false) { - *bp = g_core_codegen_interface->grpc_raw_byte_buffer_create(NULL, 0); - slice_buffer_ = &(*bp)->data.raw.slice_buffer; - } - - ~GrpcBufferWriter() override { - if (have_backup_) { - g_core_codegen_interface->grpc_slice_unref(backup_slice_); - } - } - - bool Next(void** data, int* size) override { - // Protobuf should not ask for more memory than total_size_. - GPR_CODEGEN_ASSERT(byte_count_ < total_size_); - size_t remain = total_size_ - byte_count_; - if (have_backup_) { - slice_ = backup_slice_; - have_backup_ = false; - if (GRPC_SLICE_LENGTH(slice_) > remain) { - GRPC_SLICE_SET_LENGTH(slice_, remain); - } - } else { - // When less than a whole block is needed, only allocate that much. - // But make sure the allocated slice is not inlined. - size_t allocate_length = - remain > static_cast<size_t>(block_size_) ? block_size_ : remain; - slice_ = g_core_codegen_interface->grpc_slice_malloc( - allocate_length > GRPC_SLICE_INLINED_SIZE - ? allocate_length - : GRPC_SLICE_INLINED_SIZE + 1); - } - *data = GRPC_SLICE_START_PTR(slice_); - // On win x64, int is only 32bit - GPR_CODEGEN_ASSERT(GRPC_SLICE_LENGTH(slice_) <= INT_MAX); - byte_count_ += * size = (int)GRPC_SLICE_LENGTH(slice_); - g_core_codegen_interface->grpc_slice_buffer_add(slice_buffer_, slice_); - return true; - } - - void BackUp(int count) override { - g_core_codegen_interface->grpc_slice_buffer_pop(slice_buffer_); - if ((size_t)count == GRPC_SLICE_LENGTH(slice_)) { - backup_slice_ = slice_; - } else { - backup_slice_ = g_core_codegen_interface->grpc_slice_split_tail( - &slice_, GRPC_SLICE_LENGTH(slice_) - count); - g_core_codegen_interface->grpc_slice_buffer_add(slice_buffer_, slice_); - } - // It's dangerous to keep an inlined grpc_slice as the backup slice, since - // on a following Next() call, a reference will be returned to this slice - // via GRPC_SLICE_START_PTR, which will not be an adddress held by - // slice_buffer_. - have_backup_ = backup_slice_.refcount != NULL; - byte_count_ -= count; - } - - grpc::protobuf::int64 ByteCount() const override { return byte_count_; } - - protected: - friend class GrpcBufferWriterPeer; - const int block_size_; - const int total_size_; - int64_t byte_count_; - grpc_slice_buffer* slice_buffer_; - bool have_backup_; - grpc_slice backup_slice_; - grpc_slice slice_; -}; - -class GrpcBufferReader : public ::grpc::protobuf::io::ZeroCopyInputStream { - public: - explicit GrpcBufferReader(grpc_byte_buffer* buffer) - : byte_count_(0), backup_count_(0), status_() { - if (!g_core_codegen_interface->grpc_byte_buffer_reader_init(&reader_, - buffer)) { - status_ = Status(StatusCode::INTERNAL, - "Couldn't initialize byte buffer reader"); - } - } - ~GrpcBufferReader() override { - g_core_codegen_interface->grpc_byte_buffer_reader_destroy(&reader_); - } - - bool Next(const void** data, int* size) override { - if (!status_.ok()) { - return false; - } - if (backup_count_ > 0) { - *data = GRPC_SLICE_START_PTR(slice_) + GRPC_SLICE_LENGTH(slice_) - - backup_count_; - GPR_CODEGEN_ASSERT(backup_count_ <= INT_MAX); - *size = (int)backup_count_; - backup_count_ = 0; - return true; - } - if (!g_core_codegen_interface->grpc_byte_buffer_reader_next(&reader_, - &slice_)) { - return false; - } - g_core_codegen_interface->grpc_slice_unref(slice_); - *data = GRPC_SLICE_START_PTR(slice_); - // On win x64, int is only 32bit - GPR_CODEGEN_ASSERT(GRPC_SLICE_LENGTH(slice_) <= INT_MAX); - byte_count_ += * size = (int)GRPC_SLICE_LENGTH(slice_); - return true; - } - - Status status() const { return status_; } - - void BackUp(int count) override { backup_count_ = count; } - - bool Skip(int count) override { - const void* data; - int size; - while (Next(&data, &size)) { - if (size >= count) { - BackUp(size - count); - return true; - } - // size < count; - count -= size; - } - // error or we have too large count; - return false; - } - - grpc::protobuf::int64 ByteCount() const override { - return byte_count_ - backup_count_; - } - - protected: - int64_t byte_count_; - int64_t backup_count_; - grpc_byte_buffer_reader reader_; - grpc_slice slice_; - Status status_; -}; - -// BufferWriter must be a subclass of io::ZeroCopyOutputStream. -template <class BufferWriter, class T> -Status GenericSerialize(const grpc::protobuf::Message& msg, - grpc_byte_buffer** bp, bool* own_buffer) { - static_assert( - std::is_base_of<protobuf::io::ZeroCopyOutputStream, BufferWriter>::value, - "BufferWriter must be a subclass of io::ZeroCopyOutputStream"); +// ProtoBufferWriter must be a subclass of ::protobuf::io::ZeroCopyOutputStream. +template <class ProtoBufferWriter, class T> +Status GenericSerialize(const grpc::protobuf::Message& msg, ByteBuffer* bb, + bool* own_buffer) { + static_assert(std::is_base_of<protobuf::io::ZeroCopyOutputStream, + ProtoBufferWriter>::value, + "ProtoBufferWriter must be a subclass of " + "::protobuf::io::ZeroCopyOutputStream"); *own_buffer = true; int byte_size = msg.ByteSize(); if ((size_t)byte_size <= GRPC_SLICE_INLINED_SIZE) { - grpc_slice slice = g_core_codegen_interface->grpc_slice_malloc(byte_size); - GPR_CODEGEN_ASSERT( - GRPC_SLICE_END_PTR(slice) == - msg.SerializeWithCachedSizesToArray(GRPC_SLICE_START_PTR(slice))); - *bp = g_core_codegen_interface->grpc_raw_byte_buffer_create(&slice, 1); - g_core_codegen_interface->grpc_slice_unref(slice); + Slice slice(byte_size); + // We serialize directly into the allocated slices memory + GPR_CODEGEN_ASSERT(slice.end() == msg.SerializeWithCachedSizesToArray( + const_cast<uint8_t*>(slice.begin()))); + ByteBuffer tmp(&slice, 1); + bb->Swap(&tmp); return g_core_codegen_interface->ok(); } - BufferWriter writer(bp, kGrpcBufferWriterMaxBufferLength, byte_size); + ProtoBufferWriter writer(bb, kProtoBufferWriterMaxBufferLength, byte_size); return msg.SerializeToZeroCopyStream(&writer) ? g_core_codegen_interface->ok() : Status(StatusCode::INTERNAL, "Failed to serialize message"); } -// BufferReader must be a subclass of io::ZeroCopyInputStream. -template <class BufferReader, class T> -Status GenericDeserialize(grpc_byte_buffer* buffer, - grpc::protobuf::Message* msg) { - static_assert( - std::is_base_of<protobuf::io::ZeroCopyInputStream, BufferReader>::value, - "BufferReader must be a subclass of io::ZeroCopyInputStream"); +// BufferReader must be a subclass of ::protobuf::io::ZeroCopyInputStream. +template <class ProtoBufferReader, class T> +Status GenericDeserialize(ByteBuffer* buffer, grpc::protobuf::Message* msg) { + static_assert(std::is_base_of<protobuf::io::ZeroCopyInputStream, + ProtoBufferReader>::value, + "ProtoBufferReader must be a subclass of " + "::protobuf::io::ZeroCopyInputStream"); if (buffer == nullptr) { return Status(StatusCode::INTERNAL, "No payload"); } Status result = g_core_codegen_interface->ok(); { - BufferReader reader(buffer); + ProtoBufferReader reader(buffer); if (!reader.status().ok()) { return reader.status(); } @@ -233,12 +91,10 @@ Status GenericDeserialize(grpc_byte_buffer* buffer, result = Status(StatusCode::INTERNAL, "Did not read entire message"); } } - g_core_codegen_interface->grpc_byte_buffer_destroy(buffer); + buffer->Clear(); return result; } -} // namespace internal - // this is needed so the following class does not conflict with protobuf // serializers that utilize internal-only tools. #ifdef GRPC_OPEN_SOURCE_PROTO @@ -249,16 +105,13 @@ template <class T> class SerializationTraits<T, typename std::enable_if<std::is_base_of< grpc::protobuf::Message, T>::value>::type> { public: - static Status Serialize(const grpc::protobuf::Message& msg, - grpc_byte_buffer** bp, bool* own_buffer) { - return internal::GenericSerialize<internal::GrpcBufferWriter, T>( - msg, bp, own_buffer); + static Status Serialize(const grpc::protobuf::Message& msg, ByteBuffer* bb, + bool* own_buffer) { + return GenericSerialize<ProtoBufferWriter, T>(msg, bb, own_buffer); } - static Status Deserialize(grpc_byte_buffer* buffer, - grpc::protobuf::Message* msg) { - return internal::GenericDeserialize<internal::GrpcBufferReader, T>(buffer, - msg); + static Status Deserialize(ByteBuffer* buffer, grpc::protobuf::Message* msg) { + return GenericDeserialize<ProtoBufferReader, T>(buffer, msg); } }; #endif |