diff options
Diffstat (limited to 'include/grpcpp')
-rw-r--r-- | include/grpcpp/impl/codegen/call.h | 6 | ||||
-rw-r--r-- | include/grpcpp/impl/codegen/method_handler_impl.h | 52 | ||||
-rw-r--r-- | include/grpcpp/impl/codegen/rpc_service_method.h | 16 | ||||
-rw-r--r-- | include/grpcpp/impl/codegen/server_interface.h | 7 | ||||
-rw-r--r-- | include/grpcpp/server.h | 4 |
5 files changed, 51 insertions, 34 deletions
diff --git a/include/grpcpp/impl/codegen/call.h b/include/grpcpp/impl/codegen/call.h index ccadd2af43..d591144798 100644 --- a/include/grpcpp/impl/codegen/call.h +++ b/include/grpcpp/impl/codegen/call.h @@ -997,7 +997,6 @@ class InterceptorBatchMethodsImpl server_rpc_info->interceptors_.size() == 0) { return true; } - GPR_ASSERT(false); RunServerInterceptors(); return false; } @@ -1128,7 +1127,6 @@ class InterceptorBatchMethodsImpl Status send_status_; std::multimap<grpc::string, grpc::string>* send_trailing_metadata_ = nullptr; - size_t* send_trailing_metadata_count_ = nullptr; void* recv_message_ = nullptr; @@ -1137,10 +1135,6 @@ class InterceptorBatchMethodsImpl Status* recv_status_ = nullptr; internal::MetadataMap* recv_trailing_metadata_ = nullptr; - - // void (*hijacking_state_setter_)(); - // void (*continue_after_interception_)(); - // void (*continue_after_reverse_interception_)(); }; /// Primary implementation of CallOpSetInterface. diff --git a/include/grpcpp/impl/codegen/method_handler_impl.h b/include/grpcpp/impl/codegen/method_handler_impl.h index 93f4af03ee..4f02e3e39b 100644 --- a/include/grpcpp/impl/codegen/method_handler_impl.h +++ b/include/grpcpp/impl/codegen/method_handler_impl.h @@ -60,10 +60,13 @@ class RpcMethodHandler : public MethodHandler { void RunHandler(const HandlerParameter& param) final { ResponseType rsp; - if (status_.ok()) { - status_ = CatchingFunctionHandler([this, ¶m, &rsp] { - return func_(service_, param.server_context, &this->req_, &rsp); + Status status = param.status; + if (status.ok()) { + status = CatchingFunctionHandler([this, ¶m, &rsp] { + return func_(service_, param.server_context, + static_cast<RequestType*>(param.request), &rsp); }); + delete static_cast<RequestType*>(param.request); } GPR_CODEGEN_ASSERT(!param.server_context->sent_initial_metadata_); @@ -75,22 +78,24 @@ class RpcMethodHandler : public MethodHandler { if (param.server_context->compression_level_set()) { ops.set_compression_level(param.server_context->compression_level()); } - if (status_.ok()) { - status_ = ops.SendMessage(rsp); + if (status.ok()) { + status = ops.SendMessage(rsp); } - ops.ServerSendStatus(¶m.server_context->trailing_metadata_, status_); + ops.ServerSendStatus(¶m.server_context->trailing_metadata_, status); param.call->PerformOps(&ops); param.call->cq()->Pluck(&ops); } - void* Deserialize(grpc_byte_buffer* req) final { + void* Deserialize(grpc_byte_buffer* req, Status* status) final { ByteBuffer buf; buf.set_buffer(req); - status_ = SerializationTraits<RequestType>::Deserialize(&buf, &req_); + auto* request = new RequestType(); + *status = SerializationTraits<RequestType>::Deserialize(&buf, request); buf.Release(); - if (status_.ok()) { - return &req_; + if (status->ok()) { + return request; } + delete request; return nullptr; } @@ -101,8 +106,6 @@ class RpcMethodHandler : public MethodHandler { func_; // The class the above handler function lives in. ServiceType* service_; - RequestType req_; - Status status_; }; /// A wrapper class of an application provided client streaming handler. @@ -160,11 +163,14 @@ class ServerStreamingHandler : public MethodHandler { : func_(func), service_(service) {} void RunHandler(const HandlerParameter& param) final { - if (status_.ok()) { + Status status = param.status; + if (status.ok()) { ServerWriter<ResponseType> writer(param.call, param.server_context); - status_ = CatchingFunctionHandler([this, ¶m, &writer] { - return func_(service_, param.server_context, &this->req_, &writer); + status = CatchingFunctionHandler([this, ¶m, &writer] { + return func_(service_, param.server_context, + static_cast<RequestType*>(param.request), &writer); }); + delete static_cast<RequestType*>(param.request); } CallOpSet<CallOpSendInitialMetadata, CallOpServerSendStatus> ops; @@ -175,7 +181,7 @@ class ServerStreamingHandler : public MethodHandler { ops.set_compression_level(param.server_context->compression_level()); } } - ops.ServerSendStatus(¶m.server_context->trailing_metadata_, status_); + ops.ServerSendStatus(¶m.server_context->trailing_metadata_, status); param.call->PerformOps(&ops); if (param.server_context->has_pending_ops_) { param.call->cq()->Pluck(¶m.server_context->pending_ops_); @@ -183,14 +189,16 @@ class ServerStreamingHandler : public MethodHandler { param.call->cq()->Pluck(&ops); } - void* Deserialize(grpc_byte_buffer* req) final { + void* Deserialize(grpc_byte_buffer* req, Status* status) final { ByteBuffer buf; buf.set_buffer(req); - status_ = SerializationTraits<RequestType>::Deserialize(&buf, &req_); + auto* request = new RequestType(); + *status = SerializationTraits<RequestType>::Deserialize(&buf, request); buf.Release(); - if (status_.ok()) { - return &req_; + if (status->ok()) { + return request; } + delete request; return nullptr; } @@ -199,8 +207,6 @@ class ServerStreamingHandler : public MethodHandler { ServerWriter<ResponseType>*)> func_; ServiceType* service_; - RequestType req_; - Status status_; }; /// A wrapper class of an application provided bidi-streaming handler. @@ -317,7 +323,7 @@ class ErrorMethodHandler : public MethodHandler { param.call->cq()->Pluck(&ops); } - void* Deserialize(grpc_byte_buffer* req) final { + void* Deserialize(grpc_byte_buffer* req, Status* status) final { // We have to destroy any request payload if (req != nullptr) { g_core_codegen_interface->grpc_byte_buffer_destroy(req); diff --git a/include/grpcpp/impl/codegen/rpc_service_method.h b/include/grpcpp/impl/codegen/rpc_service_method.h index 04607efd7d..44da2bd768 100644 --- a/include/grpcpp/impl/codegen/rpc_service_method.h +++ b/include/grpcpp/impl/codegen/rpc_service_method.h @@ -40,17 +40,23 @@ class MethodHandler { public: virtual ~MethodHandler() {} struct HandlerParameter { - HandlerParameter(Call* c, ServerContext* context) - : call(c), server_context(context) {} + HandlerParameter(Call* c, ServerContext* context, void* req, + Status req_status) + : call(c), server_context(context), request(req), status(req_status) {} ~HandlerParameter() {} Call* call; ServerContext* server_context; + void* request; + Status status; }; virtual void RunHandler(const HandlerParameter& param) = 0; - /* Returns pointer to the deserialized request. Ownership is retained by the - handler. Returns nullptr if deserialization failed */ - virtual void* Deserialize(grpc_byte_buffer* req) { + /* Returns a pointer to the deserialized request. \a status reflects the + result of deserialization. This pointer and the status should be filled in + a HandlerParameter and passed to RunHandler. It is illegal to access the + pointer after calling RunHandler. Ownership of the deserialized request is + retained by the handler. Returns nullptr if deserialization failed. */ + virtual void* Deserialize(grpc_byte_buffer* req, Status* status) { GPR_CODEGEN_ASSERT(req == nullptr); return nullptr; } diff --git a/include/grpcpp/impl/codegen/server_interface.h b/include/grpcpp/impl/codegen/server_interface.h index 237991cde6..9310ccb39f 100644 --- a/include/grpcpp/impl/codegen/server_interface.h +++ b/include/grpcpp/impl/codegen/server_interface.h @@ -21,6 +21,7 @@ #include <grpc/impl/codegen/grpc_types.h> #include <grpcpp/impl/codegen/byte_buffer.h> +#include <grpcpp/impl/codegen/call.h> #include <grpcpp/impl/codegen/call_hook.h> #include <grpcpp/impl/codegen/completion_queue_tag.h> #include <grpcpp/impl/codegen/core_codegen_interface.h> @@ -162,6 +163,7 @@ class ServerInterface : public internal::CallHook { void* const tag_; const bool delete_on_finalize_; grpc_call* call_; + internal::InterceptorBatchMethodsImpl interceptor_methods; }; class RegisteredAsyncRequest : public BaseAsyncRequest { @@ -295,6 +297,11 @@ class ServerInterface : public internal::CallHook { new GenericAsyncRequest(this, context, stream, call_cq, notification_cq, tag, true); } + +private: + virtual const std::vector<std::unique_ptr<experimental::ServerInterceptorFactoryInterface>>* interceptor_creators() { + return nullptr; + } }; } // namespace grpc diff --git a/include/grpcpp/server.h b/include/grpcpp/server.h index 27d1ec0cfa..a593b60550 100644 --- a/include/grpcpp/server.h +++ b/include/grpcpp/server.h @@ -191,6 +191,10 @@ class Server : public ServerInterface, private GrpcLibraryCodegen { grpc_server* server() override { return server_; }; private: + const std::vector<std::unique_ptr<experimental::ServerInterceptorFactoryInterface>>* interceptor_creators() override { + return &interceptor_creators_; + } + friend class AsyncGenericService; friend class ServerBuilder; friend class ServerInitializer; |