aboutsummaryrefslogtreecommitdiffhomepage
path: root/include/grpcpp/impl/codegen/call_op_set.h
diff options
context:
space:
mode:
Diffstat (limited to 'include/grpcpp/impl/codegen/call_op_set.h')
-rw-r--r--include/grpcpp/impl/codegen/call_op_set.h108
1 files changed, 88 insertions, 20 deletions
diff --git a/include/grpcpp/impl/codegen/call_op_set.h b/include/grpcpp/impl/codegen/call_op_set.h
index b2100c68b7..c0de5ed602 100644
--- a/include/grpcpp/impl/codegen/call_op_set.h
+++ b/include/grpcpp/impl/codegen/call_op_set.h
@@ -303,9 +303,29 @@ class CallOpSendMessage {
template <class M>
Status SendMessage(const M& message) GRPC_MUST_USE_RESULT;
+ /// Send \a message using \a options for the write. The \a options are cleared
+ /// after use. This form of SendMessage allows gRPC to reference \a message
+ /// beyond the lifetime of SendMessage.
+ template <class M>
+ Status SendMessagePtr(const M* message,
+ WriteOptions options) GRPC_MUST_USE_RESULT;
+
+ /// This form of SendMessage allows gRPC to reference \a message beyond the
+ /// lifetime of SendMessage.
+ template <class M>
+ Status SendMessagePtr(const M* message) GRPC_MUST_USE_RESULT;
+
protected:
void AddOp(grpc_op* ops, size_t* nops) {
- if (!send_buf_.Valid() || hijacked_) return;
+ if (msg_ == nullptr && !send_buf_.Valid()) return;
+ if (hijacked_) {
+ serializer_ = nullptr;
+ return;
+ }
+ if (msg_ != nullptr) {
+ GPR_CODEGEN_ASSERT(serializer_(msg_).ok());
+ }
+ serializer_ = nullptr;
grpc_op* op = &ops[(*nops)++];
op->op = GRPC_OP_SEND_MESSAGE;
op->flags = write_options_.flags();
@@ -314,21 +334,38 @@ class CallOpSendMessage {
// Flags are per-message: clear them after use.
write_options_.Clear();
}
- void FinishOp(bool* status) { send_buf_.Clear(); }
+ void FinishOp(bool* status) {
+ if (msg_ == nullptr && !send_buf_.Valid()) return;
+ if (hijacked_ && failed_send_) {
+ // Hijacking interceptor failed this Op
+ *status = false;
+ } else if (!*status) {
+ // This Op was passed down to core and the Op failed
+ failed_send_ = true;
+ }
+ }
void SetInterceptionHookPoint(
InterceptorBatchMethodsImpl* interceptor_methods) {
- if (!send_buf_.Valid()) return;
+ if (msg_ == nullptr && !send_buf_.Valid()) return;
interceptor_methods->AddInterceptionHookPoint(
experimental::InterceptionHookPoints::PRE_SEND_MESSAGE);
- interceptor_methods->SetSendMessage(&send_buf_);
+ interceptor_methods->SetSendMessage(&send_buf_, &msg_, &failed_send_,
+ serializer_);
}
void SetFinishInterceptionHookPoint(
InterceptorBatchMethodsImpl* interceptor_methods) {
+ if (msg_ != nullptr || send_buf_.Valid()) {
+ interceptor_methods->AddInterceptionHookPoint(
+ experimental::InterceptionHookPoints::POST_SEND_MESSAGE);
+ }
+ send_buf_.Clear();
+ msg_ = nullptr;
// The contents of the SendMessage value that was previously set
// has had its references stolen by core's operations
- interceptor_methods->SetSendMessage(nullptr);
+ interceptor_methods->SetSendMessage(nullptr, nullptr, &failed_send_,
+ nullptr);
}
void SetHijackingState(InterceptorBatchMethodsImpl* interceptor_methods) {
@@ -336,25 +373,38 @@ class CallOpSendMessage {
}
private:
+ const void* msg_ = nullptr; // The original non-serialized message
bool hijacked_ = false;
+ bool failed_send_ = false;
ByteBuffer send_buf_;
WriteOptions write_options_;
+ std::function<Status(const void*)> serializer_;
};
template <class M>
Status CallOpSendMessage::SendMessage(const M& message, WriteOptions options) {
write_options_ = options;
- bool own_buf;
- // TODO(vjpai): Remove the void below when possible
- // The void in the template parameter below should not be needed
- // (since it should be implicit) but is needed due to an observed
- // difference in behavior between clang and gcc for certain internal users
- Status result = SerializationTraits<M, void>::Serialize(
- message, send_buf_.bbuf_ptr(), &own_buf);
- if (!own_buf) {
- send_buf_.Duplicate();
- }
- return result;
+ serializer_ = [this](const void* message) {
+ bool own_buf;
+ send_buf_.Clear();
+ // TODO(vjpai): Remove the void below when possible
+ // The void in the template parameter below should not be needed
+ // (since it should be implicit) but is needed due to an observed
+ // difference in behavior between clang and gcc for certain internal users
+ Status result = SerializationTraits<M, void>::Serialize(
+ *static_cast<const M*>(message), send_buf_.bbuf_ptr(), &own_buf);
+ if (!own_buf) {
+ send_buf_.Duplicate();
+ }
+ return result;
+ };
+ // Serialize immediately only if we do not have access to the message pointer
+ if (msg_ == nullptr) {
+ Status result = serializer_(&message);
+ serializer_ = nullptr;
+ return result;
+ }
+ return Status();
}
template <class M>
@@ -362,6 +412,19 @@ Status CallOpSendMessage::SendMessage(const M& message) {
return SendMessage(message, WriteOptions());
}
+template <class M>
+Status CallOpSendMessage::SendMessagePtr(const M* message,
+ WriteOptions options) {
+ msg_ = message;
+ return SendMessage(*message, options);
+}
+
+template <class M>
+Status CallOpSendMessage::SendMessagePtr(const M* message) {
+ msg_ = message;
+ return SendMessage(*message, WriteOptions());
+}
+
template <class R>
class CallOpRecvMessage {
public:
@@ -410,14 +473,16 @@ class CallOpRecvMessage {
void SetInterceptionHookPoint(
InterceptorBatchMethodsImpl* interceptor_methods) {
- interceptor_methods->SetRecvMessage(message_);
+ if (message_ == nullptr) return;
+ interceptor_methods->SetRecvMessage(message_, &got_message);
}
void SetFinishInterceptionHookPoint(
InterceptorBatchMethodsImpl* interceptor_methods) {
- if (!got_message) return;
+ if (message_ == nullptr) return;
interceptor_methods->AddInterceptionHookPoint(
experimental::InterceptionHookPoints::POST_RECV_MESSAGE);
+ if (!got_message) interceptor_methods->SetRecvMessage(nullptr, nullptr);
}
void SetHijackingState(InterceptorBatchMethodsImpl* interceptor_methods) {
hijacked_ = true;
@@ -505,20 +570,23 @@ class CallOpGenericRecvMessage {
void SetInterceptionHookPoint(
InterceptorBatchMethodsImpl* interceptor_methods) {
- interceptor_methods->SetRecvMessage(message_);
+ if (!deserialize_) return;
+ interceptor_methods->SetRecvMessage(message_, &got_message);
}
void SetFinishInterceptionHookPoint(
InterceptorBatchMethodsImpl* interceptor_methods) {
- if (!got_message) return;
+ if (!deserialize_) return;
interceptor_methods->AddInterceptionHookPoint(
experimental::InterceptionHookPoints::POST_RECV_MESSAGE);
+ if (!got_message) interceptor_methods->SetRecvMessage(nullptr, nullptr);
}
void SetHijackingState(InterceptorBatchMethodsImpl* interceptor_methods) {
hijacked_ = true;
if (!deserialize_) return;
interceptor_methods->AddInterceptionHookPoint(
experimental::InterceptionHookPoints::PRE_RECV_MESSAGE);
+ got_message = true;
}
private: