| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401 | /* * * 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_METHOD_HANDLER_H#define GRPCPP_IMPL_CODEGEN_METHOD_HANDLER_H#include <grpcpp/impl/codegen/byte_buffer.h>#include <grpcpp/impl/codegen/core_codegen_interface.h>#include <grpcpp/impl/codegen/rpc_service_method.h>#include <grpcpp/impl/codegen/sync_stream.h>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 <class Callable>::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 <class ResponseType>void UnaryRunHandlerHelper(const MethodHandler::HandlerParameter& param,                           ResponseType* rsp, ::grpc::Status& status) {  GPR_CODEGEN_ASSERT(!param.server_context->sent_initial_metadata_);  ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,                              ::grpc::internal::CallOpSendMessage,                              ::grpc::internal::CallOpServerSendStatus>      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 <class RequestType>void* UnaryDeserializeHelper(grpc_call* call, grpc_byte_buffer* req,                             ::grpc::Status* status, RequestType* request) {  ::grpc::ByteBuffer buf;  buf.set_buffer(req);  *status = ::grpc::SerializationTraits<RequestType>::Deserialize(      &buf, static_cast<RequestType*>(request));  buf.Release();  if (status->ok()) {    return request;  }  request->~RequestType();  return nullptr;}/// A wrapper class of an application provided rpc method handler.template <class ServiceType, class RequestType, class ResponseType,          class BaseRequestType = RequestType,          class BaseResponseType = ResponseType>class RpcMethodHandler : public ::grpc::internal::MethodHandler { public:  RpcMethodHandler(      std::function<::grpc::Status(ServiceType*, ::grpc::ServerContext*,                                   const RequestType*, ResponseType*)>          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<::grpc::ServerContext*>(param.server_context),                     static_cast<RequestType*>(param.request), &rsp);      });      static_cast<RequestType*>(param.request)->~RequestType();    }    UnaryRunHandlerHelper(param, static_cast<BaseResponseType*>(&rsp), status);  }  void* Deserialize(grpc_call* call, grpc_byte_buffer* req,                    ::grpc::Status* status, void** /*handler_data*/) final {    auto* request =        new (::grpc::g_core_codegen_interface->grpc_call_arena_alloc(            call, sizeof(RequestType))) RequestType;    return UnaryDeserializeHelper(call, req, status,                                  static_cast<BaseRequestType*>(request));  } private:  /// Application provided rpc handler function.  std::function<::grpc::Status(ServiceType*, ::grpc::ServerContext*,                               const RequestType*, ResponseType*)>      func_;  // The class the above handler function lives in.  ServiceType* service_;};/// A wrapper class of an application provided client streaming handler.template <class ServiceType, class RequestType, class ResponseType>class ClientStreamingHandler : public ::grpc::internal::MethodHandler { public:  ClientStreamingHandler(      std::function<::grpc::Status(ServiceType*, ::grpc::ServerContext*,                                   ServerReader<RequestType>*, ResponseType*)>          func,      ServiceType* service)      : func_(func), service_(service) {}  void RunHandler(const HandlerParameter& param) final {    ServerReader<RequestType> reader(        param.call, static_cast<::grpc::ServerContext*>(param.server_context));    ResponseType rsp;    ::grpc::Status status = CatchingFunctionHandler([this, ¶m, &reader,                                                     &rsp] {      return func_(service_,                   static_cast<::grpc::ServerContext*>(param.server_context),                   &reader, &rsp);    });    ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,                                ::grpc::internal::CallOpSendMessage,                                ::grpc::internal::CallOpServerSendStatus>        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<::grpc::Status(ServiceType*, ::grpc::ServerContext*,                               ServerReader<RequestType>*, ResponseType*)>      func_;  ServiceType* service_;};/// A wrapper class of an application provided server streaming handler.template <class ServiceType, class RequestType, class ResponseType>class ServerStreamingHandler : public ::grpc::internal::MethodHandler { public:  ServerStreamingHandler(std::function<::grpc::Status(                             ServiceType*, ::grpc::ServerContext*,                             const RequestType*, ServerWriter<ResponseType>*)>                             func,                         ServiceType* service)      : func_(func), service_(service) {}  void RunHandler(const HandlerParameter& param) final {    ::grpc::Status status = param.status;    if (status.ok()) {      ServerWriter<ResponseType> writer(          param.call,          static_cast<::grpc::ServerContext*>(param.server_context));      status = CatchingFunctionHandler([this, ¶m, &writer] {        return func_(service_,                     static_cast<::grpc::ServerContext*>(param.server_context),                     static_cast<RequestType*>(param.request), &writer);      });      static_cast<RequestType*>(param.request)->~RequestType();    }    ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,                                ::grpc::internal::CallOpServerSendStatus>        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::g_core_codegen_interface->grpc_call_arena_alloc(            call, sizeof(RequestType))) RequestType();    *status =        ::grpc::SerializationTraits<RequestType>::Deserialize(&buf, request);    buf.Release();    if (status->ok()) {      return request;    }    request->~RequestType();    return nullptr;  } private:  std::function<::grpc::Status(ServiceType*, ::grpc::ServerContext*,                               const RequestType*, ServerWriter<ResponseType>*)>      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 Streamer, bool WriteNeeded>class TemplatedBidiStreamingHandler : public ::grpc::internal::MethodHandler { public:  explicit TemplatedBidiStreamingHandler(      std::function<::grpc::Status(::grpc::ServerContext*, Streamer*)> func)      : func_(func), write_needed_(WriteNeeded) {}  void RunHandler(const HandlerParameter& param) final {    Streamer stream(param.call,                    static_cast<::grpc::ServerContext*>(param.server_context));    ::grpc::Status status = CatchingFunctionHandler([this, ¶m, &stream] {      return func_(static_cast<::grpc::ServerContext*>(param.server_context),                   &stream);    });    ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,                                ::grpc::internal::CallOpServerSendStatus>        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<::grpc::Status(::grpc::ServerContext*, Streamer*)> func_;  const bool write_needed_;};template <class ServiceType, class RequestType, class ResponseType>class BidiStreamingHandler    : public TemplatedBidiStreamingHandler<          ServerReaderWriter<ResponseType, RequestType>, false> { public:  BidiStreamingHandler(std::function<::grpc::Status(                           ServiceType*, ::grpc::ServerContext*,                           ServerReaderWriter<ResponseType, RequestType>*)>                           func,                       ServiceType* service)      // TODO(vjpai): When gRPC supports C++14, move-capture func in the below      : TemplatedBidiStreamingHandler<            ServerReaderWriter<ResponseType, RequestType>, false>(            [func, service](                ::grpc::ServerContext* ctx,                ServerReaderWriter<ResponseType, RequestType>* streamer) {              return func(service, ctx, streamer);            }) {}};template <class RequestType, class ResponseType>class StreamedUnaryHandler    : public TemplatedBidiStreamingHandler<          ServerUnaryStreamer<RequestType, ResponseType>, true> { public:  explicit StreamedUnaryHandler(      std::function<          ::grpc::Status(::grpc::ServerContext*,                         ServerUnaryStreamer<RequestType, ResponseType>*)>          func)      : TemplatedBidiStreamingHandler<            ServerUnaryStreamer<RequestType, ResponseType>, true>(            std::move(func)) {}};template <class RequestType, class ResponseType>class SplitServerStreamingHandler    : public TemplatedBidiStreamingHandler<          ServerSplitStreamer<RequestType, ResponseType>, false> { public:  explicit SplitServerStreamingHandler(      std::function<          ::grpc::Status(::grpc::ServerContext*,                         ServerSplitStreamer<RequestType, ResponseType>*)>          func)      : TemplatedBidiStreamingHandler<            ServerSplitStreamer<RequestType, ResponseType>, 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 <::grpc::StatusCode code>class ErrorMethodHandler : public ::grpc::internal::MethodHandler { public:  template <class T>  static void FillOps(::grpc::ServerContextBase* context, T* ops) {    ::grpc::Status status(code, "");    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<::grpc::internal::CallOpSendInitialMetadata,                                ::grpc::internal::CallOpServerSendStatus>        ops;    FillOps(param.server_context, &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::g_core_codegen_interface->grpc_byte_buffer_destroy(req);    }    return nullptr;  }};typedef ErrorMethodHandler<::grpc::StatusCode::UNIMPLEMENTED>    UnknownMethodHandler;typedef ErrorMethodHandler<::grpc::StatusCode::RESOURCE_EXHAUSTED>    ResourceExhaustedHandler;}  // namespace internal}  // namespace grpc#endif  // GRPCPP_IMPL_CODEGEN_METHOD_HANDLER_H
 |