/* * * Copyright 2015 gRPC authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. * */ #ifndef GRPCXX_IMPL_CODEGEN_PROTO_UTILS_H #define GRPCXX_IMPL_CODEGEN_PROTO_UTILS_H #include #include #include #include #include #include #include #include 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_); if (have_backup_) { slice_ = backup_slice_; have_backup_ = false; } else { // When less than a whole block is needed, only allocate that much. // But make sure the allocated slice is not inlined. size_t remain = total_size_ - byte_count_ > block_size_ ? block_size_ : total_size_ - byte_count_; slice_ = g_core_codegen_interface->grpc_slice_malloc( remain > GRPC_SLICE_INLINED_SIZE ? remain : 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 Status GenericSerialize(const grpc::protobuf::Message& msg, grpc_byte_buffer** bp, bool* own_buffer) { static_assert( std::is_base_of::value, "BufferWriter must be a subclass of io::ZeroCopyOutputStream"); *own_buffer = true; int byte_size = msg.ByteSize(); if (byte_size == 0) { grpc_slice empty_slice = g_core_codegen_interface->grpc_empty_slice(); *bp = g_core_codegen_interface->grpc_raw_byte_buffer_create(&empty_slice, 1); return g_core_codegen_interface->ok(); } BufferWriter writer(bp, kGrpcBufferWriterMaxBufferLength, 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 Status GenericDeserialize(grpc_byte_buffer* buffer, grpc::protobuf::Message* msg) { static_assert( std::is_base_of::value, "BufferReader must be a subclass of io::ZeroCopyInputStream"); if (buffer == nullptr) { return Status(StatusCode::INTERNAL, "No payload"); } Status result = g_core_codegen_interface->ok(); { BufferReader reader(buffer); if (!reader.status().ok()) { return reader.status(); } ::grpc::protobuf::io::CodedInputStream decoder(&reader); decoder.SetTotalBytesLimit(INT_MAX, INT_MAX); if (!msg->ParseFromCodedStream(&decoder)) { result = Status(StatusCode::INTERNAL, msg->InitializationErrorString()); } if (!decoder.ConsumedEntireMessage()) { result = Status(StatusCode::INTERNAL, "Did not read entire message"); } } g_core_codegen_interface->grpc_byte_buffer_destroy(buffer); 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 // This class provides a protobuf serializer. It translates between protobuf // objects and grpc_byte_buffers. More information about SerializationTraits can // be found in include/grpc++/impl/codegen/serialization_traits.h. template class SerializationTraits::value>::type> { public: static Status Serialize(const grpc::protobuf::Message& msg, grpc_byte_buffer** bp, bool* own_buffer) { return internal::GenericSerialize( msg, bp, own_buffer); } static Status Deserialize(grpc_byte_buffer* buffer, grpc::protobuf::Message* msg) { return internal::GenericDeserialize(buffer, msg); } }; #endif } // namespace grpc #endif // GRPCXX_IMPL_CODEGEN_PROTO_UTILS_H