aboutsummaryrefslogtreecommitdiffhomepage
path: root/include/grpcpp/impl
diff options
context:
space:
mode:
authorGravatar ncteisen <ncteisen@gmail.com>2018-02-27 11:55:10 -0800
committerGravatar ncteisen <ncteisen@gmail.com>2018-04-06 07:05:27 -0700
commitcf5b2666e16f7b6ebc1bda4a884045d1ef1a4134 (patch)
treeb945a65fa777134898ec029cfc094d69b1190024 /include/grpcpp/impl
parent1c73e2c53ddecf3bd45f1c28d23a895fc7cd6e12 (diff)
Make GrpcProtoBufferReaderWriter public
Diffstat (limited to 'include/grpcpp/impl')
-rw-r--r--include/grpcpp/impl/codegen/byte_buffer.h43
-rw-r--r--include/grpcpp/impl/codegen/core_codegen.h3
-rw-r--r--include/grpcpp/impl/codegen/core_codegen_interface.h5
-rw-r--r--include/grpcpp/impl/codegen/proto_buffer_reader.h151
-rw-r--r--include/grpcpp/impl/codegen/proto_buffer_writer.h168
-rw-r--r--include/grpcpp/impl/codegen/proto_utils.h220
-rw-r--r--include/grpcpp/impl/codegen/slice.h39
7 files changed, 431 insertions, 198 deletions
diff --git a/include/grpcpp/impl/codegen/byte_buffer.h b/include/grpcpp/impl/codegen/byte_buffer.h
index e637efeb85..fda13a04e6 100644
--- a/include/grpcpp/impl/codegen/byte_buffer.h
+++ b/include/grpcpp/impl/codegen/byte_buffer.h
@@ -32,6 +32,8 @@
namespace grpc {
class ServerInterface;
+class ByteBuffer;
+class ServerInterface;
namespace internal {
class CallOpSendMessage;
@@ -45,6 +47,7 @@ template <class ServiceType, class RequestType, class ResponseType>
class ServerStreamingHandler;
template <class R>
class DeserializeFuncType;
+class GrpcByteBufferPeer;
} // namespace internal
/// A sequence of bytes.
class ByteBuffer final {
@@ -53,7 +56,30 @@ class ByteBuffer final {
ByteBuffer() : buffer_(nullptr) {}
/// Construct buffer from \a slices, of which there are \a nslices.
- ByteBuffer(const Slice* slices, size_t nslices);
+ ByteBuffer(const Slice* slices, size_t nslices) {
+ // The following assertions check that the representation of a grpc::Slice
+ // is identical to that of a grpc_slice: it has a grpc_slice field, and
+ // nothing else.
+ static_assert(std::is_same<decltype(slices[0].slice_), grpc_slice>::value,
+ "Slice must have same representation as grpc_slice");
+ static_assert(sizeof(Slice) == sizeof(grpc_slice),
+ "Slice must have same representation as grpc_slice");
+ // The following assertions check that the representation of a ByteBuffer is
+ // identical to grpc_byte_buffer*: it has a grpc_byte_buffer* field,
+ // and nothing else.
+ static_assert(std::is_same<decltype(buffer_), grpc_byte_buffer*>::value,
+ "ByteBuffer must have same representation as "
+ "grpc_byte_buffer*");
+ static_assert(sizeof(ByteBuffer) == sizeof(grpc_byte_buffer*),
+ "ByteBuffer must have same representation as "
+ "grpc_byte_buffer*");
+ // The const_cast is legal if grpc_raw_byte_buffer_create() does no more
+ // than its advertised side effect of increasing the reference count of the
+ // slices it processes, and such an increase does not affect the semantics
+ // seen by the caller of this constructor.
+ buffer_ = g_core_codegen_interface->grpc_raw_byte_buffer_create(
+ reinterpret_cast<grpc_slice*>(const_cast<Slice*>(slices)), nslices);
+ }
/// Constuct a byte buffer by referencing elements of existing buffer
/// \a buf. Wrapper of core function grpc_byte_buffer_copy
@@ -90,10 +116,18 @@ class ByteBuffer final {
void Release() { buffer_ = nullptr; }
/// Buffer size in bytes.
- size_t Length() const;
+ size_t Length() const {
+ return buffer_ == nullptr
+ ? 0
+ : g_core_codegen_interface->grpc_byte_buffer_length(buffer_);
+ }
/// Swap the state of *this and *other.
- void Swap(ByteBuffer* other);
+ void Swap(ByteBuffer* other) {
+ grpc_byte_buffer* tmp = other->buffer_;
+ other->buffer_ = buffer_;
+ buffer_ = tmp;
+ }
/// Is this ByteBuffer valid?
bool Valid() const { return (buffer_ != nullptr); }
@@ -112,6 +146,9 @@ class ByteBuffer final {
friend class internal::ServerStreamingHandler;
template <class R>
friend class internal::DeserializeFuncType;
+ friend class GrpcProtoBufferReader;
+ friend class GrpcProtoBufferWriter;
+ friend class internal::GrpcByteBufferPeer;
grpc_byte_buffer* buffer_;
diff --git a/include/grpcpp/impl/codegen/core_codegen.h b/include/grpcpp/impl/codegen/core_codegen.h
index 0ca4ad524c..e9df96bf04 100644
--- a/include/grpcpp/impl/codegen/core_codegen.h
+++ b/include/grpcpp/impl/codegen/core_codegen.h
@@ -73,6 +73,7 @@ class CoreCodegen final : public CoreCodegenInterface {
grpc_byte_buffer* grpc_byte_buffer_copy(grpc_byte_buffer* bb) override;
void grpc_byte_buffer_destroy(grpc_byte_buffer* bb) override;
+ size_t grpc_byte_buffer_length(grpc_byte_buffer* bb) override;
int grpc_byte_buffer_reader_init(grpc_byte_buffer_reader* reader,
grpc_byte_buffer* buffer) override;
@@ -86,6 +87,8 @@ class CoreCodegen final : public CoreCodegenInterface {
grpc_slice grpc_slice_new_with_user_data(void* p, size_t len,
void (*destroy)(void*),
void* user_data) override;
+ grpc_slice grpc_slice_new_with_len(void* p, size_t len,
+ void (*destroy)(void*, size_t)) override;
grpc_slice grpc_empty_slice() override;
grpc_slice grpc_slice_malloc(size_t length) override;
void grpc_slice_unref(grpc_slice slice) override;
diff --git a/include/grpcpp/impl/codegen/core_codegen_interface.h b/include/grpcpp/impl/codegen/core_codegen_interface.h
index d72f579c8e..1167a188a2 100644
--- a/include/grpcpp/impl/codegen/core_codegen_interface.h
+++ b/include/grpcpp/impl/codegen/core_codegen_interface.h
@@ -81,6 +81,8 @@ class CoreCodegenInterface {
virtual grpc_byte_buffer* grpc_byte_buffer_copy(grpc_byte_buffer* bb) = 0;
virtual void grpc_byte_buffer_destroy(grpc_byte_buffer* bb) = 0;
+ virtual size_t grpc_byte_buffer_length(grpc_byte_buffer* bb)
+ GRPC_MUST_USE_RESULT = 0;
virtual int grpc_byte_buffer_reader_init(grpc_byte_buffer_reader* reader,
grpc_byte_buffer* buffer)
@@ -95,6 +97,9 @@ class CoreCodegenInterface {
virtual grpc_slice grpc_slice_new_with_user_data(void* p, size_t len,
void (*destroy)(void*),
void* user_data) = 0;
+ virtual grpc_slice grpc_slice_new_with_len(void* p, size_t len,
+ void (*destroy)(void*,
+ size_t)) = 0;
virtual grpc_call_error grpc_call_cancel_with_status(grpc_call* call,
grpc_status_code status,
const char* description,
diff --git a/include/grpcpp/impl/codegen/proto_buffer_reader.h b/include/grpcpp/impl/codegen/proto_buffer_reader.h
new file mode 100644
index 0000000000..149e05c74f
--- /dev/null
+++ b/include/grpcpp/impl/codegen/proto_buffer_reader.h
@@ -0,0 +1,151 @@
+/*
+ *
+ * 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 GRPCPP_IMPL_CODEGEN_PROTO_BUFFER_READER_H
+#define GRPCPP_IMPL_CODEGEN_PROTO_BUFFER_READER_H
+
+#include <type_traits>
+
+#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/serialization_traits.h>
+#include <grpcpp/impl/codegen/status.h>
+
+/// This header provides an object that reads bytes directly from a
+/// grpc::ByteBuffer, via the ZeroCopyInputStream interface
+
+namespace grpc {
+
+extern CoreCodegenInterface* g_core_codegen_interface;
+
+/// This is a specialization of the protobuf class ZeroCopyInputStream
+/// The principle is to get one chunk of data at a time from the proto layer,
+/// with options to backup (re-see some bytes) or skip (forward past some bytes)
+///
+/// Read more about ZeroCopyInputStream interface here:
+/// https://developers.google.com/protocol-buffers/docs/reference/cpp/google.protobuf.io.zero_copy_stream#ZeroCopyInputStream
+class GrpcProtoBufferReader : public ::grpc::protobuf::io::ZeroCopyInputStream {
+ public:
+ /// Constructs buffer reader from \a buffer. Will set \a status() to non ok
+ /// if \a buffer is invalid (the internal buffer has not been initialized).
+ explicit GrpcProtoBufferReader(ByteBuffer* buffer)
+ : byte_count_(0), backup_count_(0), status_() {
+ /// Implemented through a grpc_byte_buffer_reader which iterates
+ /// over the slices that make up a byte buffer
+ if (!buffer->Valid() ||
+ !g_core_codegen_interface->grpc_byte_buffer_reader_init(
+ &reader_, buffer->c_buffer())) {
+ status_ = Status(StatusCode::INTERNAL,
+ "Couldn't initialize byte buffer reader");
+ }
+ }
+
+ ~GrpcProtoBufferReader() {
+ if (status_.ok()) {
+ g_core_codegen_interface->grpc_byte_buffer_reader_destroy(&reader_);
+ }
+ }
+
+ /// Give the proto library a chunk of data from the stream. The caller
+ /// may safely read from data[0, size - 1].
+ bool Next(const void** data, int* size) override {
+ if (!status_.ok()) {
+ return false;
+ }
+ /// If we have backed up previously, we need to return the backed-up slice
+ 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;
+ }
+ /// Otherwise get the next slice from the byte buffer reader
+ 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;
+ }
+
+ /// Returns the status of the buffer reader.
+ Status status() const { return status_; }
+
+ /// The proto library calls this to indicate that we should back up \a count
+ /// bytes that have already been returned by the last call of Next.
+ /// So do the backup and have that ready for a later Next.
+ void BackUp(int count) override {
+ GPR_CODEGEN_ASSERT(count <= static_cast<int>(GRPC_SLICE_LENGTH(slice_)));
+ backup_count_ = count;
+ }
+
+ /// The proto library calls this to skip over \a count bytes. Implement this
+ /// using Next and BackUp combined.
+ 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;
+ }
+
+ /// Returns the total number of bytes read since this object was created.
+ grpc::protobuf::int64 ByteCount() const override {
+ return byte_count_ - backup_count_;
+ }
+
+ // These protected members are needed to support internal optimizations.
+ // they expose internal bits of grpc core that are NOT stable. If you have
+ // a use case needs to use one of these functions, please send an email to
+ // https://groups.google.com/forum/#!forum/grpc-io.
+ protected:
+ void set_byte_count(int64_t byte_count) { byte_count_ = byte_count; }
+ int64_t backup_count() { return backup_count_; }
+ void set_backup_count(int64_t backup_count) { backup_count_ = backup_count; }
+ grpc_byte_buffer_reader* reader() { return &reader_; }
+ grpc_slice* slice() { return &slice_; }
+
+ private:
+ int64_t byte_count_; ///< total bytes read since object creation
+ int64_t backup_count_; ///< how far backed up in the stream we are
+ grpc_byte_buffer_reader reader_; ///< internal object to read \a grpc_slice
+ ///< from the \a grpc_byte_buffer
+ grpc_slice slice_; ///< current slice passed back to the caller
+ Status status_; ///< status of the entire object
+};
+
+} // namespace grpc
+
+#endif // GRPCPP_IMPL_CODEGEN_PROTO_BUFFER_READER_H
diff --git a/include/grpcpp/impl/codegen/proto_buffer_writer.h b/include/grpcpp/impl/codegen/proto_buffer_writer.h
new file mode 100644
index 0000000000..faf99800cd
--- /dev/null
+++ b/include/grpcpp/impl/codegen/proto_buffer_writer.h
@@ -0,0 +1,168 @@
+/*
+ *
+ * Copyright 2018 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 GRPCPP_IMPL_CODEGEN_PROTO_BUFFER_WRITER_H
+#define GRPCPP_IMPL_CODEGEN_PROTO_BUFFER_WRITER_H
+
+#include <type_traits>
+
+#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/serialization_traits.h>
+#include <grpcpp/impl/codegen/status.h>
+
+/// This header provides an object that writes bytes directly into a
+/// grpc::ByteBuffer, via the ZeroCopyOutputStream interface
+
+namespace grpc {
+
+extern CoreCodegenInterface* g_core_codegen_interface;
+
+// Forward declaration for testing use only
+namespace internal {
+class GrpcProtoBufferWriterPeer;
+} // namespace internal
+
+const int kGrpcProtoBufferWriterMaxBufferLength = 1024 * 1024;
+
+/// This is a specialization of the protobuf class ZeroCopyOutputStream.
+/// The principle is to give the proto layer one buffer of bytes at a time
+/// that it can use to serialize the next portion of the message, with the
+/// option to "backup" if more buffer is given than required at the last buffer.
+///
+/// Read more about ZeroCopyOutputStream interface here:
+/// https://developers.google.com/protocol-buffers/docs/reference/cpp/google.protobuf.io.zero_copy_stream#ZeroCopyOutputStream
+class GrpcProtoBufferWriter
+ : public ::grpc::protobuf::io::ZeroCopyOutputStream {
+ public:
+ /// Constructor for this derived class
+ ///
+ /// \param[out] byte_buffer A pointer to the grpc::ByteBuffer created
+ /// \param block_size How big are the chunks to allocate at a time
+ /// \param total_size How many total bytes are required for this proto
+ GrpcProtoBufferWriter(ByteBuffer* byte_buffer, int block_size, int total_size)
+ : block_size_(block_size),
+ total_size_(total_size),
+ byte_count_(0),
+ have_backup_(false) {
+ GPR_CODEGEN_ASSERT(!byte_buffer->Valid());
+ /// Create an empty raw byte buffer and look at its underlying slice buffer
+ grpc_byte_buffer* bp =
+ g_core_codegen_interface->grpc_raw_byte_buffer_create(NULL, 0);
+ byte_buffer->set_buffer(bp);
+ slice_buffer_ = &bp->data.raw.slice_buffer;
+ }
+
+ ~GrpcProtoBufferWriter() {
+ if (have_backup_) {
+ g_core_codegen_interface->grpc_slice_unref(backup_slice_);
+ }
+ }
+
+ /// Give the proto library the next buffer of bytes and its size. It is
+ /// safe for the caller to write from data[0, size - 1].
+ bool Next(void** data, int* size) override {
+ // Protobuf should not ask for more memory than total_size_.
+ GPR_CODEGEN_ASSERT(byte_count_ < total_size_);
+ // 1. Use the remaining backup slice if we have one
+ // 2. Otherwise allocate a slice, up to the remaining length needed
+ // or our maximum allocation size
+ // 3. Provide the slice start and size available
+ // 4. Add the slice being returned to the slice buffer
+ size_t remain = total_size_ - byte_count_;
+ if (have_backup_) {
+ /// If we have a backup slice, we should use it first
+ 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;
+ }
+
+ /// Backup by \a count bytes because Next returned more bytes than needed
+ /// (only used in the last buffer). \a count must be less than or equal too
+ /// the last buffer returned from next.
+ void BackUp(int count) override {
+ /// 1. Remove the partially-used last slice from the slice buffer
+ /// 2. Split it into the needed (if any) and unneeded part
+ /// 3. Add the needed part back to the slice buffer
+ /// 4. Mark that we still have the remaining part (for later use/unref)
+ GPR_CODEGEN_ASSERT(count <= static_cast<int>(GRPC_SLICE_LENGTH(slice_)));
+ 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 address held by
+ // slice_buffer_.
+ have_backup_ = backup_slice_.refcount != NULL;
+ byte_count_ -= count;
+ }
+
+ /// Returns the total number of bytes written since this object was created.
+ grpc::protobuf::int64 ByteCount() const override { return byte_count_; }
+
+ // These protected members are needed to support internal optimizations.
+ // they expose internal bits of grpc core that are NOT stable. If you have
+ // a use case needs to use one of these functions, please send an email to
+ // https://groups.google.com/forum/#!forum/grpc-io.
+ protected:
+ grpc_slice_buffer* slice_buffer() { return slice_buffer_; }
+ void set_byte_count(int64_t byte_count) { byte_count_ = byte_count; }
+
+ private:
+ // friend for testing purposes only
+ friend class internal::GrpcProtoBufferWriterPeer;
+ const int block_size_; ///< size to alloc for each new \a grpc_slice needed
+ const int total_size_; ///< byte size of proto being serialized
+ int64_t byte_count_; ///< bytes written since this object was created
+ grpc_slice_buffer*
+ slice_buffer_; ///< internal buffer of slices holding the serialized data
+ bool have_backup_; ///< if we are holding a backup slice or not
+ grpc_slice backup_slice_; ///< holds space we can still write to, if the
+ ///< caller has called BackUp
+ grpc_slice slice_; ///< current slice passed back to the caller
+};
+
+} // namespace grpc
+
+#endif // GRPCPP_IMPL_CODEGEN_PROTO_BUFFER_WRITER_H
diff --git a/include/grpcpp/impl/codegen/proto_utils.h b/include/grpcpp/impl/codegen/proto_utils.h
index 81438ee1d5..ec1dccaa74 100644
--- a/include/grpcpp/impl/codegen/proto_utils.h
+++ b/include/grpcpp/impl/codegen/proto_utils.h
@@ -24,203 +24,62 @@
#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, kGrpcProtoBufferWriterMaxBufferLength,
+ 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 +92,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 +106,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<GrpcProtoBufferWriter, 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<GrpcProtoBufferReader, T>(buffer, msg);
}
};
#endif
diff --git a/include/grpcpp/impl/codegen/slice.h b/include/grpcpp/impl/codegen/slice.h
index fcccd4b68e..8966559dc8 100644
--- a/include/grpcpp/impl/codegen/slice.h
+++ b/include/grpcpp/impl/codegen/slice.h
@@ -35,34 +35,43 @@ namespace grpc {
class Slice final {
public:
/// Construct an empty slice.
- Slice();
+ Slice() : slice_(g_core_codegen_interface->grpc_empty_slice()) {}
/// Destructor - drops one reference.
- ~Slice();
+ ~Slice() { g_core_codegen_interface->grpc_slice_unref(slice_); }
enum AddRef { ADD_REF };
/// Construct a slice from \a slice, adding a reference.
- Slice(grpc_slice slice, AddRef);
+ Slice(grpc_slice slice, AddRef)
+ : slice_(g_core_codegen_interface->grpc_slice_ref(slice)) {}
enum StealRef { STEAL_REF };
/// Construct a slice from \a slice, stealing a reference.
- Slice(grpc_slice slice, StealRef);
+ Slice(grpc_slice slice, StealRef) : slice_(slice) {}
/// Allocate a slice of specified size
- Slice(size_t len);
+ Slice(size_t len)
+ : slice_(g_core_codegen_interface->grpc_slice_malloc(len)) {}
/// Construct a slice from a copied buffer
- Slice(const void* buf, size_t len);
+ Slice(const void* buf, size_t len)
+ : slice_(g_core_codegen_interface->grpc_slice_from_copied_buffer(
+ reinterpret_cast<const char*>(buf), len)) {}
/// Construct a slice from a copied string
- Slice(const grpc::string& str);
+ Slice(const grpc::string& str)
+ : slice_(g_core_codegen_interface->grpc_slice_from_copied_buffer(
+ str.c_str(), str.length())) {}
enum StaticSlice { STATIC_SLICE };
/// Construct a slice from a static buffer
- Slice(const void* buf, size_t len, StaticSlice);
+ Slice(const void* buf, size_t len, StaticSlice)
+ : slice_(g_core_codegen_interface->grpc_slice_from_static_buffer(
+ reinterpret_cast<const char*>(buf), len)) {}
/// Copy constructor, adds a reference.
- Slice(const Slice& other);
+ Slice(const Slice& other)
+ : slice_(g_core_codegen_interface->grpc_slice_ref(other.slice_)) {}
/// Assignment, reference count is unchanged.
Slice& operator=(Slice other) {
@@ -75,14 +84,18 @@ class Slice final {
/// user data pointer passed in at destruction. Can be the same as buf or
/// different (e.g., if data is part of a larger structure that must be
/// destroyed when the data is no longer needed)
- Slice(void* buf, size_t len, void (*destroy)(void*), void* user_data);
+ Slice(void* buf, size_t len, void (*destroy)(void*), void* user_data)
+ : slice_(g_core_codegen_interface->grpc_slice_new_with_user_data(
+ buf, len, destroy, user_data)) {}
/// Specialization of above for common case where buf == user_data
Slice(void* buf, size_t len, void (*destroy)(void*))
: Slice(buf, len, destroy, buf) {}
/// Similar to the above but has a destroy that also takes slice length
- Slice(void* buf, size_t len, void (*destroy)(void*, size_t));
+ Slice(void* buf, size_t len, void (*destroy)(void*, size_t))
+ : slice_(g_core_codegen_interface->grpc_slice_new_with_len(buf, len,
+ destroy)) {}
/// Byte size.
size_t size() const { return GRPC_SLICE_LENGTH(slice_); }
@@ -94,7 +107,9 @@ class Slice final {
const uint8_t* end() const { return GRPC_SLICE_END_PTR(slice_); }
/// Raw C slice. Caller needs to call grpc_slice_unref when done.
- grpc_slice c_slice() const;
+ grpc_slice c_slice() const {
+ return g_core_codegen_interface->grpc_slice_ref(slice_);
+ }
private:
friend class ByteBuffer;