// // // 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_SUPPORT_METHOD_HANDLER_H #define GRPCPP_SUPPORT_METHOD_HANDLER_H #include #include #include #include #include namespace grpc { namespace internal { // Invoke the method handler, fill in the status, and // return whether or not we finished safely (without an exception). // Note that exception handling is 0-cost in most compiler/library // implementations (except when an exception is actually thrown), // so this process doesn't require additional overhead in the common case. // Additionally, we don't need to return if we caught an exception or not; // the handling is the same in either case. template ::grpc::Status CatchingFunctionHandler(Callable&& handler) { #if GRPC_ALLOW_EXCEPTIONS try { return handler(); } catch (...) { return grpc::Status(grpc::StatusCode::UNKNOWN, "Unexpected error in RPC handling"); } #else // GRPC_ALLOW_EXCEPTIONS return handler(); #endif // GRPC_ALLOW_EXCEPTIONS } /// A helper function with reduced templating to do the common work needed to /// actually send the server response. Uses non-const parameter for Status since /// this should only ever be called from the end of the RunHandler method. template void UnaryRunHandlerHelper(const MethodHandler::HandlerParameter& param, ResponseType* rsp, grpc::Status& status) { GPR_ASSERT(!param.server_context->sent_initial_metadata_); grpc::internal::CallOpSet ops; ops.SendInitialMetadata(¶m.server_context->initial_metadata_, param.server_context->initial_metadata_flags()); if (param.server_context->compression_level_set()) { ops.set_compression_level(param.server_context->compression_level()); } if (status.ok()) { status = ops.SendMessagePtr(rsp); } ops.ServerSendStatus(¶m.server_context->trailing_metadata_, status); param.call->PerformOps(&ops); param.call->cq()->Pluck(&ops); } /// A helper function with reduced templating to do deserializing. template void* UnaryDeserializeHelper(grpc_byte_buffer* req, grpc::Status* status, RequestType* request) { grpc::ByteBuffer buf; buf.set_buffer(req); *status = grpc::SerializationTraits::Deserialize( &buf, static_cast(request)); buf.Release(); if (status->ok()) { return request; } request->~RequestType(); return nullptr; } /// A wrapper class of an application provided rpc method handler. template class RpcMethodHandler : public grpc::internal::MethodHandler { public: RpcMethodHandler( std::function func, ServiceType* service) : func_(func), service_(service) {} void RunHandler(const HandlerParameter& param) final { ResponseType rsp; grpc::Status status = param.status; if (status.ok()) { status = CatchingFunctionHandler([this, ¶m, &rsp] { return func_(service_, static_cast(param.server_context), static_cast(param.request), &rsp); }); static_cast(param.request)->~RequestType(); } UnaryRunHandlerHelper(param, static_cast(&rsp), status); } void* Deserialize(grpc_call* call, grpc_byte_buffer* req, grpc::Status* status, void** /*handler_data*/) final { auto* request = new (grpc_call_arena_alloc(call, sizeof(RequestType))) RequestType; return UnaryDeserializeHelper(req, status, static_cast(request)); } private: /// Application provided rpc handler function. std::function func_; // The class the above handler function lives in. ServiceType* service_; }; /// A wrapper class of an application provided client streaming handler. template class ClientStreamingHandler : public grpc::internal::MethodHandler { public: ClientStreamingHandler( std::function*, ResponseType*)> func, ServiceType* service) : func_(func), service_(service) {} void RunHandler(const HandlerParameter& param) final { ServerReader reader( param.call, static_cast(param.server_context)); ResponseType rsp; grpc::Status status = CatchingFunctionHandler([this, ¶m, &reader, &rsp] { return func_(service_, static_cast(param.server_context), &reader, &rsp); }); grpc::internal::CallOpSet ops; if (!param.server_context->sent_initial_metadata_) { ops.SendInitialMetadata(¶m.server_context->initial_metadata_, param.server_context->initial_metadata_flags()); if (param.server_context->compression_level_set()) { ops.set_compression_level(param.server_context->compression_level()); } } if (status.ok()) { status = ops.SendMessagePtr(&rsp); } ops.ServerSendStatus(¶m.server_context->trailing_metadata_, status); param.call->PerformOps(&ops); param.call->cq()->Pluck(&ops); } private: std::function*, ResponseType*)> func_; ServiceType* service_; }; /// A wrapper class of an application provided server streaming handler. template class ServerStreamingHandler : public grpc::internal::MethodHandler { public: ServerStreamingHandler(std::function*)> func, ServiceType* service) : func_(func), service_(service) {} void RunHandler(const HandlerParameter& param) final { grpc::Status status = param.status; if (status.ok()) { ServerWriter writer( param.call, static_cast(param.server_context)); status = CatchingFunctionHandler([this, ¶m, &writer] { return func_(service_, static_cast(param.server_context), static_cast(param.request), &writer); }); static_cast(param.request)->~RequestType(); } grpc::internal::CallOpSet ops; if (!param.server_context->sent_initial_metadata_) { ops.SendInitialMetadata(¶m.server_context->initial_metadata_, param.server_context->initial_metadata_flags()); if (param.server_context->compression_level_set()) { ops.set_compression_level(param.server_context->compression_level()); } } 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_); } param.call->cq()->Pluck(&ops); } void* Deserialize(grpc_call* call, grpc_byte_buffer* req, grpc::Status* status, void** /*handler_data*/) final { grpc::ByteBuffer buf; buf.set_buffer(req); auto* request = new (grpc_call_arena_alloc(call, sizeof(RequestType))) RequestType(); *status = grpc::SerializationTraits::Deserialize(&buf, request); buf.Release(); if (status->ok()) { return request; } request->~RequestType(); return nullptr; } private: std::function*)> func_; ServiceType* service_; }; /// A wrapper class of an application provided bidi-streaming handler. /// This also applies to server-streamed implementation of a unary method /// with the additional requirement that such methods must have done a /// write for status to be ok /// Since this is used by more than 1 class, the service is not passed in. /// Instead, it is expected to be an implicitly-captured argument of func /// (through bind or something along those lines) template class TemplatedBidiStreamingHandler : public grpc::internal::MethodHandler { public: explicit TemplatedBidiStreamingHandler( std::function func) : func_(func), write_needed_(WriteNeeded) {} void RunHandler(const HandlerParameter& param) final { Streamer stream(param.call, static_cast(param.server_context)); grpc::Status status = CatchingFunctionHandler([this, ¶m, &stream] { return func_(static_cast(param.server_context), &stream); }); grpc::internal::CallOpSet ops; if (!param.server_context->sent_initial_metadata_) { ops.SendInitialMetadata(¶m.server_context->initial_metadata_, param.server_context->initial_metadata_flags()); if (param.server_context->compression_level_set()) { ops.set_compression_level(param.server_context->compression_level()); } if (write_needed_ && status.ok()) { // If we needed a write but never did one, we need to mark the // status as a fail status = grpc::Status(grpc::StatusCode::INTERNAL, "Service did not provide response message"); } } 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_); } param.call->cq()->Pluck(&ops); } private: std::function func_; const bool write_needed_; }; template class BidiStreamingHandler : public TemplatedBidiStreamingHandler< ServerReaderWriter, false> { public: BidiStreamingHandler(std::function*)> func, ServiceType* service) // TODO(vjpai): When gRPC supports C++14, move-capture func in the below : TemplatedBidiStreamingHandler< ServerReaderWriter, false>( [func, service]( grpc::ServerContext* ctx, ServerReaderWriter* streamer) { return func(service, ctx, streamer); }) {} }; template class StreamedUnaryHandler : public TemplatedBidiStreamingHandler< ServerUnaryStreamer, true> { public: explicit StreamedUnaryHandler( std::function< grpc::Status(grpc::ServerContext*, ServerUnaryStreamer*)> func) : TemplatedBidiStreamingHandler< ServerUnaryStreamer, true>( std::move(func)) {} }; template class SplitServerStreamingHandler : public TemplatedBidiStreamingHandler< ServerSplitStreamer, false> { public: explicit SplitServerStreamingHandler( std::function< grpc::Status(grpc::ServerContext*, ServerSplitStreamer*)> func) : TemplatedBidiStreamingHandler< ServerSplitStreamer, false>( std::move(func)) {} }; /// General method handler class for errors that prevent real method use /// e.g., handle unknown method by returning UNIMPLEMENTED error. template class ErrorMethodHandler : public grpc::internal::MethodHandler { public: explicit ErrorMethodHandler(const std::string& message) : message_(message) {} template static void FillOps(grpc::ServerContextBase* context, const std::string& message, T* ops) { grpc::Status status(code, message); if (!context->sent_initial_metadata_) { ops->SendInitialMetadata(&context->initial_metadata_, context->initial_metadata_flags()); if (context->compression_level_set()) { ops->set_compression_level(context->compression_level()); } context->sent_initial_metadata_ = true; } ops->ServerSendStatus(&context->trailing_metadata_, status); } void RunHandler(const HandlerParameter& param) final { grpc::internal::CallOpSet ops; FillOps(param.server_context, message_, &ops); param.call->PerformOps(&ops); param.call->cq()->Pluck(&ops); } void* Deserialize(grpc_call* /*call*/, grpc_byte_buffer* req, grpc::Status* /*status*/, void** /*handler_data*/) final { // We have to destroy any request payload if (req != nullptr) { grpc_byte_buffer_destroy(req); } return nullptr; } private: const std::string message_; }; typedef ErrorMethodHandler UnknownMethodHandler; typedef ErrorMethodHandler ResourceExhaustedHandler; } // namespace internal } // namespace grpc #endif // GRPCPP_SUPPORT_METHOD_HANDLER_H