From 84e763f10a1e10d36c7de35970f9d25958ee2e16 Mon Sep 17 00:00:00 2001 From: Vijay Pai Date: Mon, 8 Oct 2018 13:28:10 -0700 Subject: Experimental C++ server callback unary API --- src/compiler/cpp_generator.cc | 186 +++++++++++++++++++++++++++++++++++++++++- 1 file changed, 185 insertions(+), 1 deletion(-) (limited to 'src/compiler') diff --git a/src/compiler/cpp_generator.cc b/src/compiler/cpp_generator.cc index 56716493dc..1e17760092 100644 --- a/src/compiler/cpp_generator.cc +++ b/src/compiler/cpp_generator.cc @@ -135,6 +135,7 @@ grpc::string GetHeaderIncludes(grpc_generator::File* file, "grpcpp/impl/codegen/method_handler_impl.h", "grpcpp/impl/codegen/proto_utils.h", "grpcpp/impl/codegen/rpc_method.h", + "grpcpp/impl/codegen/server_callback.h", "grpcpp/impl/codegen/service_type.h", "grpcpp/impl/codegen/status.h", "grpcpp/impl/codegen/stub_options.h", @@ -702,7 +703,7 @@ void PrintHeaderServerMethodSync(grpc_generator::Printer* printer, // Helper generator. Disabled the sync API for Request and Response, then adds // in an async API for RealRequest and RealResponse types. This is to be used -// to generate async and raw APIs. +// to generate async and raw async APIs. void PrintHeaderServerAsyncMethodsHelper( grpc_generator::Printer* printer, const grpc_generator::Method* method, std::map* vars) { @@ -829,6 +830,164 @@ void PrintHeaderServerMethodAsync(grpc_generator::Printer* printer, printer->Print(*vars, "};\n"); } +// Helper generator. Disabled the sync API for Request and Response, then adds +// in a callback API for RealRequest and RealResponse types. This is to be used +// to generate callback and raw callback APIs. +void PrintHeaderServerCallbackMethodsHelper( + grpc_generator::Printer* printer, const grpc_generator::Method* method, + std::map* vars) { + if (method->NoStreaming()) { + printer->Print( + *vars, + "// disable synchronous version of this method\n" + "::grpc::Status $Method$(" + "::grpc::ServerContext* context, const $Request$* request, " + "$Response$* response) override {\n" + " abort();\n" + " return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, \"\");\n" + "}\n"); + printer->Print( + *vars, + "virtual void $Method$(" + "::grpc::ServerContext* context, const $RealRequest$* request, " + "$RealResponse$* response, " + "::grpc::experimental::ServerCallbackRpcController* " + "controller) { controller->Finish(::grpc::Status(" + "::grpc::StatusCode::UNIMPLEMENTED, \"\")); }\n"); + } else if (ClientOnlyStreaming(method)) { + printer->Print( + *vars, + "// disable synchronous version of this method\n" + "::grpc::Status $Method$(" + "::grpc::ServerContext* context, " + "::grpc::ServerReader< $Request$>* reader, " + "$Response$* response) override {\n" + " abort();\n" + " return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, \"\");\n" + "}\n"); + } else if (ServerOnlyStreaming(method)) { + printer->Print( + *vars, + "// disable synchronous version of this method\n" + "::grpc::Status $Method$(" + "::grpc::ServerContext* context, const $Request$* request, " + "::grpc::ServerWriter< $Response$>* writer) override " + "{\n" + " abort();\n" + " return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, \"\");\n" + "}\n"); + } else if (method->BidiStreaming()) { + printer->Print( + *vars, + "// disable synchronous version of this method\n" + "::grpc::Status $Method$(" + "::grpc::ServerContext* context, " + "::grpc::ServerReaderWriter< $Response$, $Request$>* stream) " + " override {\n" + " abort();\n" + " return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, \"\");\n" + "}\n"); + } +} + +void PrintHeaderServerMethodCallback( + grpc_generator::Printer* printer, const grpc_generator::Method* method, + std::map* vars) { + (*vars)["Method"] = method->name(); + // These will be disabled + (*vars)["Request"] = method->input_type_name(); + (*vars)["Response"] = method->output_type_name(); + // These will be used for the callback API + (*vars)["RealRequest"] = method->input_type_name(); + (*vars)["RealResponse"] = method->output_type_name(); + printer->Print(*vars, "template \n"); + printer->Print( + *vars, + "class ExperimentalWithCallbackMethod_$Method$ : public BaseClass {\n"); + printer->Print( + " private:\n" + " void BaseClassMustBeDerivedFromService(const Service *service) {}\n"); + printer->Print(" public:\n"); + printer->Indent(); + printer->Print(*vars, "ExperimentalWithCallbackMethod_$Method$() {\n"); + if (method->NoStreaming()) { + printer->Print( + *vars, + " ::grpc::Service::experimental().MarkMethodCallback($Idx$,\n" + " new ::grpc::internal::CallbackUnaryHandler< " + "ExperimentalWithCallbackMethod_$Method$, $RealRequest$, " + "$RealResponse$>(\n" + " [this](::grpc::ServerContext* context,\n" + " const $RealRequest$* request,\n" + " $RealResponse$* response,\n" + " ::grpc::experimental::ServerCallbackRpcController* " + "controller) {\n" + " this->$" + "Method$(context, request, response, controller);\n" + " }, this));\n"); + } else if (ClientOnlyStreaming(method)) { + } else if (ServerOnlyStreaming(method)) { + } else if (method->BidiStreaming()) { + } + printer->Print(*vars, "}\n"); + printer->Print(*vars, + "~ExperimentalWithCallbackMethod_$Method$() override {\n" + " BaseClassMustBeDerivedFromService(this);\n" + "}\n"); + PrintHeaderServerCallbackMethodsHelper(printer, method, vars); + printer->Outdent(); + printer->Print(*vars, "};\n"); +} + +void PrintHeaderServerMethodRawCallback( + grpc_generator::Printer* printer, const grpc_generator::Method* method, + std::map* vars) { + (*vars)["Method"] = method->name(); + // These will be disabled + (*vars)["Request"] = method->input_type_name(); + (*vars)["Response"] = method->output_type_name(); + // These will be used for raw API + (*vars)["RealRequest"] = "::grpc::ByteBuffer"; + (*vars)["RealResponse"] = "::grpc::ByteBuffer"; + printer->Print(*vars, "template \n"); + printer->Print(*vars, + "class ExperimentalWithRawCallbackMethod_$Method$ : public " + "BaseClass {\n"); + printer->Print( + " private:\n" + " void BaseClassMustBeDerivedFromService(const Service *service) {}\n"); + printer->Print(" public:\n"); + printer->Indent(); + printer->Print(*vars, "ExperimentalWithRawCallbackMethod_$Method$() {\n"); + if (method->NoStreaming()) { + printer->Print( + *vars, + " ::grpc::Service::experimental().MarkMethodRawCallback($Idx$,\n" + " new ::grpc::internal::CallbackUnaryHandler< " + "ExperimentalWithRawCallbackMethod_$Method$, $RealRequest$, " + "$RealResponse$>(\n" + " [this](::grpc::ServerContext* context,\n" + " const $RealRequest$* request,\n" + " $RealResponse$* response,\n" + " ::grpc::experimental::ServerCallbackRpcController* " + "controller) {\n" + " this->$" + "Method$(context, request, response, controller);\n" + " }, this));\n"); + } else if (ClientOnlyStreaming(method)) { + } else if (ServerOnlyStreaming(method)) { + } else if (method->BidiStreaming()) { + } + printer->Print(*vars, "}\n"); + printer->Print(*vars, + "~ExperimentalWithRawCallbackMethod_$Method$() override {\n" + " BaseClassMustBeDerivedFromService(this);\n" + "}\n"); + PrintHeaderServerCallbackMethodsHelper(printer, method, vars); + printer->Outdent(); + printer->Print(*vars, "};\n"); +} + void PrintHeaderServerMethodStreamedUnary( grpc_generator::Printer* printer, const grpc_generator::Method* method, std::map* vars) { @@ -1146,6 +1305,24 @@ void PrintHeaderService(grpc_generator::Printer* printer, } printer->Print(" AsyncService;\n"); + // Server side - Callback + for (int i = 0; i < service->method_count(); ++i) { + (*vars)["Idx"] = as_string(i); + PrintHeaderServerMethodCallback(printer, service->method(i).get(), vars); + } + + printer->Print("typedef "); + + for (int i = 0; i < service->method_count(); ++i) { + (*vars)["method_name"] = service->method(i).get()->name(); + printer->Print(*vars, "ExperimentalWithCallbackMethod_$method_name$<"); + } + printer->Print("Service"); + for (int i = 0; i < service->method_count(); ++i) { + printer->Print(" >"); + } + printer->Print(" ExperimentalCallbackService;\n"); + // Server side - Generic for (int i = 0; i < service->method_count(); ++i) { (*vars)["Idx"] = as_string(i); @@ -1158,6 +1335,12 @@ void PrintHeaderService(grpc_generator::Printer* printer, PrintHeaderServerMethodRaw(printer, service->method(i).get(), vars); } + // Server side - Raw Callback + for (int i = 0; i < service->method_count(); ++i) { + (*vars)["Idx"] = as_string(i); + PrintHeaderServerMethodRawCallback(printer, service->method(i).get(), vars); + } + // Server side - Streamed Unary for (int i = 0; i < service->method_count(); ++i) { (*vars)["Idx"] = as_string(i); @@ -1333,6 +1516,7 @@ grpc::string GetSourceIncludes(grpc_generator::File* file, "grpcpp/impl/codegen/client_callback.h", "grpcpp/impl/codegen/method_handler_impl.h", "grpcpp/impl/codegen/rpc_service_method.h", + "grpcpp/impl/codegen/server_callback.h", "grpcpp/impl/codegen/service_type.h", "grpcpp/impl/codegen/sync_stream.h"}; std::vector headers(headers_strs, array_end(headers_strs)); -- cgit v1.2.3 From b460622c2d8be2bddcb60cf0682d1b88d9c2357e Mon Sep 17 00:00:00 2001 From: Vijay Pai Date: Tue, 30 Oct 2018 12:35:56 -0700 Subject: Address reviewer comments --- include/grpcpp/impl/codegen/rpc_service_method.h | 10 +++++++++- include/grpcpp/impl/codegen/server_context.h | 1 + src/compiler/cpp_generator.cc | 4 ++-- 3 files changed, 12 insertions(+), 3 deletions(-) (limited to 'src/compiler') diff --git a/include/grpcpp/impl/codegen/rpc_service_method.h b/include/grpcpp/impl/codegen/rpc_service_method.h index 44d9b8ad63..1890e4254d 100644 --- a/include/grpcpp/impl/codegen/rpc_service_method.h +++ b/include/grpcpp/impl/codegen/rpc_service_method.h @@ -40,6 +40,14 @@ class MethodHandler { public: virtual ~MethodHandler() {} struct HandlerParameter { + /// Constructor for HandlerParameter + /// + /// \param c : the gRPC Call structure for this server call + /// \param context : the ServerContext structure for this server call + /// \param req : the request payload, if appropriate for this RPC + /// \param req_status : the request status after any interceptors have run + /// \param renew : used only by the callback API. It is a function + /// called by the RPC Controller to request another RPC HandlerParameter(Call* c, ServerContext* context, void* req, Status req_status, std::function renew) : call(c), @@ -120,7 +128,7 @@ class RpcServiceMethod : public RpcMethod { const char* TypeToString(RpcServiceMethod::ApiType type) { switch (type) { case ApiType::SYNC: - return "unset"; + return "sync"; case ApiType::ASYNC: return "async"; case ApiType::RAW: diff --git a/include/grpcpp/impl/codegen/server_context.h b/include/grpcpp/impl/codegen/server_context.h index ebbf64bc6d..ecb9073cf9 100644 --- a/include/grpcpp/impl/codegen/server_context.h +++ b/include/grpcpp/impl/codegen/server_context.h @@ -292,6 +292,7 @@ class ServerContext { void BindDeadlineAndMetadata(gpr_timespec deadline, grpc_metadata_array* arr); void Clear(); + void Setup(gpr_timespec deadline); uint32_t initial_metadata_flags() const { return 0; } diff --git a/src/compiler/cpp_generator.cc b/src/compiler/cpp_generator.cc index 1e17760092..fcff75a420 100644 --- a/src/compiler/cpp_generator.cc +++ b/src/compiler/cpp_generator.cc @@ -701,7 +701,7 @@ void PrintHeaderServerMethodSync(grpc_generator::Printer* printer, printer->Print(method->GetTrailingComments("//").c_str()); } -// Helper generator. Disabled the sync API for Request and Response, then adds +// Helper generator. Disables the sync API for Request and Response, then adds // in an async API for RealRequest and RealResponse types. This is to be used // to generate async and raw async APIs. void PrintHeaderServerAsyncMethodsHelper( @@ -830,7 +830,7 @@ void PrintHeaderServerMethodAsync(grpc_generator::Printer* printer, printer->Print(*vars, "};\n"); } -// Helper generator. Disabled the sync API for Request and Response, then adds +// Helper generator. Disables the sync API for Request and Response, then adds // in a callback API for RealRequest and RealResponse types. This is to be used // to generate callback and raw callback APIs. void PrintHeaderServerCallbackMethodsHelper( -- cgit v1.2.3 From 932abf48a3fd0757a05bdbae4038c20fd521f312 Mon Sep 17 00:00:00 2001 From: Vijay Pai Date: Wed, 31 Oct 2018 01:02:07 -0700 Subject: Address reviewer comments. --- include/grpcpp/impl/codegen/async_unary_call.h | 2 +- include/grpcpp/impl/codegen/call_op_set.h | 22 +++--- .../grpcpp/impl/codegen/call_op_set_interface.h | 4 +- include/grpcpp/impl/codegen/client_callback.h | 2 +- include/grpcpp/impl/codegen/rpc_service_method.h | 11 +-- include/grpcpp/impl/codegen/server_callback.h | 80 ++++++++++------------ src/compiler/cpp_generator.cc | 6 ++ src/cpp/client/channel_cc.cc | 4 +- src/cpp/common/completion_queue_cc.cc | 10 +-- src/cpp/server/server_cc.cc | 4 +- src/cpp/server/server_context.cc | 16 ++--- 11 files changed, 82 insertions(+), 79 deletions(-) (limited to 'src/compiler') diff --git a/include/grpcpp/impl/codegen/async_unary_call.h b/include/grpcpp/impl/codegen/async_unary_call.h index 744b128141..89dcb12418 100644 --- a/include/grpcpp/impl/codegen/async_unary_call.h +++ b/include/grpcpp/impl/codegen/async_unary_call.h @@ -240,7 +240,7 @@ class ServerAsyncResponseWriter final /// metadata. void Finish(const W& msg, const Status& status, void* tag) { finish_buf_.set_output_tag(tag); - finish_buf_.set_cq_tag(&finish_buf_); + finish_buf_.set_core_cq_tag(&finish_buf_); if (!ctx_->sent_initial_metadata_) { finish_buf_.SendInitialMetadata(&ctx_->initial_metadata_, ctx_->initial_metadata_flags()); diff --git a/include/grpcpp/impl/codegen/call_op_set.h b/include/grpcpp/impl/codegen/call_op_set.h index 785688e67f..5c52b027b2 100644 --- a/include/grpcpp/impl/codegen/call_op_set.h +++ b/include/grpcpp/impl/codegen/call_op_set.h @@ -770,19 +770,19 @@ class CallOpSet : public CallOpSetInterface, public Op5, public Op6 { public: - CallOpSet() : cq_tag_(this), return_tag_(this) {} + CallOpSet() : core_cq_tag_(this), return_tag_(this) {} // The copy constructor and assignment operator reset the value of - // cq_tag_, return_tag_, done_intercepting_ and interceptor_methods_ since - // those are only meaningful on a specific object, not across objects. + // core_cq_tag_, return_tag_, done_intercepting_ and interceptor_methods_ + // since those are only meaningful on a specific object, not across objects. CallOpSet(const CallOpSet& other) - : cq_tag_(this), + : core_cq_tag_(this), return_tag_(this), call_(other.call_), done_intercepting_(false), interceptor_methods_(InterceptorBatchMethodsImpl()) {} CallOpSet& operator=(const CallOpSet& other) { - cq_tag_ = this; + core_cq_tag_ = this; return_tag_ = this; call_ = other.call_; done_intercepting_ = false; @@ -834,13 +834,13 @@ class CallOpSet : public CallOpSetInterface, void set_output_tag(void* return_tag) { return_tag_ = return_tag; } - void* cq_tag() override { return cq_tag_; } + void* core_cq_tag() override { return core_cq_tag_; } - /// set_cq_tag is used to provide a different core CQ tag than "this". + /// set_core_cq_tag is used to provide a different core CQ tag than "this". /// This is used for callback-based tags, where the core tag is the core /// callback function. It does not change the use or behavior of any other /// function (such as FinalizeResult) - void set_cq_tag(void* cq_tag) { cq_tag_ = cq_tag; } + void set_core_cq_tag(void* core_cq_tag) { core_cq_tag_ = core_cq_tag; } // This will be called while interceptors are run if the RPC is a hijacked // RPC. This should set hijacking state for each of the ops. @@ -866,7 +866,7 @@ class CallOpSet : public CallOpSetInterface, this->Op6::AddOp(ops, &nops); GPR_CODEGEN_ASSERT(GRPC_CALL_OK == g_core_codegen_interface->grpc_call_start_batch( - call_.call(), ops, nops, cq_tag(), nullptr)); + call_.call(), ops, nops, core_cq_tag(), nullptr)); } // Should be called after interceptors are done running on the finalize result @@ -875,7 +875,7 @@ class CallOpSet : public CallOpSetInterface, done_intercepting_ = true; GPR_CODEGEN_ASSERT(GRPC_CALL_OK == g_core_codegen_interface->grpc_call_start_batch( - call_.call(), nullptr, 0, cq_tag(), nullptr)); + call_.call(), nullptr, 0, core_cq_tag(), nullptr)); } private: @@ -906,7 +906,7 @@ class CallOpSet : public CallOpSetInterface, return interceptor_methods_.RunInterceptors(); } - void* cq_tag_; + void* core_cq_tag_; void* return_tag_; Call call_; bool done_intercepting_ = false; diff --git a/include/grpcpp/impl/codegen/call_op_set_interface.h b/include/grpcpp/impl/codegen/call_op_set_interface.h index 815227a299..3b74566a6d 100644 --- a/include/grpcpp/impl/codegen/call_op_set_interface.h +++ b/include/grpcpp/impl/codegen/call_op_set_interface.h @@ -38,9 +38,9 @@ class CallOpSetInterface : public CompletionQueueTag { virtual void FillOps(internal::Call* call) = 0; /// Get the tag to be used at the core completion queue. Generally, the - /// value of cq_tag will be "this". However, it can be overridden if we + /// value of core_cq_tag will be "this". However, it can be overridden if we /// want core to process the tag differently (e.g., as a core callback) - virtual void* cq_tag() = 0; + virtual void* core_cq_tag() = 0; // This will be called while interceptors are run if the RPC is a hijacked // RPC. This should set hijacking state for each of the ops. diff --git a/include/grpcpp/impl/codegen/client_callback.h b/include/grpcpp/impl/codegen/client_callback.h index ecb00a0769..4baa819091 100644 --- a/include/grpcpp/impl/codegen/client_callback.h +++ b/include/grpcpp/impl/codegen/client_callback.h @@ -84,7 +84,7 @@ class CallbackUnaryCallImpl { ops->AllowNoMessage(); ops->ClientSendClose(); ops->ClientRecvStatus(context, tag->status_ptr()); - ops->set_cq_tag(tag); + ops->set_core_cq_tag(tag); call.PerformOps(ops); } }; diff --git a/include/grpcpp/impl/codegen/rpc_service_method.h b/include/grpcpp/impl/codegen/rpc_service_method.h index 1890e4254d..f465c5fc2f 100644 --- a/include/grpcpp/impl/codegen/rpc_service_method.h +++ b/include/grpcpp/impl/codegen/rpc_service_method.h @@ -46,21 +46,22 @@ class MethodHandler { /// \param context : the ServerContext structure for this server call /// \param req : the request payload, if appropriate for this RPC /// \param req_status : the request status after any interceptors have run - /// \param renew : used only by the callback API. It is a function - /// called by the RPC Controller to request another RPC + /// \param rpc_requester : used only by the callback API. It is a function + /// called by the RPC Controller to request another RPC (and also + /// to set up the state required to make that request possible) HandlerParameter(Call* c, ServerContext* context, void* req, - Status req_status, std::function renew) + Status req_status, std::function requester) : call(c), server_context(context), request(req), status(req_status), - renewer(std::move(renew)) {} + call_requester(std::move(requester)) {} ~HandlerParameter() {} Call* call; ServerContext* server_context; void* request; Status status; - std::function renewer; + std::function call_requester; }; virtual void RunHandler(const HandlerParameter& param) = 0; diff --git a/include/grpcpp/impl/codegen/server_callback.h b/include/grpcpp/impl/codegen/server_callback.h index c8f7510ed5..5d56cbf1df 100644 --- a/include/grpcpp/impl/codegen/server_callback.h +++ b/include/grpcpp/impl/codegen/server_callback.h @@ -48,7 +48,6 @@ class ServerCallbackRpcController { // The method handler must call this function when it is done so that // the library knows to free its resources virtual void Finish(Status s) = 0; - virtual void FinishWithError(Status s) = 0; // Allow the method handler to push out the initial metadata before // the response and status are ready @@ -75,7 +74,8 @@ class CallbackUnaryHandler : public MethodHandler { param.call->call(), sizeof(ServerCallbackRpcControllerImpl))) ServerCallbackRpcControllerImpl( param.server_context, param.call, - static_cast(param.request), std::move(param.renewer)); + static_cast(param.request), + std::move(param.call_requester)); Status status = param.status; if (status.ok()) { @@ -112,52 +112,19 @@ class CallbackUnaryHandler : public MethodHandler { // The implementation class of ServerCallbackRpcController is a private member // of CallbackUnaryHandler since it is never exposed anywhere, and this allows // it to take advantage of CallbackUnaryHandler's friendships. - class ServerCallbackRpcControllerImpl : public experimental::ServerCallbackRpcController { public: - void Finish(Status s) override { FinishInternal(std::move(s), false); } - - void FinishWithError(Status s) override { - FinishInternal(std::move(s), true); - } - - void SendInitialMetadata(std::function f) override { - GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_); - - meta_tag_ = - CallbackWithSuccessTag(call_.call(), std::move(f), &meta_buf_); - meta_buf_.SendInitialMetadata(&ctx_->initial_metadata_, - ctx_->initial_metadata_flags()); - if (ctx_->compression_level_set()) { - meta_buf_.set_compression_level(ctx_->compression_level()); - } - ctx_->sent_initial_metadata_ = true; - meta_buf_.set_cq_tag(&meta_tag_); - call_.PerformOps(&meta_buf_); - } - - private: - template - friend class CallbackUnaryHandler; - - ServerCallbackRpcControllerImpl(ServerContext* ctx, Call* call, - RequestType* req, - std::function renewer) - : ctx_(ctx), call_(*call), req_(req), renewer_(std::move(renewer)) {} - - ~ServerCallbackRpcControllerImpl() { req_->~RequestType(); } - - void FinishInternal(Status s, bool allow_error) { + void Finish(Status s) override { finish_tag_ = CallbackWithSuccessTag( call_.call(), [this](bool) { grpc_call* call = call_.call(); - auto renewer = std::move(renewer_); + auto call_requester = std::move(call_requester_); this->~ServerCallbackRpcControllerImpl(); // explicitly call // destructor g_core_codegen_interface->grpc_call_unref(call); - renewer(); + call_requester(); }, &finish_buf_); if (!ctx_->sent_initial_metadata_) { @@ -168,17 +135,46 @@ class CallbackUnaryHandler : public MethodHandler { } ctx_->sent_initial_metadata_ = true; } - // The response may be dropped if the status is not OK. - if (allow_error || s.ok()) { + // The response is dropped if the status is not OK. + if (s.ok()) { finish_buf_.ServerSendStatus(&ctx_->trailing_metadata_, finish_buf_.SendMessage(resp_)); } else { finish_buf_.ServerSendStatus(&ctx_->trailing_metadata_, s); } - finish_buf_.set_cq_tag(&finish_tag_); + finish_buf_.set_core_cq_tag(&finish_tag_); call_.PerformOps(&finish_buf_); } + void SendInitialMetadata(std::function f) override { + GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_); + + meta_tag_ = + CallbackWithSuccessTag(call_.call(), std::move(f), &meta_buf_); + meta_buf_.SendInitialMetadata(&ctx_->initial_metadata_, + ctx_->initial_metadata_flags()); + if (ctx_->compression_level_set()) { + meta_buf_.set_compression_level(ctx_->compression_level()); + } + ctx_->sent_initial_metadata_ = true; + meta_buf_.set_core_cq_tag(&meta_tag_); + call_.PerformOps(&meta_buf_); + } + + private: + template + friend class CallbackUnaryHandler; + + ServerCallbackRpcControllerImpl(ServerContext* ctx, Call* call, + RequestType* req, + std::function call_requester) + : ctx_(ctx), + call_(*call), + req_(req), + call_requester_(std::move(call_requester)) {} + + ~ServerCallbackRpcControllerImpl() { req_->~RequestType(); } + RequestType* request() { return req_; } ResponseType* response() { return &resp_; } @@ -193,7 +189,7 @@ class CallbackUnaryHandler : public MethodHandler { Call call_; RequestType* req_; ResponseType resp_; - std::function renewer_; + std::function call_requester_; }; }; diff --git a/src/compiler/cpp_generator.cc b/src/compiler/cpp_generator.cc index fcff75a420..34d5beb653 100644 --- a/src/compiler/cpp_generator.cc +++ b/src/compiler/cpp_generator.cc @@ -926,8 +926,11 @@ void PrintHeaderServerMethodCallback( "Method$(context, request, response, controller);\n" " }, this));\n"); } else if (ClientOnlyStreaming(method)) { + // TODO(vjpai): Add in code generation for all streaming methods } else if (ServerOnlyStreaming(method)) { + // TODO(vjpai): Add in code generation for all streaming methods } else if (method->BidiStreaming()) { + // TODO(vjpai): Add in code generation for all streaming methods } printer->Print(*vars, "}\n"); printer->Print(*vars, @@ -975,8 +978,11 @@ void PrintHeaderServerMethodRawCallback( "Method$(context, request, response, controller);\n" " }, this));\n"); } else if (ClientOnlyStreaming(method)) { + // TODO(vjpai): Add in code generation for all streaming methods } else if (ServerOnlyStreaming(method)) { + // TODO(vjpai): Add in code generation for all streaming methods } else if (method->BidiStreaming()) { + // TODO(vjpai): Add in code generation for all streaming methods } printer->Print(*vars, "}\n"); printer->Print(*vars, diff --git a/src/cpp/client/channel_cc.cc b/src/cpp/client/channel_cc.cc index 15e3ccb3c9..f5c1b4e2c5 100644 --- a/src/cpp/client/channel_cc.cc +++ b/src/cpp/client/channel_cc.cc @@ -225,7 +225,7 @@ class ShutdownCallback : public grpc_experimental_completion_queue_functor { static void Run(grpc_experimental_completion_queue_functor* cb, int) { auto* callback = static_cast(cb); delete callback->cq_; - grpc_core::Delete(callback); + delete callback; } private: @@ -238,7 +238,7 @@ CompletionQueue* Channel::CallbackCQ() { // if there is no explicit per-channel CQ registered std::lock_guard l(mu_); if (callback_cq_ == nullptr) { - auto* shutdown_callback = grpc_core::New(); + auto* shutdown_callback = new ShutdownCallback; callback_cq_ = new CompletionQueue(grpc_completion_queue_attributes{ GRPC_CQ_CURRENT_VERSION, GRPC_CQ_CALLBACK, GRPC_CQ_DEFAULT_POLLING, shutdown_callback}); diff --git a/src/cpp/common/completion_queue_cc.cc b/src/cpp/common/completion_queue_cc.cc index 6893201e2e..d93a54aed7 100644 --- a/src/cpp/common/completion_queue_cc.cc +++ b/src/cpp/common/completion_queue_cc.cc @@ -60,10 +60,10 @@ CompletionQueue::NextStatus CompletionQueue::AsyncNextInternal( case GRPC_QUEUE_SHUTDOWN: return SHUTDOWN; case GRPC_OP_COMPLETE: - auto cq_tag = static_cast(ev.tag); + auto core_cq_tag = static_cast(ev.tag); *ok = ev.success != 0; - *tag = cq_tag; - if (cq_tag->FinalizeResult(tag, ok)) { + *tag = core_cq_tag; + if (core_cq_tag->FinalizeResult(tag, ok)) { return GOT_EVENT; } break; @@ -87,9 +87,9 @@ bool CompletionQueue::CompletionQueueTLSCache::Flush(void** tag, bool* ok) { flushed_ = true; if (grpc_completion_queue_thread_local_cache_flush(cq_->cq_, &res_tag, &res)) { - auto cq_tag = static_cast(res_tag); + auto core_cq_tag = static_cast(res_tag); *ok = res == 1; - if (cq_tag->FinalizeResult(tag, ok)) { + if (core_cq_tag->FinalizeResult(tag, ok)) { return true; } } diff --git a/src/cpp/server/server_cc.cc b/src/cpp/server/server_cc.cc index a9f0315af6..870ee84e3e 100644 --- a/src/cpp/server/server_cc.cc +++ b/src/cpp/server/server_cc.cc @@ -1111,7 +1111,7 @@ class ShutdownCallback : public grpc_experimental_completion_queue_functor { static void Run(grpc_experimental_completion_queue_functor* cb, int) { auto* callback = static_cast(cb); delete callback->cq_; - grpc_core::Delete(callback); + delete callback; } private: @@ -1124,7 +1124,7 @@ CompletionQueue* Server::CallbackCQ() { // if there is no explicit per-server CQ registered std::lock_guard l(mu_); if (callback_cq_ == nullptr) { - auto* shutdown_callback = grpc_core::New(); + auto* shutdown_callback = new ShutdownCallback; callback_cq_ = new CompletionQueue(grpc_completion_queue_attributes{ GRPC_CQ_CURRENT_VERSION, GRPC_CQ_CALLBACK, GRPC_CQ_DEFAULT_POLLING, shutdown_callback}); diff --git a/src/cpp/server/server_context.cc b/src/cpp/server/server_context.cc index 8ded7a8972..355debb3fb 100644 --- a/src/cpp/server/server_context.cc +++ b/src/cpp/server/server_context.cc @@ -45,7 +45,7 @@ class ServerContext::CompletionOp final : public internal::CallOpSetInterface { : call_(*call), has_tag_(false), tag_(nullptr), - cq_tag_(this), + core_cq_tag_(this), refs_(2), finalized_(false), cancelled_(0), @@ -92,9 +92,9 @@ class ServerContext::CompletionOp final : public internal::CallOpSetInterface { tag_ = tag; } - void set_cq_tag(void* cq_tag) { cq_tag_ = cq_tag; } + void set_core_cq_tag(void* core_cq_tag) { core_cq_tag_ = core_cq_tag; } - void* cq_tag() override { return cq_tag_; } + void* core_cq_tag() override { return core_cq_tag_; } void Unref(); @@ -138,7 +138,7 @@ class ServerContext::CompletionOp final : public internal::CallOpSetInterface { internal::Call call_; bool has_tag_; void* tag_; - void* cq_tag_; + void* core_cq_tag_; std::mutex mu_; int refs_; bool finalized_; @@ -166,8 +166,8 @@ void ServerContext::CompletionOp::FillOps(internal::Call* call) { interceptor_methods_.SetCall(&call_); interceptor_methods_.SetReverse(); interceptor_methods_.SetCallOpSetInterface(this); - GPR_ASSERT(GRPC_CALL_OK == - grpc_call_start_batch(call->call(), &ops, 1, cq_tag_, nullptr)); + GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(call->call(), &ops, 1, + core_cq_tag_, nullptr)); /* No interceptors to run here */ } @@ -272,7 +272,7 @@ void ServerContext::BeginCompletionOp(internal::Call* call, bool callback) { if (callback) { completion_tag_ = internal::CallbackWithSuccessTag(call->call(), nullptr, completion_op_); - completion_op_->set_cq_tag(&completion_tag_); + completion_op_->set_core_cq_tag(&completion_tag_); } else if (has_notify_when_done_tag_) { completion_op_->set_tag(async_notify_when_done_tag_); } @@ -305,7 +305,7 @@ bool ServerContext::IsCancelled() const { if (completion_tag_) { // When using callback API, this result is always valid. return completion_op_->CheckCancelledAsync(); - } else if (completion_tag_ || has_notify_when_done_tag_) { + } else if (has_notify_when_done_tag_) { // When using async API, the result is only valid // if the tag has already been delivered at the completion queue return completion_op_ && completion_op_->CheckCancelledAsync(); -- cgit v1.2.3 From eea8a8b8264a0cf18a927fd07e765aafa41c717b Mon Sep 17 00:00:00 2001 From: Vijay Pai Date: Thu, 1 Nov 2018 09:53:31 -0700 Subject: Remove redunant get on unique_ptr --- src/compiler/cpp_generator.cc | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) (limited to 'src/compiler') diff --git a/src/compiler/cpp_generator.cc b/src/compiler/cpp_generator.cc index 34d5beb653..7986aca696 100644 --- a/src/compiler/cpp_generator.cc +++ b/src/compiler/cpp_generator.cc @@ -1302,7 +1302,7 @@ void PrintHeaderService(grpc_generator::Printer* printer, printer->Print("typedef "); for (int i = 0; i < service->method_count(); ++i) { - (*vars)["method_name"] = service->method(i).get()->name(); + (*vars)["method_name"] = service->method(i)->name(); printer->Print(*vars, "WithAsyncMethod_$method_name$<"); } printer->Print("Service"); @@ -1320,7 +1320,7 @@ void PrintHeaderService(grpc_generator::Printer* printer, printer->Print("typedef "); for (int i = 0; i < service->method_count(); ++i) { - (*vars)["method_name"] = service->method(i).get()->name(); + (*vars)["method_name"] = service->method(i)->name(); printer->Print(*vars, "ExperimentalWithCallbackMethod_$method_name$<"); } printer->Print("Service"); @@ -1356,7 +1356,7 @@ void PrintHeaderService(grpc_generator::Printer* printer, printer->Print("typedef "); for (int i = 0; i < service->method_count(); ++i) { - (*vars)["method_name"] = service->method(i).get()->name(); + (*vars)["method_name"] = service->method(i)->name(); if (service->method(i)->NoStreaming()) { printer->Print(*vars, "WithStreamedUnaryMethod_$method_name$<"); } @@ -1378,7 +1378,7 @@ void PrintHeaderService(grpc_generator::Printer* printer, printer->Print("typedef "); for (int i = 0; i < service->method_count(); ++i) { - (*vars)["method_name"] = service->method(i).get()->name(); + (*vars)["method_name"] = service->method(i)->name(); auto method = service->method(i); if (ServerOnlyStreaming(method.get())) { printer->Print(*vars, "WithSplitStreamingMethod_$method_name$<"); @@ -1396,7 +1396,7 @@ void PrintHeaderService(grpc_generator::Printer* printer, // Server side - typedef for controlled both unary and server-side streaming printer->Print("typedef "); for (int i = 0; i < service->method_count(); ++i) { - (*vars)["method_name"] = service->method(i).get()->name(); + (*vars)["method_name"] = service->method(i)->name(); auto method = service->method(i); if (ServerOnlyStreaming(method.get())) { printer->Print(*vars, "WithSplitStreamingMethod_$method_name$<"); @@ -1767,7 +1767,7 @@ void PrintSourceService(grpc_generator::Printer* printer, printer->Print(*vars, "static const char* $prefix$$Service$_method_names[] = {\n"); for (int i = 0; i < service->method_count(); ++i) { - (*vars)["Method"] = service->method(i).get()->name(); + (*vars)["Method"] = service->method(i)->name(); printer->Print(*vars, " \"/$Package$$Service$/$Method$\",\n"); } printer->Print(*vars, "};\n\n"); -- cgit v1.2.3 From a50c61edc11568e5e9a38af105d53b802fb20ec9 Mon Sep 17 00:00:00 2001 From: Jan Tattermusch Date: Thu, 8 Nov 2018 10:57:21 +0100 Subject: generate alternative BindService method --- src/compiler/csharp_generator.cc | 34 ++++++++++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) (limited to 'src/compiler') diff --git a/src/compiler/csharp_generator.cc b/src/compiler/csharp_generator.cc index a923ce8e38..82deabe064 100644 --- a/src/compiler/csharp_generator.cc +++ b/src/compiler/csharp_generator.cc @@ -609,6 +609,39 @@ void GenerateBindServiceMethod(Printer* out, const ServiceDescriptor* service) { out->Print("\n"); } +void GenerateBindServiceWithBinderMethod(Printer* out, + const ServiceDescriptor* service) { + out->Print( + "/// Register service method implementations with a service " + "binder. Useful when customizing the service binding logic.\n"); + out->Print( + "/// Service methods will be bound by " + "calling AddMethod on this object." + "\n"); + out->Print( + "/// An object implementing the server-side" + " handling logic.\n"); + out->Print( + "public static void BindService(grpc::ServiceBinderBase serviceBinder, " + "$implclass$ " + "serviceImpl)\n", + "implclass", GetServerClassName(service)); + out->Print("{\n"); + out->Indent(); + + for (int i = 0; i < service->method_count(); i++) { + const MethodDescriptor* method = service->method(i); + out->Print( + "serviceBinder.AddMethod($methodfield$, serviceImpl.$methodname$);\n", + "methodfield", GetMethodFieldName(method), "methodname", + method->name()); + } + + out->Outdent(); + out->Print("}\n"); + out->Print("\n"); +} + void GenerateService(Printer* out, const ServiceDescriptor* service, bool generate_client, bool generate_server, bool internal_access) { @@ -637,6 +670,7 @@ void GenerateService(Printer* out, const ServiceDescriptor* service, } if (generate_server) { GenerateBindServiceMethod(out, service); + GenerateBindServiceWithBinderMethod(out, service); } out->Outdent(); -- cgit v1.2.3 From bff1452e5ed589dde8137f7f0b58db9a5e8eb6e4 Mon Sep 17 00:00:00 2001 From: Jan Tattermusch Date: Wed, 14 Nov 2018 18:15:39 +0100 Subject: mark as experimental --- src/compiler/csharp_generator.cc | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) (limited to 'src/compiler') diff --git a/src/compiler/csharp_generator.cc b/src/compiler/csharp_generator.cc index 82deabe064..59ddbd82f6 100644 --- a/src/compiler/csharp_generator.cc +++ b/src/compiler/csharp_generator.cc @@ -613,7 +613,10 @@ void GenerateBindServiceWithBinderMethod(Printer* out, const ServiceDescriptor* service) { out->Print( "/// Register service method implementations with a service " - "binder. Useful when customizing the service binding logic.\n"); + "binder. Useful when customizing the service binding logic.\n" + "/// Note: this method is part of an experimental API that can change or " + "be " + "removed without any prior notice.\n"); out->Print( "/// Service methods will be bound by " "calling AddMethod on this object." -- cgit v1.2.3 From d7eb26648d33912199573b842557b0e92ec3071f Mon Sep 17 00:00:00 2001 From: Vijay Pai Date: Sun, 4 Nov 2018 23:37:00 -0800 Subject: Client callback streaming --- include/grpcpp/generic/generic_stub.h | 5 + include/grpcpp/impl/codegen/callback_common.h | 8 +- include/grpcpp/impl/codegen/channel_interface.h | 12 + include/grpcpp/impl/codegen/client_callback.h | 505 +++++++++++++++++++++++ include/grpcpp/impl/codegen/client_context.h | 12 + src/compiler/cpp_generator.cc | 84 +++- src/cpp/client/generic_stub.cc | 12 + test/cpp/codegen/compiler_test_golden | 7 + test/cpp/end2end/client_callback_end2end_test.cc | 228 ++++++++++ 9 files changed, 860 insertions(+), 13 deletions(-) (limited to 'src/compiler') diff --git a/include/grpcpp/generic/generic_stub.h b/include/grpcpp/generic/generic_stub.h index d509d9a520..ccbf8a0e55 100644 --- a/include/grpcpp/generic/generic_stub.h +++ b/include/grpcpp/generic/generic_stub.h @@ -24,6 +24,7 @@ #include #include #include +#include #include namespace grpc { @@ -76,6 +77,10 @@ class GenericStub final { const ByteBuffer* request, ByteBuffer* response, std::function on_completion); + experimental::ClientCallbackReaderWriter* + PrepareBidiStreamingCall(ClientContext* context, const grpc::string& method, + experimental::ClientBidiReactor* reactor); + private: GenericStub* stub_; }; diff --git a/include/grpcpp/impl/codegen/callback_common.h b/include/grpcpp/impl/codegen/callback_common.h index 51367cf550..f7a24204dc 100644 --- a/include/grpcpp/impl/codegen/callback_common.h +++ b/include/grpcpp/impl/codegen/callback_common.h @@ -145,18 +145,19 @@ class CallbackWithSuccessTag // or on a tag that has been Set before unless the tag has been cleared. void Set(grpc_call* call, std::function f, CompletionQueueTag* ops) { + GPR_CODEGEN_ASSERT(call_ == nullptr); + g_core_codegen_interface->grpc_call_ref(call); call_ = call; func_ = std::move(f); ops_ = ops; - g_core_codegen_interface->grpc_call_ref(call); functor_run = &CallbackWithSuccessTag::StaticRun; } void Clear() { if (call_ != nullptr) { - func_ = nullptr; grpc_call* call = call_; call_ = nullptr; + func_ = nullptr; g_core_codegen_interface->grpc_call_unref(call); } } @@ -182,10 +183,9 @@ class CallbackWithSuccessTag } void Run(bool ok) { void* ignored = ops_; - bool new_ok = ok; // Allow a "false" return value from FinalizeResult to silence the // callback, just as it silences a CQ tag in the async cases - bool do_callback = ops_->FinalizeResult(&ignored, &new_ok); + bool do_callback = ops_->FinalizeResult(&ignored, &ok); GPR_CODEGEN_ASSERT(ignored == ops_); if (do_callback) { diff --git a/include/grpcpp/impl/codegen/channel_interface.h b/include/grpcpp/impl/codegen/channel_interface.h index 6ec1ffb8c7..728a7b9049 100644 --- a/include/grpcpp/impl/codegen/channel_interface.h +++ b/include/grpcpp/impl/codegen/channel_interface.h @@ -53,6 +53,12 @@ template class ClientAsyncReaderWriterFactory; template class ClientAsyncResponseReaderFactory; +template +class ClientCallbackReaderWriterFactory; +template +class ClientCallbackReaderFactory; +template +class ClientCallbackWriterFactory; class InterceptedChannel; } // namespace internal @@ -106,6 +112,12 @@ class ChannelInterface { friend class ::grpc::internal::ClientAsyncReaderWriterFactory; template friend class ::grpc::internal::ClientAsyncResponseReaderFactory; + template + friend class ::grpc::internal::ClientCallbackReaderWriterFactory; + template + friend class ::grpc::internal::ClientCallbackReaderFactory; + template + friend class ::grpc::internal::ClientCallbackWriterFactory; template friend class ::grpc::internal::BlockingUnaryCallImpl; template diff --git a/include/grpcpp/impl/codegen/client_callback.h b/include/grpcpp/impl/codegen/client_callback.h index 4baa819091..01a28a9e6f 100644 --- a/include/grpcpp/impl/codegen/client_callback.h +++ b/include/grpcpp/impl/codegen/client_callback.h @@ -22,6 +22,7 @@ #include #include +#include #include #include #include @@ -88,6 +89,510 @@ class CallbackUnaryCallImpl { call.PerformOps(ops); } }; +} // namespace internal + +namespace experimental { + +// The user must implement this reactor interface with reactions to each event +// type that gets called by the library. An empty reaction is provided by +// default + +class ClientBidiReactor { + public: + virtual ~ClientBidiReactor() {} + virtual void OnDone(Status s) {} + virtual void OnReadInitialMetadataDone(bool ok) {} + virtual void OnReadDone(bool ok) {} + virtual void OnWriteDone(bool ok) {} + virtual void OnWritesDoneDone(bool ok) {} +}; + +class ClientReadReactor { + public: + virtual ~ClientReadReactor() {} + virtual void OnDone(Status s) {} + virtual void OnReadInitialMetadataDone(bool ok) {} + virtual void OnReadDone(bool ok) {} +}; + +class ClientWriteReactor { + public: + virtual ~ClientWriteReactor() {} + virtual void OnDone(Status s) {} + virtual void OnReadInitialMetadataDone(bool ok) {} + virtual void OnWriteDone(bool ok) {} + virtual void OnWritesDoneDone(bool ok) {} +}; + +template +class ClientCallbackReaderWriter { + public: + virtual ~ClientCallbackReaderWriter() {} + virtual void StartCall() = 0; + void Write(const Request* req) { Write(req, WriteOptions()); } + virtual void Write(const Request* req, WriteOptions options) = 0; + void WriteLast(const Request* req, WriteOptions options) { + Write(req, options.set_last_message()); + } + virtual void WritesDone() = 0; + virtual void Read(Response* resp) = 0; +}; + +template +class ClientCallbackReader { + public: + virtual ~ClientCallbackReader() {} + virtual void StartCall() = 0; + virtual void Read(Response* resp) = 0; +}; + +template +class ClientCallbackWriter { + public: + virtual ~ClientCallbackWriter() {} + virtual void StartCall() = 0; + void Write(const Request* req) { Write(req, WriteOptions()); } + virtual void Write(const Request* req, WriteOptions options) = 0; + void WriteLast(const Request* req, WriteOptions options) { + Write(req, options.set_last_message()); + } + virtual void WritesDone() = 0; +}; + +} // namespace experimental + +namespace internal { + +// Forward declare factory classes for friendship +template +class ClientCallbackReaderWriterFactory; +template +class ClientCallbackReaderFactory; +template +class ClientCallbackWriterFactory; + +template +class ClientCallbackReaderWriterImpl + : public ::grpc::experimental::ClientCallbackReaderWriter { + public: + // always allocated against a call arena, no memory free required + static void operator delete(void* ptr, std::size_t size) { + assert(size == sizeof(ClientCallbackReaderWriterImpl)); + } + + // This operator should never be called as the memory should be freed as part + // of the arena destruction. It only exists to provide a matching operator + // delete to the operator new so that some compilers will not complain (see + // https://github.com/grpc/grpc/issues/11301) Note at the time of adding this + // there are no tests catching the compiler warning. + static void operator delete(void*, void*) { assert(0); } + + void MaybeFinish() { + if (--callbacks_outstanding_ == 0) { + reactor_->OnDone(std::move(finish_status_)); + auto* call = call_.call(); + this->~ClientCallbackReaderWriterImpl(); + g_core_codegen_interface->grpc_call_unref(call); + } + } + + void StartCall() override { + // This call initiates two batches + // 1. Send initial metadata (unless corked)/recv initial metadata + // 2. Recv trailing metadata, on_completion callback + callbacks_outstanding_ = 2; + + start_tag_.Set(call_.call(), + [this](bool ok) { + reactor_->OnReadInitialMetadataDone(ok); + MaybeFinish(); + }, + &start_ops_); + start_corked_ = context_->initial_metadata_corked_; + if (!start_corked_) { + start_ops_.SendInitialMetadata(&context_->send_initial_metadata_, + context_->initial_metadata_flags()); + } + start_ops_.RecvInitialMetadata(context_); + start_ops_.set_core_cq_tag(&start_tag_); + call_.PerformOps(&start_ops_); + + finish_tag_.Set(call_.call(), [this](bool ok) { MaybeFinish(); }, + &finish_ops_); + finish_ops_.ClientRecvStatus(context_, &finish_status_); + finish_ops_.set_core_cq_tag(&finish_tag_); + call_.PerformOps(&finish_ops_); + + // Also set up the read and write tags so that they don't have to be set up + // each time + write_tag_.Set(call_.call(), + [this](bool ok) { + reactor_->OnWriteDone(ok); + MaybeFinish(); + }, + &write_ops_); + write_ops_.set_core_cq_tag(&write_tag_); + + read_tag_.Set(call_.call(), + [this](bool ok) { + reactor_->OnReadDone(ok); + MaybeFinish(); + }, + &read_ops_); + read_ops_.set_core_cq_tag(&read_tag_); + } + + void Read(Response* msg) override { + read_ops_.RecvMessage(msg); + callbacks_outstanding_++; + call_.PerformOps(&read_ops_); + } + + void Write(const Request* msg, WriteOptions options) override { + if (start_corked_) { + write_ops_.SendInitialMetadata(&context_->send_initial_metadata_, + context_->initial_metadata_flags()); + start_corked_ = false; + } + // TODO(vjpai): don't assert + GPR_CODEGEN_ASSERT(write_ops_.SendMessage(*msg).ok()); + + if (options.is_last_message()) { + options.set_buffer_hint(); + write_ops_.ClientSendClose(); + } + callbacks_outstanding_++; + call_.PerformOps(&write_ops_); + } + void WritesDone() override { + if (start_corked_) { + writes_done_ops_.SendInitialMetadata(&context_->send_initial_metadata_, + context_->initial_metadata_flags()); + start_corked_ = false; + } + writes_done_ops_.ClientSendClose(); + writes_done_tag_.Set(call_.call(), + [this](bool ok) { + reactor_->OnWritesDoneDone(ok); + MaybeFinish(); + }, + &writes_done_ops_); + writes_done_ops_.set_core_cq_tag(&writes_done_tag_); + callbacks_outstanding_++; + call_.PerformOps(&writes_done_ops_); + } + + private: + friend class ClientCallbackReaderWriterFactory; + + ClientCallbackReaderWriterImpl( + Call call, ClientContext* context, + ::grpc::experimental::ClientBidiReactor* reactor) + : context_(context), call_(call), reactor_(reactor) {} + + ClientContext* context_; + Call call_; + ::grpc::experimental::ClientBidiReactor* reactor_; + + CallOpSet start_ops_; + CallbackWithSuccessTag start_tag_; + bool start_corked_; + + CallOpSet finish_ops_; + CallbackWithSuccessTag finish_tag_; + Status finish_status_; + + CallOpSet + write_ops_; + CallbackWithSuccessTag write_tag_; + + CallOpSet writes_done_ops_; + CallbackWithSuccessTag writes_done_tag_; + + CallOpSet> read_ops_; + CallbackWithSuccessTag read_tag_; + + std::atomic_int callbacks_outstanding_; +}; + +template +class ClientCallbackReaderWriterFactory { + public: + static experimental::ClientCallbackReaderWriter* Create( + ChannelInterface* channel, const ::grpc::internal::RpcMethod& method, + ClientContext* context, + ::grpc::experimental::ClientBidiReactor* reactor) { + Call call = channel->CreateCall(method, context, channel->CallbackCQ()); + + g_core_codegen_interface->grpc_call_ref(call.call()); + return new (g_core_codegen_interface->grpc_call_arena_alloc( + call.call(), sizeof(ClientCallbackReaderWriterImpl))) + ClientCallbackReaderWriterImpl(call, context, + reactor); + } +}; + +template +class ClientCallbackReaderImpl + : public ::grpc::experimental::ClientCallbackReader { + public: + // always allocated against a call arena, no memory free required + static void operator delete(void* ptr, std::size_t size) { + assert(size == sizeof(ClientCallbackReaderImpl)); + } + + // This operator should never be called as the memory should be freed as part + // of the arena destruction. It only exists to provide a matching operator + // delete to the operator new so that some compilers will not complain (see + // https://github.com/grpc/grpc/issues/11301) Note at the time of adding this + // there are no tests catching the compiler warning. + static void operator delete(void*, void*) { assert(0); } + + void MaybeFinish() { + if (--callbacks_outstanding_ == 0) { + reactor_->OnDone(std::move(finish_status_)); + auto* call = call_.call(); + this->~ClientCallbackReaderImpl(); + g_core_codegen_interface->grpc_call_unref(call); + } + } + + void StartCall() override { + // This call initiates two batches + // 1. Send initial metadata (unless corked)/recv initial metadata + // 2. Recv trailing metadata, on_completion callback + callbacks_outstanding_ = 2; + + start_tag_.Set(call_.call(), + [this](bool ok) { + reactor_->OnReadInitialMetadataDone(ok); + MaybeFinish(); + }, + &start_ops_); + start_ops_.SendInitialMetadata(&context_->send_initial_metadata_, + context_->initial_metadata_flags()); + start_ops_.RecvInitialMetadata(context_); + start_ops_.set_core_cq_tag(&start_tag_); + call_.PerformOps(&start_ops_); + + finish_tag_.Set(call_.call(), [this](bool ok) { MaybeFinish(); }, + &finish_ops_); + finish_ops_.ClientRecvStatus(context_, &finish_status_); + finish_ops_.set_core_cq_tag(&finish_tag_); + call_.PerformOps(&finish_ops_); + + // Also set up the read tag so it doesn't have to be set up each time + read_tag_.Set(call_.call(), + [this](bool ok) { + reactor_->OnReadDone(ok); + MaybeFinish(); + }, + &read_ops_); + read_ops_.set_core_cq_tag(&read_tag_); + } + + void Read(Response* msg) override { + read_ops_.RecvMessage(msg); + callbacks_outstanding_++; + call_.PerformOps(&read_ops_); + } + + private: + friend class ClientCallbackReaderFactory; + + template + ClientCallbackReaderImpl(Call call, ClientContext* context, Request* request, + ::grpc::experimental::ClientReadReactor* reactor) + : context_(context), call_(call), reactor_(reactor) { + // TODO(vjpai): don't assert + GPR_CODEGEN_ASSERT(start_ops_.SendMessage(*request).ok()); + start_ops_.ClientSendClose(); + } + + ClientContext* context_; + Call call_; + ::grpc::experimental::ClientReadReactor* reactor_; + + CallOpSet + start_ops_; + CallbackWithSuccessTag start_tag_; + + CallOpSet finish_ops_; + CallbackWithSuccessTag finish_tag_; + Status finish_status_; + + CallOpSet> read_ops_; + CallbackWithSuccessTag read_tag_; + + std::atomic_int callbacks_outstanding_; +}; + +template +class ClientCallbackReaderFactory { + public: + template + static experimental::ClientCallbackReader* Create( + ChannelInterface* channel, const ::grpc::internal::RpcMethod& method, + ClientContext* context, const Request* request, + ::grpc::experimental::ClientReadReactor* reactor) { + Call call = channel->CreateCall(method, context, channel->CallbackCQ()); + + g_core_codegen_interface->grpc_call_ref(call.call()); + return new (g_core_codegen_interface->grpc_call_arena_alloc( + call.call(), sizeof(ClientCallbackReaderImpl))) + ClientCallbackReaderImpl(call, context, request, reactor); + } +}; + +template +class ClientCallbackWriterImpl + : public ::grpc::experimental::ClientCallbackWriter { + public: + // always allocated against a call arena, no memory free required + static void operator delete(void* ptr, std::size_t size) { + assert(size == sizeof(ClientCallbackWriterImpl)); + } + + // This operator should never be called as the memory should be freed as part + // of the arena destruction. It only exists to provide a matching operator + // delete to the operator new so that some compilers will not complain (see + // https://github.com/grpc/grpc/issues/11301) Note at the time of adding this + // there are no tests catching the compiler warning. + static void operator delete(void*, void*) { assert(0); } + + void MaybeFinish() { + if (--callbacks_outstanding_ == 0) { + reactor_->OnDone(std::move(finish_status_)); + auto* call = call_.call(); + this->~ClientCallbackWriterImpl(); + g_core_codegen_interface->grpc_call_unref(call); + } + } + + void StartCall() override { + // This call initiates two batches + // 1. Send initial metadata (unless corked)/recv initial metadata + // 2. Recv message + trailing metadata, on_completion callback + callbacks_outstanding_ = 2; + + start_tag_.Set(call_.call(), + [this](bool ok) { + reactor_->OnReadInitialMetadataDone(ok); + MaybeFinish(); + }, + &start_ops_); + start_corked_ = context_->initial_metadata_corked_; + if (!start_corked_) { + start_ops_.SendInitialMetadata(&context_->send_initial_metadata_, + context_->initial_metadata_flags()); + } + start_ops_.RecvInitialMetadata(context_); + start_ops_.set_core_cq_tag(&start_tag_); + call_.PerformOps(&start_ops_); + + finish_tag_.Set(call_.call(), [this](bool ok) { MaybeFinish(); }, + &finish_ops_); + finish_ops_.ClientRecvStatus(context_, &finish_status_); + finish_ops_.set_core_cq_tag(&finish_tag_); + call_.PerformOps(&finish_ops_); + + // Also set up the read and write tags so that they don't have to be set up + // each time + write_tag_.Set(call_.call(), + [this](bool ok) { + reactor_->OnWriteDone(ok); + MaybeFinish(); + }, + &write_ops_); + write_ops_.set_core_cq_tag(&write_tag_); + } + + void Write(const Request* msg, WriteOptions options) override { + if (start_corked_) { + write_ops_.SendInitialMetadata(&context_->send_initial_metadata_, + context_->initial_metadata_flags()); + start_corked_ = false; + } + // TODO(vjpai): don't assert + GPR_CODEGEN_ASSERT(write_ops_.SendMessage(*msg).ok()); + + if (options.is_last_message()) { + options.set_buffer_hint(); + write_ops_.ClientSendClose(); + } + callbacks_outstanding_++; + call_.PerformOps(&write_ops_); + } + void WritesDone() override { + if (start_corked_) { + writes_done_ops_.SendInitialMetadata(&context_->send_initial_metadata_, + context_->initial_metadata_flags()); + start_corked_ = false; + } + writes_done_ops_.ClientSendClose(); + writes_done_tag_.Set(call_.call(), + [this](bool ok) { + reactor_->OnWritesDoneDone(ok); + MaybeFinish(); + }, + &writes_done_ops_); + writes_done_ops_.set_core_cq_tag(&writes_done_tag_); + callbacks_outstanding_++; + call_.PerformOps(&writes_done_ops_); + } + + private: + friend class ClientCallbackWriterFactory; + + template + ClientCallbackWriterImpl(Call call, ClientContext* context, + Response* response, + ::grpc::experimental::ClientWriteReactor* reactor) + : context_(context), call_(call), reactor_(reactor) { + finish_ops_.RecvMessage(response); + finish_ops_.AllowNoMessage(); + } + + ClientContext* context_; + Call call_; + ::grpc::experimental::ClientWriteReactor* reactor_; + + CallOpSet start_ops_; + CallbackWithSuccessTag start_tag_; + bool start_corked_; + + CallOpSet finish_ops_; + CallbackWithSuccessTag finish_tag_; + Status finish_status_; + + CallOpSet + write_ops_; + CallbackWithSuccessTag write_tag_; + + CallOpSet writes_done_ops_; + CallbackWithSuccessTag writes_done_tag_; + + std::atomic_int callbacks_outstanding_; +}; + +template +class ClientCallbackWriterFactory { + public: + template + static experimental::ClientCallbackWriter* Create( + ChannelInterface* channel, const ::grpc::internal::RpcMethod& method, + ClientContext* context, Response* response, + ::grpc::experimental::ClientWriteReactor* reactor) { + Call call = channel->CreateCall(method, context, channel->CallbackCQ()); + + g_core_codegen_interface->grpc_call_ref(call.call()); + return new (g_core_codegen_interface->grpc_call_arena_alloc( + call.call(), sizeof(ClientCallbackWriterImpl))) + ClientCallbackWriterImpl(call, context, response, reactor); + } +}; } // namespace internal } // namespace grpc diff --git a/include/grpcpp/impl/codegen/client_context.h b/include/grpcpp/impl/codegen/client_context.h index 75b955e760..6059c3c58a 100644 --- a/include/grpcpp/impl/codegen/client_context.h +++ b/include/grpcpp/impl/codegen/client_context.h @@ -71,6 +71,12 @@ template class BlockingUnaryCallImpl; template class CallbackUnaryCallImpl; +template +class ClientCallbackReaderWriterImpl; +template +class ClientCallbackReaderImpl; +template +class ClientCallbackWriterImpl; } // namespace internal template @@ -394,6 +400,12 @@ class ClientContext { friend class ::grpc::internal::BlockingUnaryCallImpl; template friend class ::grpc::internal::CallbackUnaryCallImpl; + template + friend class ::grpc::internal::ClientCallbackReaderWriterImpl; + template + friend class ::grpc::internal::ClientCallbackReaderImpl; + template + friend class ::grpc::internal::ClientCallbackWriterImpl; // Used by friend class CallOpClientRecvStatus void set_debug_error_string(const grpc::string& debug_error_string) { diff --git a/src/compiler/cpp_generator.cc b/src/compiler/cpp_generator.cc index 7986aca696..473e57166f 100644 --- a/src/compiler/cpp_generator.cc +++ b/src/compiler/cpp_generator.cc @@ -132,6 +132,7 @@ grpc::string GetHeaderIncludes(grpc_generator::File* file, "grpcpp/impl/codegen/async_generic_service.h", "grpcpp/impl/codegen/async_stream.h", "grpcpp/impl/codegen/async_unary_call.h", + "grpcpp/impl/codegen/client_callback.h", "grpcpp/impl/codegen/method_handler_impl.h", "grpcpp/impl/codegen/proto_utils.h", "grpcpp/impl/codegen/rpc_method.h", @@ -580,11 +581,23 @@ void PrintHeaderClientMethodCallbackInterfaces( "const $Request$* request, $Response$* response, " "std::function) = 0;\n"); } else if (ClientOnlyStreaming(method)) { - // TODO(vjpai): Add support for client-side streaming + printer->Print(*vars, + "virtual ::grpc::experimental::ClientCallbackWriter< " + "$Request$>* $Method$(::grpc::ClientContext* context, " + "$Response$* response, " + "::grpc::experimental::ClientWriteReactor* reactor) = 0;\n"); } else if (ServerOnlyStreaming(method)) { - // TODO(vjpai): Add support for server-side streaming + printer->Print(*vars, + "virtual ::grpc::experimental::ClientCallbackReader< " + "$Response$>* $Method$(::grpc::ClientContext* context, " + "$Request$* request, " + "::grpc::experimental::ClientReadReactor* reactor) = 0;\n"); } else if (method->BidiStreaming()) { - // TODO(vjpai): Add support for bidi streaming + printer->Print( + *vars, + "virtual ::grpc::experimental::ClientCallbackReaderWriter< $Request$, " + "$Response$>* $Method$(::grpc::ClientContext* context, " + "::grpc::experimental::ClientBidiReactor* reactor) = 0;\n"); } } @@ -631,11 +644,26 @@ void PrintHeaderClientMethodCallback(grpc_generator::Printer* printer, "const $Request$* request, $Response$* response, " "std::function) override;\n"); } else if (ClientOnlyStreaming(method)) { - // TODO(vjpai): Add support for client-side streaming + printer->Print( + *vars, + "::grpc::experimental::ClientCallbackWriter< $Request$>* " + "$Method$(::grpc::ClientContext* context, " + "$Response$* response, " + "::grpc::experimental::ClientWriteReactor* reactor) override;\n"); } else if (ServerOnlyStreaming(method)) { - // TODO(vjpai): Add support for server-side streaming + printer->Print( + *vars, + "::grpc::experimental::ClientCallbackReader< $Response$>* " + "$Method$(::grpc::ClientContext* context, " + "$Request$* request, " + "::grpc::experimental::ClientReadReactor* reactor) override;\n"); + } else if (method->BidiStreaming()) { - // TODO(vjpai): Add support for bidi streaming + printer->Print( + *vars, + "::grpc::experimental::ClientCallbackReaderWriter< $Request$, " + "$Response$>* $Method$(::grpc::ClientContext* context, " + "::grpc::experimental::ClientBidiReactor* reactor) override;\n"); } } @@ -1607,7 +1635,20 @@ void PrintSourceClientMethod(grpc_generator::Printer* printer, "context, response);\n" "}\n\n"); - // TODO(vjpai): Add callback version + printer->Print( + *vars, + "::grpc::experimental::ClientCallbackWriter< $Request$>* " + "$ns$$Service$::" + "Stub::experimental_async::$Method$(::grpc::ClientContext* context, " + "$Response$* response, " + "::grpc::experimental::ClientWriteReactor* reactor) {\n"); + printer->Print(*vars, + " return ::grpc::internal::ClientCallbackWriterFactory< " + "$Request$>::Create(" + "stub_->channel_.get(), " + "stub_->rpcmethod_$Method$_, " + "context, response, reactor);\n" + "}\n\n"); for (auto async_prefix : async_prefixes) { (*vars)["AsyncPrefix"] = async_prefix.prefix; @@ -1641,7 +1682,19 @@ void PrintSourceClientMethod(grpc_generator::Printer* printer, "context, request);\n" "}\n\n"); - // TODO(vjpai): Add callback version + printer->Print(*vars, + "::grpc::experimental::ClientCallbackReader< $Response$>* " + "$ns$$Service$::Stub::experimental_async::$Method$(::grpc::" + "ClientContext* context, " + "$Request$* request, " + "::grpc::experimental::ClientReadReactor* reactor) {\n"); + printer->Print(*vars, + " return ::grpc::internal::ClientCallbackReaderFactory< " + "$Response$>::Create(" + "stub_->channel_.get(), " + "stub_->rpcmethod_$Method$_, " + "context, request, reactor);\n" + "}\n\n"); for (auto async_prefix : async_prefixes) { (*vars)["AsyncPrefix"] = async_prefix.prefix; @@ -1675,7 +1728,20 @@ void PrintSourceClientMethod(grpc_generator::Printer* printer, "context);\n" "}\n\n"); - // TODO(vjpai): Add callback version + printer->Print(*vars, + "::grpc::experimental::ClientCallbackReaderWriter< " + "$Request$,$Response$>* " + "$ns$$Service$::Stub::experimental_async::$Method$(::grpc::" + "ClientContext* context, " + "::grpc::experimental::ClientBidiReactor* reactor) {\n"); + printer->Print( + *vars, + " return ::grpc::internal::ClientCallbackReaderWriterFactory< " + "$Request$,$Response$>::Create(" + "stub_->channel_.get(), " + "stub_->rpcmethod_$Method$_, " + "context, reactor);\n" + "}\n\n"); for (auto async_prefix : async_prefixes) { (*vars)["AsyncPrefix"] = async_prefix.prefix; diff --git a/src/cpp/client/generic_stub.cc b/src/cpp/client/generic_stub.cc index 87902b26f0..f029daec65 100644 --- a/src/cpp/client/generic_stub.cc +++ b/src/cpp/client/generic_stub.cc @@ -72,4 +72,16 @@ void GenericStub::experimental_type::UnaryCall( context, request, response, std::move(on_completion)); } +experimental::ClientCallbackReaderWriter* +GenericStub::experimental_type::PrepareBidiStreamingCall( + ClientContext* context, const grpc::string& method, + experimental::ClientBidiReactor* reactor) { + return internal::ClientCallbackReaderWriterFactory< + ByteBuffer, ByteBuffer>::Create(stub_->channel_.get(), + internal::RpcMethod( + method.c_str(), + internal::RpcMethod::BIDI_STREAMING), + context, reactor); +} + } // namespace grpc diff --git a/test/cpp/codegen/compiler_test_golden b/test/cpp/codegen/compiler_test_golden index fdc67969d9..7a25f51d10 100644 --- a/test/cpp/codegen/compiler_test_golden +++ b/test/cpp/codegen/compiler_test_golden @@ -30,6 +30,7 @@ #include #include #include +#include #include #include #include @@ -117,10 +118,13 @@ class ServiceA final { // // Method A2 leading comment 1 // Method A2 leading comment 2 + virtual ::grpc::experimental::ClientCallbackWriter< ::grpc::testing::Request>* MethodA2(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::experimental::ClientWriteReactor* reactor) = 0; // MethodA2 trailing comment 1 // Method A3 leading comment 1 + virtual ::grpc::experimental::ClientCallbackReader< ::grpc::testing::Response>* MethodA3(::grpc::ClientContext* context, ::grpc::testing::Request* request, ::grpc::experimental::ClientReadReactor* reactor) = 0; // Method A3 trailing comment 1 // Method A4 leading comment 1 + virtual ::grpc::experimental::ClientCallbackReaderWriter< ::grpc::testing::Request, ::grpc::testing::Response>* MethodA4(::grpc::ClientContext* context, ::grpc::experimental::ClientBidiReactor* reactor) = 0; // Method A4 trailing comment 1 }; virtual class experimental_async_interface* experimental_async() { return nullptr; } @@ -178,6 +182,9 @@ class ServiceA final { public StubInterface::experimental_async_interface { public: void MethodA1(::grpc::ClientContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response, std::function) override; + ::grpc::experimental::ClientCallbackWriter< ::grpc::testing::Request>* MethodA2(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::experimental::ClientWriteReactor* reactor) override; + ::grpc::experimental::ClientCallbackReader< ::grpc::testing::Response>* MethodA3(::grpc::ClientContext* context, ::grpc::testing::Request* request, ::grpc::experimental::ClientReadReactor* reactor) override; + ::grpc::experimental::ClientCallbackReaderWriter< ::grpc::testing::Request, ::grpc::testing::Response>* MethodA4(::grpc::ClientContext* context, ::grpc::experimental::ClientBidiReactor* reactor) override; private: friend class Stub; explicit experimental_async(Stub* stub): stub_(stub) { } diff --git a/test/cpp/end2end/client_callback_end2end_test.cc b/test/cpp/end2end/client_callback_end2end_test.cc index a35991396a..e39fc5aab2 100644 --- a/test/cpp/end2end/client_callback_end2end_test.cc +++ b/test/cpp/end2end/client_callback_end2end_test.cc @@ -182,6 +182,59 @@ class ClientCallbackEnd2endTest } } + void SendGenericEchoAsBidi(int num_rpcs) { + const grpc::string kMethodName("/grpc.testing.EchoTestService/Echo"); + grpc::string test_string(""); + for (int i = 0; i < num_rpcs; i++) { + test_string += "Hello world. "; + class Client : public grpc::experimental::ClientBidiReactor { + public: + Client(ClientCallbackEnd2endTest* test, const grpc::string& method_name, + const grpc::string& test_str) { + stream_ = + test->generic_stub_->experimental().PrepareBidiStreamingCall( + &cli_ctx_, method_name, this); + stream_->StartCall(); + request_.set_message(test_str); + send_buf_ = SerializeToByteBuffer(&request_); + stream_->Read(&recv_buf_); + stream_->Write(send_buf_.get()); + } + void OnWriteDone(bool ok) override { stream_->WritesDone(); } + void OnReadDone(bool ok) override { + EchoResponse response; + EXPECT_TRUE(ParseFromByteBuffer(&recv_buf_, &response)); + EXPECT_EQ(request_.message(), response.message()); + }; + void OnDone(Status s) override { + // The stream is invalid once OnDone is called + stream_ = nullptr; + EXPECT_TRUE(s.ok()); + std::unique_lock l(mu_); + done_ = true; + cv_.notify_one(); + } + void Await() { + std::unique_lock l(mu_); + while (!done_) { + cv_.wait(l); + } + } + + EchoRequest request_; + std::unique_ptr send_buf_; + ByteBuffer recv_buf_; + ClientContext cli_ctx_; + experimental::ClientCallbackReaderWriter* + stream_; + std::mutex mu_; + std::condition_variable cv_; + bool done_ = false; + } rpc{this, kMethodName, test_string}; + + rpc.Await(); + } + } bool is_server_started_; std::shared_ptr channel_; std::unique_ptr stub_; @@ -211,6 +264,11 @@ TEST_P(ClientCallbackEnd2endTest, SequentialGenericRpcs) { SendRpcsGeneric(10, false); } +TEST_P(ClientCallbackEnd2endTest, SequentialGenericRpcsAsBidi) { + ResetStub(); + SendGenericEchoAsBidi(10); +} + #if GRPC_ALLOW_EXCEPTIONS TEST_P(ClientCallbackEnd2endTest, ExceptingRpc) { ResetStub(); @@ -267,6 +325,176 @@ TEST_P(ClientCallbackEnd2endTest, CancelRpcBeforeStart) { } } +TEST_P(ClientCallbackEnd2endTest, RequestStream) { + // TODO(vjpai): test with callback server once supported + if (GetParam().callback_server) { + return; + } + + ResetStub(); + class Client : public grpc::experimental::ClientWriteReactor { + public: + explicit Client(grpc::testing::EchoTestService::Stub* stub) { + context_.set_initial_metadata_corked(true); + stream_ = stub->experimental_async()->RequestStream(&context_, &response_, + this); + stream_->StartCall(); + request_.set_message("Hello server."); + stream_->Write(&request_); + } + void OnWriteDone(bool ok) override { + writes_left_--; + if (writes_left_ > 1) { + stream_->Write(&request_); + } else if (writes_left_ == 1) { + stream_->WriteLast(&request_, WriteOptions()); + } + } + void OnDone(Status s) override { + stream_ = nullptr; + EXPECT_TRUE(s.ok()); + EXPECT_EQ(response_.message(), "Hello server.Hello server.Hello server."); + std::unique_lock l(mu_); + done_ = true; + cv_.notify_one(); + } + void Await() { + std::unique_lock l(mu_); + while (!done_) { + cv_.wait(l); + } + } + + private: + ::grpc::experimental::ClientCallbackWriter* stream_; + EchoRequest request_; + EchoResponse response_; + ClientContext context_; + int writes_left_{3}; + std::mutex mu_; + std::condition_variable cv_; + bool done_ = false; + } test{stub_.get()}; + + test.Await(); +} + +TEST_P(ClientCallbackEnd2endTest, ResponseStream) { + // TODO(vjpai): test with callback server once supported + if (GetParam().callback_server) { + return; + } + + ResetStub(); + class Client : public grpc::experimental::ClientReadReactor { + public: + explicit Client(grpc::testing::EchoTestService::Stub* stub) { + request_.set_message("Hello client "); + stream_ = stub->experimental_async()->ResponseStream(&context_, &request_, + this); + stream_->StartCall(); + stream_->Read(&response_); + } + void OnReadDone(bool ok) override { + // Note that != is the boolean XOR operator + EXPECT_NE(ok, reads_complete_ == kServerDefaultResponseStreamsToSend); + if (ok) { + EXPECT_EQ(response_.message(), + request_.message() + grpc::to_string(reads_complete_)); + reads_complete_++; + stream_->Read(&response_); + } + } + void OnDone(Status s) override { + stream_ = nullptr; + EXPECT_TRUE(s.ok()); + std::unique_lock l(mu_); + done_ = true; + cv_.notify_one(); + } + void Await() { + std::unique_lock l(mu_); + while (!done_) { + cv_.wait(l); + } + } + + private: + ::grpc::experimental::ClientCallbackReader* stream_; + EchoRequest request_; + EchoResponse response_; + ClientContext context_; + int reads_complete_{0}; + std::mutex mu_; + std::condition_variable cv_; + bool done_ = false; + } test{stub_.get()}; + + test.Await(); +} + +TEST_P(ClientCallbackEnd2endTest, BidiStream) { + // TODO(vjpai): test with callback server once supported + if (GetParam().callback_server) { + return; + } + ResetStub(); + class Client : public grpc::experimental::ClientBidiReactor { + public: + explicit Client(grpc::testing::EchoTestService::Stub* stub) { + request_.set_message("Hello fren "); + stream_ = stub->experimental_async()->BidiStream(&context_, this); + stream_->StartCall(); + stream_->Read(&response_); + stream_->Write(&request_); + } + void OnReadDone(bool ok) override { + // Note that != is the boolean XOR operator + EXPECT_NE(ok, reads_complete_ == kServerDefaultResponseStreamsToSend); + if (ok) { + EXPECT_EQ(response_.message(), request_.message()); + reads_complete_++; + stream_->Read(&response_); + } + } + void OnWriteDone(bool ok) override { + EXPECT_TRUE(ok); + if (++writes_complete_ == kServerDefaultResponseStreamsToSend) { + stream_->WritesDone(); + } else { + stream_->Write(&request_); + } + } + void OnDone(Status s) override { + stream_ = nullptr; + EXPECT_TRUE(s.ok()); + std::unique_lock l(mu_); + done_ = true; + cv_.notify_one(); + } + void Await() { + std::unique_lock l(mu_); + while (!done_) { + cv_.wait(l); + } + } + + private: + ::grpc::experimental::ClientCallbackReaderWriter* + stream_; + EchoRequest request_; + EchoResponse response_; + ClientContext context_; + int reads_complete_{0}; + int writes_complete_{0}; + std::mutex mu_; + std::condition_variable cv_; + bool done_ = false; + } test{stub_.get()}; + + test.Await(); +} + TestScenario scenarios[] = {TestScenario{false}, TestScenario{true}}; INSTANTIATE_TEST_CASE_P(ClientCallbackEnd2endTest, ClientCallbackEnd2endTest, -- cgit v1.2.3 From ea1156da3fb3d6fb8660d078e70cf5486fc71a65 Mon Sep 17 00:00:00 2001 From: Vijay Pai Date: Fri, 30 Nov 2018 02:16:08 -0800 Subject: Stop exposing streaming object class --- include/grpcpp/generic/generic_stub.h | 6 +- include/grpcpp/impl/codegen/client_callback.h | 302 +++--- src/compiler/cpp_generator.cc | 96 +- src/cpp/client/generic_stub.cc | 15 +- test/cpp/codegen/compiler_test_golden | 1105 +++++++++++++++------- test/cpp/end2end/client_callback_end2end_test.cc | 63 +- 6 files changed, 1053 insertions(+), 534 deletions(-) (limited to 'src/compiler') diff --git a/include/grpcpp/generic/generic_stub.h b/include/grpcpp/generic/generic_stub.h index ccbf8a0e55..eb014184e4 100644 --- a/include/grpcpp/generic/generic_stub.h +++ b/include/grpcpp/generic/generic_stub.h @@ -77,9 +77,9 @@ class GenericStub final { const ByteBuffer* request, ByteBuffer* response, std::function on_completion); - experimental::ClientCallbackReaderWriter* - PrepareBidiStreamingCall(ClientContext* context, const grpc::string& method, - experimental::ClientBidiReactor* reactor); + void PrepareBidiStreamingCall( + ClientContext* context, const grpc::string& method, + experimental::ClientBidiReactor* reactor); private: GenericStub* stub_; diff --git a/include/grpcpp/impl/codegen/client_callback.h b/include/grpcpp/impl/codegen/client_callback.h index 92a588b3c3..999c1c8a3e 100644 --- a/include/grpcpp/impl/codegen/client_callback.h +++ b/include/grpcpp/impl/codegen/client_callback.h @@ -93,9 +93,67 @@ class CallbackUnaryCallImpl { namespace experimental { +// Forward declarations +template +class ClientBidiReactor; +template +class ClientReadReactor; +template +class ClientWriteReactor; + +// NOTE: The streaming objects are not actually implemented in the public API. +// These interfaces are provided for mocking only. Typical applications +// will interact exclusively with the reactors that they define. +template +class ClientCallbackReaderWriter { + public: + virtual ~ClientCallbackReaderWriter() {} + virtual void StartCall() = 0; + virtual void Write(const Request* req, WriteOptions options) = 0; + virtual void WritesDone() = 0; + virtual void Read(Response* resp) = 0; + + protected: + void BindReactor(ClientBidiReactor* reactor) { + reactor->BindStream(this); + } +}; + +template +class ClientCallbackReader { + public: + virtual ~ClientCallbackReader() {} + virtual void StartCall() = 0; + virtual void Read(Response* resp) = 0; + + protected: + void BindReactor(ClientReadReactor* reactor) { + reactor->BindReader(this); + } +}; + +template +class ClientCallbackWriter { + public: + virtual ~ClientCallbackWriter() {} + virtual void StartCall() = 0; + void Write(const Request* req) { Write(req, WriteOptions()); } + virtual void Write(const Request* req, WriteOptions options) = 0; + void WriteLast(const Request* req, WriteOptions options) { + Write(req, options.set_last_message()); + } + virtual void WritesDone() = 0; + + protected: + void BindReactor(ClientWriteReactor* reactor) { + reactor->BindWriter(this); + } +}; + // The user must implement this reactor interface with reactions to each event // type that gets called by the library. An empty reaction is provided by // default +template class ClientBidiReactor { public: virtual ~ClientBidiReactor() {} @@ -104,16 +162,44 @@ class ClientBidiReactor { virtual void OnReadDone(bool ok) {} virtual void OnWriteDone(bool ok) {} virtual void OnWritesDoneDone(bool ok) {} + + void StartCall() { stream_->StartCall(); } + void StartRead(Response* resp) { stream_->Read(resp); } + void StartWrite(const Request* req) { StartWrite(req, WriteOptions()); } + void StartWrite(const Request* req, WriteOptions options) { + stream_->Write(req, std::move(options)); + } + void StartWriteLast(const Request* req, WriteOptions options) { + StartWrite(req, std::move(options.set_last_message())); + } + void StartWritesDone() { stream_->WritesDone(); } + + private: + friend class ClientCallbackReaderWriter; + void BindStream(ClientCallbackReaderWriter* stream) { + stream_ = stream; + } + ClientCallbackReaderWriter* stream_; }; +template class ClientReadReactor { public: virtual ~ClientReadReactor() {} virtual void OnDone(Status s) {} virtual void OnReadInitialMetadataDone(bool ok) {} virtual void OnReadDone(bool ok) {} + + void StartCall() { reader_->StartCall(); } + void StartRead(Response* resp) { reader_->Read(resp); } + + private: + friend class ClientCallbackReader; + void BindReader(ClientCallbackReader* reader) { reader_ = reader; } + ClientCallbackReader* reader_; }; +template class ClientWriteReactor { public: virtual ~ClientWriteReactor() {} @@ -121,41 +207,21 @@ class ClientWriteReactor { virtual void OnReadInitialMetadataDone(bool ok) {} virtual void OnWriteDone(bool ok) {} virtual void OnWritesDoneDone(bool ok) {} -}; -template -class ClientCallbackReaderWriter { - public: - virtual ~ClientCallbackReaderWriter() {} - virtual void StartCall() = 0; - void Write(const Request* req) { Write(req, WriteOptions()); } - virtual void Write(const Request* req, WriteOptions options) = 0; - void WriteLast(const Request* req, WriteOptions options) { - Write(req, options.set_last_message()); + void StartCall() { writer_->StartCall(); } + void StartWrite(const Request* req) { StartWrite(req, WriteOptions()); } + void StartWrite(const Request* req, WriteOptions options) { + writer_->Write(req, std::move(options)); } - virtual void WritesDone() = 0; - virtual void Read(Response* resp) = 0; -}; - -template -class ClientCallbackReader { - public: - virtual ~ClientCallbackReader() {} - virtual void StartCall() = 0; - virtual void Read(Response* resp) = 0; -}; - -template -class ClientCallbackWriter { - public: - virtual ~ClientCallbackWriter() {} - virtual void StartCall() = 0; - void Write(const Request* req) { Write(req, WriteOptions()); } - virtual void Write(const Request* req, WriteOptions options) = 0; - void WriteLast(const Request* req, WriteOptions options) { - Write(req, options.set_last_message()); + void StartWriteLast(const Request* req, WriteOptions options) { + StartWrite(req, std::move(options.set_last_message())); } - virtual void WritesDone() = 0; + void StartWritesDone() { writer_->WritesDone(); } + + private: + friend class ClientCallbackWriter; + void BindWriter(ClientCallbackWriter* writer) { writer_ = writer; } + ClientCallbackWriter* writer_; }; } // namespace experimental @@ -204,12 +270,13 @@ class ClientCallbackReaderWriterImpl // 4. Any write backlog started_ = true; - start_tag_.Set(call_.call(), - [this](bool ok) { - reactor_->OnReadInitialMetadataDone(ok); - MaybeFinish(); - }, - &start_ops_); + start_tag_.Set( + call_.call(), + [this](bool ok) { + reactor_->OnReadInitialMetadataDone(ok); + MaybeFinish(); + }, + &start_ops_); if (!start_corked_) { start_ops_.SendInitialMetadata(&context_->send_initial_metadata_, context_->initial_metadata_flags()); @@ -220,27 +287,29 @@ class ClientCallbackReaderWriterImpl // Also set up the read and write tags so that they don't have to be set up // each time - write_tag_.Set(call_.call(), - [this](bool ok) { - reactor_->OnWriteDone(ok); - MaybeFinish(); - }, - &write_ops_); + write_tag_.Set( + call_.call(), + [this](bool ok) { + reactor_->OnWriteDone(ok); + MaybeFinish(); + }, + &write_ops_); write_ops_.set_core_cq_tag(&write_tag_); - read_tag_.Set(call_.call(), - [this](bool ok) { - reactor_->OnReadDone(ok); - MaybeFinish(); - }, - &read_ops_); + read_tag_.Set( + call_.call(), + [this](bool ok) { + reactor_->OnReadDone(ok); + MaybeFinish(); + }, + &read_ops_); read_ops_.set_core_cq_tag(&read_tag_); if (read_ops_at_start_) { call_.PerformOps(&read_ops_); } - finish_tag_.Set(call_.call(), [this](bool ok) { MaybeFinish(); }, - &finish_ops_); + finish_tag_.Set( + call_.call(), [this](bool ok) { MaybeFinish(); }, &finish_ops_); finish_ops_.ClientRecvStatus(context_, &finish_status_); finish_ops_.set_core_cq_tag(&finish_tag_); call_.PerformOps(&finish_ops_); @@ -291,12 +360,13 @@ class ClientCallbackReaderWriterImpl start_corked_ = false; } writes_done_ops_.ClientSendClose(); - writes_done_tag_.Set(call_.call(), - [this](bool ok) { - reactor_->OnWritesDoneDone(ok); - MaybeFinish(); - }, - &writes_done_ops_); + writes_done_tag_.Set( + call_.call(), + [this](bool ok) { + reactor_->OnWritesDoneDone(ok); + MaybeFinish(); + }, + &writes_done_ops_); writes_done_ops_.set_core_cq_tag(&writes_done_tag_); callbacks_outstanding_++; if (started_) { @@ -311,15 +381,17 @@ class ClientCallbackReaderWriterImpl ClientCallbackReaderWriterImpl( Call call, ClientContext* context, - ::grpc::experimental::ClientBidiReactor* reactor) + ::grpc::experimental::ClientBidiReactor* reactor) : context_(context), call_(call), reactor_(reactor), - start_corked_(context_->initial_metadata_corked_) {} + start_corked_(context_->initial_metadata_corked_) { + this->BindReactor(reactor); + } ClientContext* context_; Call call_; - ::grpc::experimental::ClientBidiReactor* reactor_; + ::grpc::experimental::ClientBidiReactor* reactor_; CallOpSet start_ops_; CallbackWithSuccessTag start_tag_; @@ -350,14 +422,14 @@ class ClientCallbackReaderWriterImpl template class ClientCallbackReaderWriterFactory { public: - static experimental::ClientCallbackReaderWriter* Create( + static void Create( ChannelInterface* channel, const ::grpc::internal::RpcMethod& method, ClientContext* context, - ::grpc::experimental::ClientBidiReactor* reactor) { + ::grpc::experimental::ClientBidiReactor* reactor) { Call call = channel->CreateCall(method, context, channel->CallbackCQ()); g_core_codegen_interface->grpc_call_ref(call.call()); - return new (g_core_codegen_interface->grpc_call_arena_alloc( + new (g_core_codegen_interface->grpc_call_arena_alloc( call.call(), sizeof(ClientCallbackReaderWriterImpl))) ClientCallbackReaderWriterImpl(call, context, reactor); @@ -396,12 +468,13 @@ class ClientCallbackReaderImpl // 3. Recv trailing metadata, on_completion callback started_ = true; - start_tag_.Set(call_.call(), - [this](bool ok) { - reactor_->OnReadInitialMetadataDone(ok); - MaybeFinish(); - }, - &start_ops_); + start_tag_.Set( + call_.call(), + [this](bool ok) { + reactor_->OnReadInitialMetadataDone(ok); + MaybeFinish(); + }, + &start_ops_); start_ops_.SendInitialMetadata(&context_->send_initial_metadata_, context_->initial_metadata_flags()); start_ops_.RecvInitialMetadata(context_); @@ -409,19 +482,20 @@ class ClientCallbackReaderImpl call_.PerformOps(&start_ops_); // Also set up the read tag so it doesn't have to be set up each time - read_tag_.Set(call_.call(), - [this](bool ok) { - reactor_->OnReadDone(ok); - MaybeFinish(); - }, - &read_ops_); + read_tag_.Set( + call_.call(), + [this](bool ok) { + reactor_->OnReadDone(ok); + MaybeFinish(); + }, + &read_ops_); read_ops_.set_core_cq_tag(&read_tag_); if (read_ops_at_start_) { call_.PerformOps(&read_ops_); } - finish_tag_.Set(call_.call(), [this](bool ok) { MaybeFinish(); }, - &finish_ops_); + finish_tag_.Set( + call_.call(), [this](bool ok) { MaybeFinish(); }, &finish_ops_); finish_ops_.ClientRecvStatus(context_, &finish_status_); finish_ops_.set_core_cq_tag(&finish_tag_); call_.PerformOps(&finish_ops_); @@ -441,9 +515,11 @@ class ClientCallbackReaderImpl friend class ClientCallbackReaderFactory; template - ClientCallbackReaderImpl(Call call, ClientContext* context, Request* request, - ::grpc::experimental::ClientReadReactor* reactor) + ClientCallbackReaderImpl( + Call call, ClientContext* context, Request* request, + ::grpc::experimental::ClientReadReactor* reactor) : context_(context), call_(call), reactor_(reactor) { + this->BindReactor(reactor); // TODO(vjpai): don't assert GPR_CODEGEN_ASSERT(start_ops_.SendMessage(*request).ok()); start_ops_.ClientSendClose(); @@ -451,7 +527,7 @@ class ClientCallbackReaderImpl ClientContext* context_; Call call_; - ::grpc::experimental::ClientReadReactor* reactor_; + ::grpc::experimental::ClientReadReactor* reactor_; CallOpSet @@ -475,14 +551,14 @@ template class ClientCallbackReaderFactory { public: template - static experimental::ClientCallbackReader* Create( + static void Create( ChannelInterface* channel, const ::grpc::internal::RpcMethod& method, ClientContext* context, const Request* request, - ::grpc::experimental::ClientReadReactor* reactor) { + ::grpc::experimental::ClientReadReactor* reactor) { Call call = channel->CreateCall(method, context, channel->CallbackCQ()); g_core_codegen_interface->grpc_call_ref(call.call()); - return new (g_core_codegen_interface->grpc_call_arena_alloc( + new (g_core_codegen_interface->grpc_call_arena_alloc( call.call(), sizeof(ClientCallbackReaderImpl))) ClientCallbackReaderImpl(call, context, request, reactor); } @@ -520,12 +596,13 @@ class ClientCallbackWriterImpl // 3. Any backlog started_ = true; - start_tag_.Set(call_.call(), - [this](bool ok) { - reactor_->OnReadInitialMetadataDone(ok); - MaybeFinish(); - }, - &start_ops_); + start_tag_.Set( + call_.call(), + [this](bool ok) { + reactor_->OnReadInitialMetadataDone(ok); + MaybeFinish(); + }, + &start_ops_); if (!start_corked_) { start_ops_.SendInitialMetadata(&context_->send_initial_metadata_, context_->initial_metadata_flags()); @@ -536,16 +613,17 @@ class ClientCallbackWriterImpl // Also set up the read and write tags so that they don't have to be set up // each time - write_tag_.Set(call_.call(), - [this](bool ok) { - reactor_->OnWriteDone(ok); - MaybeFinish(); - }, - &write_ops_); + write_tag_.Set( + call_.call(), + [this](bool ok) { + reactor_->OnWriteDone(ok); + MaybeFinish(); + }, + &write_ops_); write_ops_.set_core_cq_tag(&write_tag_); - finish_tag_.Set(call_.call(), [this](bool ok) { MaybeFinish(); }, - &finish_ops_); + finish_tag_.Set( + call_.call(), [this](bool ok) { MaybeFinish(); }, &finish_ops_); finish_ops_.ClientRecvStatus(context_, &finish_status_); finish_ops_.set_core_cq_tag(&finish_tag_); call_.PerformOps(&finish_ops_); @@ -586,12 +664,13 @@ class ClientCallbackWriterImpl start_corked_ = false; } writes_done_ops_.ClientSendClose(); - writes_done_tag_.Set(call_.call(), - [this](bool ok) { - reactor_->OnWritesDoneDone(ok); - MaybeFinish(); - }, - &writes_done_ops_); + writes_done_tag_.Set( + call_.call(), + [this](bool ok) { + reactor_->OnWritesDoneDone(ok); + MaybeFinish(); + }, + &writes_done_ops_); writes_done_ops_.set_core_cq_tag(&writes_done_tag_); callbacks_outstanding_++; if (started_) { @@ -605,20 +684,21 @@ class ClientCallbackWriterImpl friend class ClientCallbackWriterFactory; template - ClientCallbackWriterImpl(Call call, ClientContext* context, - Response* response, - ::grpc::experimental::ClientWriteReactor* reactor) + ClientCallbackWriterImpl( + Call call, ClientContext* context, Response* response, + ::grpc::experimental::ClientWriteReactor* reactor) : context_(context), call_(call), reactor_(reactor), start_corked_(context_->initial_metadata_corked_) { + this->BindReactor(reactor); finish_ops_.RecvMessage(response); finish_ops_.AllowNoMessage(); } ClientContext* context_; Call call_; - ::grpc::experimental::ClientWriteReactor* reactor_; + ::grpc::experimental::ClientWriteReactor* reactor_; CallOpSet start_ops_; CallbackWithSuccessTag start_tag_; @@ -646,14 +726,14 @@ template class ClientCallbackWriterFactory { public: template - static experimental::ClientCallbackWriter* Create( + static void Create( ChannelInterface* channel, const ::grpc::internal::RpcMethod& method, ClientContext* context, Response* response, - ::grpc::experimental::ClientWriteReactor* reactor) { + ::grpc::experimental::ClientWriteReactor* reactor) { Call call = channel->CreateCall(method, context, channel->CallbackCQ()); g_core_codegen_interface->grpc_call_ref(call.call()); - return new (g_core_codegen_interface->grpc_call_arena_alloc( + new (g_core_codegen_interface->grpc_call_arena_alloc( call.call(), sizeof(ClientCallbackWriterImpl))) ClientCallbackWriterImpl(call, context, response, reactor); } diff --git a/src/compiler/cpp_generator.cc b/src/compiler/cpp_generator.cc index 473e57166f..a368b47f01 100644 --- a/src/compiler/cpp_generator.cc +++ b/src/compiler/cpp_generator.cc @@ -582,22 +582,21 @@ void PrintHeaderClientMethodCallbackInterfaces( "std::function) = 0;\n"); } else if (ClientOnlyStreaming(method)) { printer->Print(*vars, - "virtual ::grpc::experimental::ClientCallbackWriter< " - "$Request$>* $Method$(::grpc::ClientContext* context, " + "virtual void $Method$(::grpc::ClientContext* context, " "$Response$* response, " - "::grpc::experimental::ClientWriteReactor* reactor) = 0;\n"); + "::grpc::experimental::ClientWriteReactor< $Request$>* " + "reactor) = 0;\n"); } else if (ServerOnlyStreaming(method)) { printer->Print(*vars, - "virtual ::grpc::experimental::ClientCallbackReader< " - "$Response$>* $Method$(::grpc::ClientContext* context, " + "virtual void $Method$(::grpc::ClientContext* context, " "$Request$* request, " - "::grpc::experimental::ClientReadReactor* reactor) = 0;\n"); + "::grpc::experimental::ClientReadReactor< $Response$>* " + "reactor) = 0;\n"); } else if (method->BidiStreaming()) { - printer->Print( - *vars, - "virtual ::grpc::experimental::ClientCallbackReaderWriter< $Request$, " - "$Response$>* $Method$(::grpc::ClientContext* context, " - "::grpc::experimental::ClientBidiReactor* reactor) = 0;\n"); + printer->Print(*vars, + "virtual void $Method$(::grpc::ClientContext* context, " + "::grpc::experimental::ClientBidiReactor< " + "$Request$,$Response$>* reactor) = 0;\n"); } } @@ -644,26 +643,23 @@ void PrintHeaderClientMethodCallback(grpc_generator::Printer* printer, "const $Request$* request, $Response$* response, " "std::function) override;\n"); } else if (ClientOnlyStreaming(method)) { - printer->Print( - *vars, - "::grpc::experimental::ClientCallbackWriter< $Request$>* " - "$Method$(::grpc::ClientContext* context, " - "$Response$* response, " - "::grpc::experimental::ClientWriteReactor* reactor) override;\n"); + printer->Print(*vars, + "void $Method$(::grpc::ClientContext* context, " + "$Response$* response, " + "::grpc::experimental::ClientWriteReactor< $Request$>* " + "reactor) override;\n"); } else if (ServerOnlyStreaming(method)) { - printer->Print( - *vars, - "::grpc::experimental::ClientCallbackReader< $Response$>* " - "$Method$(::grpc::ClientContext* context, " - "$Request$* request, " - "::grpc::experimental::ClientReadReactor* reactor) override;\n"); + printer->Print(*vars, + "void $Method$(::grpc::ClientContext* context, " + "$Request$* request, " + "::grpc::experimental::ClientReadReactor< $Response$>* " + "reactor) override;\n"); } else if (method->BidiStreaming()) { - printer->Print( - *vars, - "::grpc::experimental::ClientCallbackReaderWriter< $Request$, " - "$Response$>* $Method$(::grpc::ClientContext* context, " - "::grpc::experimental::ClientBidiReactor* reactor) override;\n"); + printer->Print(*vars, + "void $Method$(::grpc::ClientContext* context, " + "::grpc::experimental::ClientBidiReactor< " + "$Request$,$Response$>* reactor) override;\n"); } } @@ -1637,13 +1633,12 @@ void PrintSourceClientMethod(grpc_generator::Printer* printer, printer->Print( *vars, - "::grpc::experimental::ClientCallbackWriter< $Request$>* " - "$ns$$Service$::" + "void $ns$$Service$::" "Stub::experimental_async::$Method$(::grpc::ClientContext* context, " "$Response$* response, " - "::grpc::experimental::ClientWriteReactor* reactor) {\n"); + "::grpc::experimental::ClientWriteReactor< $Request$>* reactor) {\n"); printer->Print(*vars, - " return ::grpc::internal::ClientCallbackWriterFactory< " + " ::grpc::internal::ClientCallbackWriterFactory< " "$Request$>::Create(" "stub_->channel_.get(), " "stub_->rpcmethod_$Method$_, " @@ -1682,14 +1677,14 @@ void PrintSourceClientMethod(grpc_generator::Printer* printer, "context, request);\n" "}\n\n"); + printer->Print( + *vars, + "void $ns$$Service$::Stub::experimental_async::$Method$(::grpc::" + "ClientContext* context, " + "$Request$* request, " + "::grpc::experimental::ClientReadReactor< $Response$>* reactor) {\n"); printer->Print(*vars, - "::grpc::experimental::ClientCallbackReader< $Response$>* " - "$ns$$Service$::Stub::experimental_async::$Method$(::grpc::" - "ClientContext* context, " - "$Request$* request, " - "::grpc::experimental::ClientReadReactor* reactor) {\n"); - printer->Print(*vars, - " return ::grpc::internal::ClientCallbackReaderFactory< " + " ::grpc::internal::ClientCallbackReaderFactory< " "$Response$>::Create(" "stub_->channel_.get(), " "stub_->rpcmethod_$Method$_, " @@ -1728,20 +1723,19 @@ void PrintSourceClientMethod(grpc_generator::Printer* printer, "context);\n" "}\n\n"); - printer->Print(*vars, - "::grpc::experimental::ClientCallbackReaderWriter< " - "$Request$,$Response$>* " - "$ns$$Service$::Stub::experimental_async::$Method$(::grpc::" - "ClientContext* context, " - "::grpc::experimental::ClientBidiReactor* reactor) {\n"); printer->Print( *vars, - " return ::grpc::internal::ClientCallbackReaderWriterFactory< " - "$Request$,$Response$>::Create(" - "stub_->channel_.get(), " - "stub_->rpcmethod_$Method$_, " - "context, reactor);\n" - "}\n\n"); + "void $ns$$Service$::Stub::experimental_async::$Method$(::grpc::" + "ClientContext* context, " + "::grpc::experimental::ClientBidiReactor< $Request$,$Response$>* " + "reactor) {\n"); + printer->Print(*vars, + " ::grpc::internal::ClientCallbackReaderWriterFactory< " + "$Request$,$Response$>::Create(" + "stub_->channel_.get(), " + "stub_->rpcmethod_$Method$_, " + "context, reactor);\n" + "}\n\n"); for (auto async_prefix : async_prefixes) { (*vars)["AsyncPrefix"] = async_prefix.prefix; diff --git a/src/cpp/client/generic_stub.cc b/src/cpp/client/generic_stub.cc index f029daec65..f61c1b5317 100644 --- a/src/cpp/client/generic_stub.cc +++ b/src/cpp/client/generic_stub.cc @@ -72,16 +72,13 @@ void GenericStub::experimental_type::UnaryCall( context, request, response, std::move(on_completion)); } -experimental::ClientCallbackReaderWriter* -GenericStub::experimental_type::PrepareBidiStreamingCall( +void GenericStub::experimental_type::PrepareBidiStreamingCall( ClientContext* context, const grpc::string& method, - experimental::ClientBidiReactor* reactor) { - return internal::ClientCallbackReaderWriterFactory< - ByteBuffer, ByteBuffer>::Create(stub_->channel_.get(), - internal::RpcMethod( - method.c_str(), - internal::RpcMethod::BIDI_STREAMING), - context, reactor); + experimental::ClientBidiReactor* reactor) { + internal::ClientCallbackReaderWriterFactory::Create( + stub_->channel_.get(), + internal::RpcMethod(method.c_str(), internal::RpcMethod::BIDI_STREAMING), + context, reactor); } } // namespace grpc diff --git a/test/cpp/codegen/compiler_test_golden b/test/cpp/codegen/compiler_test_golden index 7a25f51d10..1a5fe27932 100644 --- a/test/cpp/codegen/compiler_test_golden +++ b/test/cpp/codegen/compiler_test_golden @@ -26,7 +26,6 @@ #include "src/proto/grpc/testing/compiler_test.pb.h" -#include #include #include #include @@ -39,6 +38,7 @@ #include #include #include +#include namespace grpc { class CompletionQueue; @@ -64,294 +64,556 @@ class ServiceA final { public: virtual ~StubInterface() {} // MethodA1 leading comment 1 - virtual ::grpc::Status MethodA1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::testing::Response* response) = 0; - std::unique_ptr< ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>> AsyncMethodA1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) { - return std::unique_ptr< ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>>(AsyncMethodA1Raw(context, request, cq)); - } - std::unique_ptr< ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>> PrepareAsyncMethodA1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) { - return std::unique_ptr< ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>>(PrepareAsyncMethodA1Raw(context, request, cq)); + virtual ::grpc::Status MethodA1(::grpc::ClientContext* context, + const ::grpc::testing::Request& request, + ::grpc::testing::Response* response) = 0; + std::unique_ptr< + ::grpc::ClientAsyncResponseReaderInterface<::grpc::testing::Response>> + AsyncMethodA1(::grpc::ClientContext* context, + const ::grpc::testing::Request& request, + ::grpc::CompletionQueue* cq) { + return std::unique_ptr<::grpc::ClientAsyncResponseReaderInterface< + ::grpc::testing::Response>>(AsyncMethodA1Raw(context, request, cq)); + } + std::unique_ptr< + ::grpc::ClientAsyncResponseReaderInterface<::grpc::testing::Response>> + PrepareAsyncMethodA1(::grpc::ClientContext* context, + const ::grpc::testing::Request& request, + ::grpc::CompletionQueue* cq) { + return std::unique_ptr<::grpc::ClientAsyncResponseReaderInterface< + ::grpc::testing::Response>>( + PrepareAsyncMethodA1Raw(context, request, cq)); } // MethodA1 trailing comment 1 // MethodA2 detached leading comment 1 // // Method A2 leading comment 1 // Method A2 leading comment 2 - std::unique_ptr< ::grpc::ClientWriterInterface< ::grpc::testing::Request>> MethodA2(::grpc::ClientContext* context, ::grpc::testing::Response* response) { - return std::unique_ptr< ::grpc::ClientWriterInterface< ::grpc::testing::Request>>(MethodA2Raw(context, response)); - } - std::unique_ptr< ::grpc::ClientAsyncWriterInterface< ::grpc::testing::Request>> AsyncMethodA2(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::CompletionQueue* cq, void* tag) { - return std::unique_ptr< ::grpc::ClientAsyncWriterInterface< ::grpc::testing::Request>>(AsyncMethodA2Raw(context, response, cq, tag)); - } - std::unique_ptr< ::grpc::ClientAsyncWriterInterface< ::grpc::testing::Request>> PrepareAsyncMethodA2(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::CompletionQueue* cq) { - return std::unique_ptr< ::grpc::ClientAsyncWriterInterface< ::grpc::testing::Request>>(PrepareAsyncMethodA2Raw(context, response, cq)); + std::unique_ptr<::grpc::ClientWriterInterface<::grpc::testing::Request>> + MethodA2(::grpc::ClientContext* context, + ::grpc::testing::Response* response) { + return std::unique_ptr< + ::grpc::ClientWriterInterface<::grpc::testing::Request>>( + MethodA2Raw(context, response)); + } + std::unique_ptr< + ::grpc::ClientAsyncWriterInterface<::grpc::testing::Request>> + AsyncMethodA2(::grpc::ClientContext* context, + ::grpc::testing::Response* response, + ::grpc::CompletionQueue* cq, void* tag) { + return std::unique_ptr< + ::grpc::ClientAsyncWriterInterface<::grpc::testing::Request>>( + AsyncMethodA2Raw(context, response, cq, tag)); + } + std::unique_ptr< + ::grpc::ClientAsyncWriterInterface<::grpc::testing::Request>> + PrepareAsyncMethodA2(::grpc::ClientContext* context, + ::grpc::testing::Response* response, + ::grpc::CompletionQueue* cq) { + return std::unique_ptr< + ::grpc::ClientAsyncWriterInterface<::grpc::testing::Request>>( + PrepareAsyncMethodA2Raw(context, response, cq)); } // MethodA2 trailing comment 1 // Method A3 leading comment 1 - std::unique_ptr< ::grpc::ClientReaderInterface< ::grpc::testing::Response>> MethodA3(::grpc::ClientContext* context, const ::grpc::testing::Request& request) { - return std::unique_ptr< ::grpc::ClientReaderInterface< ::grpc::testing::Response>>(MethodA3Raw(context, request)); - } - std::unique_ptr< ::grpc::ClientAsyncReaderInterface< ::grpc::testing::Response>> AsyncMethodA3(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq, void* tag) { - return std::unique_ptr< ::grpc::ClientAsyncReaderInterface< ::grpc::testing::Response>>(AsyncMethodA3Raw(context, request, cq, tag)); - } - std::unique_ptr< ::grpc::ClientAsyncReaderInterface< ::grpc::testing::Response>> PrepareAsyncMethodA3(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) { - return std::unique_ptr< ::grpc::ClientAsyncReaderInterface< ::grpc::testing::Response>>(PrepareAsyncMethodA3Raw(context, request, cq)); + std::unique_ptr<::grpc::ClientReaderInterface<::grpc::testing::Response>> + MethodA3(::grpc::ClientContext* context, + const ::grpc::testing::Request& request) { + return std::unique_ptr< + ::grpc::ClientReaderInterface<::grpc::testing::Response>>( + MethodA3Raw(context, request)); + } + std::unique_ptr< + ::grpc::ClientAsyncReaderInterface<::grpc::testing::Response>> + AsyncMethodA3(::grpc::ClientContext* context, + const ::grpc::testing::Request& request, + ::grpc::CompletionQueue* cq, void* tag) { + return std::unique_ptr< + ::grpc::ClientAsyncReaderInterface<::grpc::testing::Response>>( + AsyncMethodA3Raw(context, request, cq, tag)); + } + std::unique_ptr< + ::grpc::ClientAsyncReaderInterface<::grpc::testing::Response>> + PrepareAsyncMethodA3(::grpc::ClientContext* context, + const ::grpc::testing::Request& request, + ::grpc::CompletionQueue* cq) { + return std::unique_ptr< + ::grpc::ClientAsyncReaderInterface<::grpc::testing::Response>>( + PrepareAsyncMethodA3Raw(context, request, cq)); } // Method A3 trailing comment 1 // Method A4 leading comment 1 - std::unique_ptr< ::grpc::ClientReaderWriterInterface< ::grpc::testing::Request, ::grpc::testing::Response>> MethodA4(::grpc::ClientContext* context) { - return std::unique_ptr< ::grpc::ClientReaderWriterInterface< ::grpc::testing::Request, ::grpc::testing::Response>>(MethodA4Raw(context)); - } - std::unique_ptr< ::grpc::ClientAsyncReaderWriterInterface< ::grpc::testing::Request, ::grpc::testing::Response>> AsyncMethodA4(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq, void* tag) { - return std::unique_ptr< ::grpc::ClientAsyncReaderWriterInterface< ::grpc::testing::Request, ::grpc::testing::Response>>(AsyncMethodA4Raw(context, cq, tag)); - } - std::unique_ptr< ::grpc::ClientAsyncReaderWriterInterface< ::grpc::testing::Request, ::grpc::testing::Response>> PrepareAsyncMethodA4(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq) { - return std::unique_ptr< ::grpc::ClientAsyncReaderWriterInterface< ::grpc::testing::Request, ::grpc::testing::Response>>(PrepareAsyncMethodA4Raw(context, cq)); + std::unique_ptr<::grpc::ClientReaderWriterInterface< + ::grpc::testing::Request, ::grpc::testing::Response>> + MethodA4(::grpc::ClientContext* context) { + return std::unique_ptr<::grpc::ClientReaderWriterInterface< + ::grpc::testing::Request, ::grpc::testing::Response>>( + MethodA4Raw(context)); + } + std::unique_ptr<::grpc::ClientAsyncReaderWriterInterface< + ::grpc::testing::Request, ::grpc::testing::Response>> + AsyncMethodA4(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq, + void* tag) { + return std::unique_ptr<::grpc::ClientAsyncReaderWriterInterface< + ::grpc::testing::Request, ::grpc::testing::Response>>( + AsyncMethodA4Raw(context, cq, tag)); + } + std::unique_ptr<::grpc::ClientAsyncReaderWriterInterface< + ::grpc::testing::Request, ::grpc::testing::Response>> + PrepareAsyncMethodA4(::grpc::ClientContext* context, + ::grpc::CompletionQueue* cq) { + return std::unique_ptr<::grpc::ClientAsyncReaderWriterInterface< + ::grpc::testing::Request, ::grpc::testing::Response>>( + PrepareAsyncMethodA4Raw(context, cq)); } // Method A4 trailing comment 1 class experimental_async_interface { public: virtual ~experimental_async_interface() {} // MethodA1 leading comment 1 - virtual void MethodA1(::grpc::ClientContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response, std::function) = 0; + virtual void MethodA1(::grpc::ClientContext* context, + const ::grpc::testing::Request* request, + ::grpc::testing::Response* response, + std::function) = 0; // MethodA1 trailing comment 1 // MethodA2 detached leading comment 1 // // Method A2 leading comment 1 // Method A2 leading comment 2 - virtual ::grpc::experimental::ClientCallbackWriter< ::grpc::testing::Request>* MethodA2(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::experimental::ClientWriteReactor* reactor) = 0; + virtual void MethodA2( + ::grpc::ClientContext* context, ::grpc::testing::Response* response, + ::grpc::experimental::ClientWriteReactor<::grpc::testing::Request>* + reactor) = 0; // MethodA2 trailing comment 1 // Method A3 leading comment 1 - virtual ::grpc::experimental::ClientCallbackReader< ::grpc::testing::Response>* MethodA3(::grpc::ClientContext* context, ::grpc::testing::Request* request, ::grpc::experimental::ClientReadReactor* reactor) = 0; + virtual void MethodA3( + ::grpc::ClientContext* context, ::grpc::testing::Request* request, + ::grpc::experimental::ClientReadReactor<::grpc::testing::Response>* + reactor) = 0; // Method A3 trailing comment 1 // Method A4 leading comment 1 - virtual ::grpc::experimental::ClientCallbackReaderWriter< ::grpc::testing::Request, ::grpc::testing::Response>* MethodA4(::grpc::ClientContext* context, ::grpc::experimental::ClientBidiReactor* reactor) = 0; + virtual void MethodA4( + ::grpc::ClientContext* context, + ::grpc::experimental::ClientBidiReactor<::grpc::testing::Request, + ::grpc::testing::Response>* + reactor) = 0; // Method A4 trailing comment 1 }; - virtual class experimental_async_interface* experimental_async() { return nullptr; } - private: - virtual ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>* AsyncMethodA1Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) = 0; - virtual ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>* PrepareAsyncMethodA1Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) = 0; - virtual ::grpc::ClientWriterInterface< ::grpc::testing::Request>* MethodA2Raw(::grpc::ClientContext* context, ::grpc::testing::Response* response) = 0; - virtual ::grpc::ClientAsyncWriterInterface< ::grpc::testing::Request>* AsyncMethodA2Raw(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::CompletionQueue* cq, void* tag) = 0; - virtual ::grpc::ClientAsyncWriterInterface< ::grpc::testing::Request>* PrepareAsyncMethodA2Raw(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::CompletionQueue* cq) = 0; - virtual ::grpc::ClientReaderInterface< ::grpc::testing::Response>* MethodA3Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request) = 0; - virtual ::grpc::ClientAsyncReaderInterface< ::grpc::testing::Response>* AsyncMethodA3Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq, void* tag) = 0; - virtual ::grpc::ClientAsyncReaderInterface< ::grpc::testing::Response>* PrepareAsyncMethodA3Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) = 0; - virtual ::grpc::ClientReaderWriterInterface< ::grpc::testing::Request, ::grpc::testing::Response>* MethodA4Raw(::grpc::ClientContext* context) = 0; - virtual ::grpc::ClientAsyncReaderWriterInterface< ::grpc::testing::Request, ::grpc::testing::Response>* AsyncMethodA4Raw(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq, void* tag) = 0; - virtual ::grpc::ClientAsyncReaderWriterInterface< ::grpc::testing::Request, ::grpc::testing::Response>* PrepareAsyncMethodA4Raw(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq) = 0; + virtual class experimental_async_interface* experimental_async() { + return nullptr; + } + + private: + virtual ::grpc::ClientAsyncResponseReaderInterface< + ::grpc::testing::Response>* + AsyncMethodA1Raw(::grpc::ClientContext* context, + const ::grpc::testing::Request& request, + ::grpc::CompletionQueue* cq) = 0; + virtual ::grpc::ClientAsyncResponseReaderInterface< + ::grpc::testing::Response>* + PrepareAsyncMethodA1Raw(::grpc::ClientContext* context, + const ::grpc::testing::Request& request, + ::grpc::CompletionQueue* cq) = 0; + virtual ::grpc::ClientWriterInterface<::grpc::testing::Request>* + MethodA2Raw(::grpc::ClientContext* context, + ::grpc::testing::Response* response) = 0; + virtual ::grpc::ClientAsyncWriterInterface<::grpc::testing::Request>* + AsyncMethodA2Raw(::grpc::ClientContext* context, + ::grpc::testing::Response* response, + ::grpc::CompletionQueue* cq, void* tag) = 0; + virtual ::grpc::ClientAsyncWriterInterface<::grpc::testing::Request>* + PrepareAsyncMethodA2Raw(::grpc::ClientContext* context, + ::grpc::testing::Response* response, + ::grpc::CompletionQueue* cq) = 0; + virtual ::grpc::ClientReaderInterface<::grpc::testing::Response>* + MethodA3Raw(::grpc::ClientContext* context, + const ::grpc::testing::Request& request) = 0; + virtual ::grpc::ClientAsyncReaderInterface<::grpc::testing::Response>* + AsyncMethodA3Raw(::grpc::ClientContext* context, + const ::grpc::testing::Request& request, + ::grpc::CompletionQueue* cq, void* tag) = 0; + virtual ::grpc::ClientAsyncReaderInterface<::grpc::testing::Response>* + PrepareAsyncMethodA3Raw(::grpc::ClientContext* context, + const ::grpc::testing::Request& request, + ::grpc::CompletionQueue* cq) = 0; + virtual ::grpc::ClientReaderWriterInterface<::grpc::testing::Request, + ::grpc::testing::Response>* + MethodA4Raw(::grpc::ClientContext* context) = 0; + virtual ::grpc::ClientAsyncReaderWriterInterface<::grpc::testing::Request, + ::grpc::testing::Response>* + AsyncMethodA4Raw(::grpc::ClientContext* context, + ::grpc::CompletionQueue* cq, void* tag) = 0; + virtual ::grpc::ClientAsyncReaderWriterInterface<::grpc::testing::Request, + ::grpc::testing::Response>* + PrepareAsyncMethodA4Raw(::grpc::ClientContext* context, + ::grpc::CompletionQueue* cq) = 0; }; class Stub final : public StubInterface { public: - Stub(const std::shared_ptr< ::grpc::ChannelInterface>& channel); - ::grpc::Status MethodA1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::testing::Response* response) override; - std::unique_ptr< ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>> AsyncMethodA1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) { - return std::unique_ptr< ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>>(AsyncMethodA1Raw(context, request, cq)); - } - std::unique_ptr< ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>> PrepareAsyncMethodA1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) { - return std::unique_ptr< ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>>(PrepareAsyncMethodA1Raw(context, request, cq)); - } - std::unique_ptr< ::grpc::ClientWriter< ::grpc::testing::Request>> MethodA2(::grpc::ClientContext* context, ::grpc::testing::Response* response) { - return std::unique_ptr< ::grpc::ClientWriter< ::grpc::testing::Request>>(MethodA2Raw(context, response)); - } - std::unique_ptr< ::grpc::ClientAsyncWriter< ::grpc::testing::Request>> AsyncMethodA2(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::CompletionQueue* cq, void* tag) { - return std::unique_ptr< ::grpc::ClientAsyncWriter< ::grpc::testing::Request>>(AsyncMethodA2Raw(context, response, cq, tag)); - } - std::unique_ptr< ::grpc::ClientAsyncWriter< ::grpc::testing::Request>> PrepareAsyncMethodA2(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::CompletionQueue* cq) { - return std::unique_ptr< ::grpc::ClientAsyncWriter< ::grpc::testing::Request>>(PrepareAsyncMethodA2Raw(context, response, cq)); - } - std::unique_ptr< ::grpc::ClientReader< ::grpc::testing::Response>> MethodA3(::grpc::ClientContext* context, const ::grpc::testing::Request& request) { - return std::unique_ptr< ::grpc::ClientReader< ::grpc::testing::Response>>(MethodA3Raw(context, request)); - } - std::unique_ptr< ::grpc::ClientAsyncReader< ::grpc::testing::Response>> AsyncMethodA3(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq, void* tag) { - return std::unique_ptr< ::grpc::ClientAsyncReader< ::grpc::testing::Response>>(AsyncMethodA3Raw(context, request, cq, tag)); - } - std::unique_ptr< ::grpc::ClientAsyncReader< ::grpc::testing::Response>> PrepareAsyncMethodA3(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) { - return std::unique_ptr< ::grpc::ClientAsyncReader< ::grpc::testing::Response>>(PrepareAsyncMethodA3Raw(context, request, cq)); - } - std::unique_ptr< ::grpc::ClientReaderWriter< ::grpc::testing::Request, ::grpc::testing::Response>> MethodA4(::grpc::ClientContext* context) { - return std::unique_ptr< ::grpc::ClientReaderWriter< ::grpc::testing::Request, ::grpc::testing::Response>>(MethodA4Raw(context)); - } - std::unique_ptr< ::grpc::ClientAsyncReaderWriter< ::grpc::testing::Request, ::grpc::testing::Response>> AsyncMethodA4(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq, void* tag) { - return std::unique_ptr< ::grpc::ClientAsyncReaderWriter< ::grpc::testing::Request, ::grpc::testing::Response>>(AsyncMethodA4Raw(context, cq, tag)); - } - std::unique_ptr< ::grpc::ClientAsyncReaderWriter< ::grpc::testing::Request, ::grpc::testing::Response>> PrepareAsyncMethodA4(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq) { - return std::unique_ptr< ::grpc::ClientAsyncReaderWriter< ::grpc::testing::Request, ::grpc::testing::Response>>(PrepareAsyncMethodA4Raw(context, cq)); - } - class experimental_async final : - public StubInterface::experimental_async_interface { + Stub(const std::shared_ptr<::grpc::ChannelInterface>& channel); + ::grpc::Status MethodA1(::grpc::ClientContext* context, + const ::grpc::testing::Request& request, + ::grpc::testing::Response* response) override; + std::unique_ptr< + ::grpc::ClientAsyncResponseReader<::grpc::testing::Response>> + AsyncMethodA1(::grpc::ClientContext* context, + const ::grpc::testing::Request& request, + ::grpc::CompletionQueue* cq) { + return std::unique_ptr< + ::grpc::ClientAsyncResponseReader<::grpc::testing::Response>>( + AsyncMethodA1Raw(context, request, cq)); + } + std::unique_ptr< + ::grpc::ClientAsyncResponseReader<::grpc::testing::Response>> + PrepareAsyncMethodA1(::grpc::ClientContext* context, + const ::grpc::testing::Request& request, + ::grpc::CompletionQueue* cq) { + return std::unique_ptr< + ::grpc::ClientAsyncResponseReader<::grpc::testing::Response>>( + PrepareAsyncMethodA1Raw(context, request, cq)); + } + std::unique_ptr<::grpc::ClientWriter<::grpc::testing::Request>> MethodA2( + ::grpc::ClientContext* context, ::grpc::testing::Response* response) { + return std::unique_ptr<::grpc::ClientWriter<::grpc::testing::Request>>( + MethodA2Raw(context, response)); + } + std::unique_ptr<::grpc::ClientAsyncWriter<::grpc::testing::Request>> + AsyncMethodA2(::grpc::ClientContext* context, + ::grpc::testing::Response* response, + ::grpc::CompletionQueue* cq, void* tag) { + return std::unique_ptr< + ::grpc::ClientAsyncWriter<::grpc::testing::Request>>( + AsyncMethodA2Raw(context, response, cq, tag)); + } + std::unique_ptr<::grpc::ClientAsyncWriter<::grpc::testing::Request>> + PrepareAsyncMethodA2(::grpc::ClientContext* context, + ::grpc::testing::Response* response, + ::grpc::CompletionQueue* cq) { + return std::unique_ptr< + ::grpc::ClientAsyncWriter<::grpc::testing::Request>>( + PrepareAsyncMethodA2Raw(context, response, cq)); + } + std::unique_ptr<::grpc::ClientReader<::grpc::testing::Response>> MethodA3( + ::grpc::ClientContext* context, + const ::grpc::testing::Request& request) { + return std::unique_ptr<::grpc::ClientReader<::grpc::testing::Response>>( + MethodA3Raw(context, request)); + } + std::unique_ptr<::grpc::ClientAsyncReader<::grpc::testing::Response>> + AsyncMethodA3(::grpc::ClientContext* context, + const ::grpc::testing::Request& request, + ::grpc::CompletionQueue* cq, void* tag) { + return std::unique_ptr< + ::grpc::ClientAsyncReader<::grpc::testing::Response>>( + AsyncMethodA3Raw(context, request, cq, tag)); + } + std::unique_ptr<::grpc::ClientAsyncReader<::grpc::testing::Response>> + PrepareAsyncMethodA3(::grpc::ClientContext* context, + const ::grpc::testing::Request& request, + ::grpc::CompletionQueue* cq) { + return std::unique_ptr< + ::grpc::ClientAsyncReader<::grpc::testing::Response>>( + PrepareAsyncMethodA3Raw(context, request, cq)); + } + std::unique_ptr<::grpc::ClientReaderWriter<::grpc::testing::Request, + ::grpc::testing::Response>> + MethodA4(::grpc::ClientContext* context) { + return std::unique_ptr<::grpc::ClientReaderWriter< + ::grpc::testing::Request, ::grpc::testing::Response>>( + MethodA4Raw(context)); + } + std::unique_ptr<::grpc::ClientAsyncReaderWriter<::grpc::testing::Request, + ::grpc::testing::Response>> + AsyncMethodA4(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq, + void* tag) { + return std::unique_ptr<::grpc::ClientAsyncReaderWriter< + ::grpc::testing::Request, ::grpc::testing::Response>>( + AsyncMethodA4Raw(context, cq, tag)); + } + std::unique_ptr<::grpc::ClientAsyncReaderWriter<::grpc::testing::Request, + ::grpc::testing::Response>> + PrepareAsyncMethodA4(::grpc::ClientContext* context, + ::grpc::CompletionQueue* cq) { + return std::unique_ptr<::grpc::ClientAsyncReaderWriter< + ::grpc::testing::Request, ::grpc::testing::Response>>( + PrepareAsyncMethodA4Raw(context, cq)); + } + class experimental_async final + : public StubInterface::experimental_async_interface { public: - void MethodA1(::grpc::ClientContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response, std::function) override; - ::grpc::experimental::ClientCallbackWriter< ::grpc::testing::Request>* MethodA2(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::experimental::ClientWriteReactor* reactor) override; - ::grpc::experimental::ClientCallbackReader< ::grpc::testing::Response>* MethodA3(::grpc::ClientContext* context, ::grpc::testing::Request* request, ::grpc::experimental::ClientReadReactor* reactor) override; - ::grpc::experimental::ClientCallbackReaderWriter< ::grpc::testing::Request, ::grpc::testing::Response>* MethodA4(::grpc::ClientContext* context, ::grpc::experimental::ClientBidiReactor* reactor) override; + void MethodA1(::grpc::ClientContext* context, + const ::grpc::testing::Request* request, + ::grpc::testing::Response* response, + std::function) override; + void MethodA2( + ::grpc::ClientContext* context, ::grpc::testing::Response* response, + ::grpc::experimental::ClientWriteReactor<::grpc::testing::Request>* + reactor) override; + void MethodA3( + ::grpc::ClientContext* context, ::grpc::testing::Request* request, + ::grpc::experimental::ClientReadReactor<::grpc::testing::Response>* + reactor) override; + void MethodA4(::grpc::ClientContext* context, + ::grpc::experimental::ClientBidiReactor< + ::grpc::testing::Request, ::grpc::testing::Response>* + reactor) override; + private: friend class Stub; - explicit experimental_async(Stub* stub): stub_(stub) { } + explicit experimental_async(Stub* stub) : stub_(stub) {} Stub* stub() { return stub_; } Stub* stub_; }; - class experimental_async_interface* experimental_async() override { return &async_stub_; } - - private: - std::shared_ptr< ::grpc::ChannelInterface> channel_; - class experimental_async async_stub_{this}; - ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>* AsyncMethodA1Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) override; - ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>* PrepareAsyncMethodA1Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) override; - ::grpc::ClientWriter< ::grpc::testing::Request>* MethodA2Raw(::grpc::ClientContext* context, ::grpc::testing::Response* response) override; - ::grpc::ClientAsyncWriter< ::grpc::testing::Request>* AsyncMethodA2Raw(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::CompletionQueue* cq, void* tag) override; - ::grpc::ClientAsyncWriter< ::grpc::testing::Request>* PrepareAsyncMethodA2Raw(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::CompletionQueue* cq) override; - ::grpc::ClientReader< ::grpc::testing::Response>* MethodA3Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request) override; - ::grpc::ClientAsyncReader< ::grpc::testing::Response>* AsyncMethodA3Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq, void* tag) override; - ::grpc::ClientAsyncReader< ::grpc::testing::Response>* PrepareAsyncMethodA3Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) override; - ::grpc::ClientReaderWriter< ::grpc::testing::Request, ::grpc::testing::Response>* MethodA4Raw(::grpc::ClientContext* context) override; - ::grpc::ClientAsyncReaderWriter< ::grpc::testing::Request, ::grpc::testing::Response>* AsyncMethodA4Raw(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq, void* tag) override; - ::grpc::ClientAsyncReaderWriter< ::grpc::testing::Request, ::grpc::testing::Response>* PrepareAsyncMethodA4Raw(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq) override; + class experimental_async_interface* experimental_async() override { + return &async_stub_; + } + + private: + std::shared_ptr<::grpc::ChannelInterface> channel_; + class experimental_async async_stub_ { + this + }; + ::grpc::ClientAsyncResponseReader<::grpc::testing::Response>* + AsyncMethodA1Raw(::grpc::ClientContext* context, + const ::grpc::testing::Request& request, + ::grpc::CompletionQueue* cq) override; + ::grpc::ClientAsyncResponseReader<::grpc::testing::Response>* + PrepareAsyncMethodA1Raw(::grpc::ClientContext* context, + const ::grpc::testing::Request& request, + ::grpc::CompletionQueue* cq) override; + ::grpc::ClientWriter<::grpc::testing::Request>* MethodA2Raw( + ::grpc::ClientContext* context, + ::grpc::testing::Response* response) override; + ::grpc::ClientAsyncWriter<::grpc::testing::Request>* AsyncMethodA2Raw( + ::grpc::ClientContext* context, ::grpc::testing::Response* response, + ::grpc::CompletionQueue* cq, void* tag) override; + ::grpc::ClientAsyncWriter<::grpc::testing::Request>* + PrepareAsyncMethodA2Raw(::grpc::ClientContext* context, + ::grpc::testing::Response* response, + ::grpc::CompletionQueue* cq) override; + ::grpc::ClientReader<::grpc::testing::Response>* MethodA3Raw( + ::grpc::ClientContext* context, + const ::grpc::testing::Request& request) override; + ::grpc::ClientAsyncReader<::grpc::testing::Response>* AsyncMethodA3Raw( + ::grpc::ClientContext* context, const ::grpc::testing::Request& request, + ::grpc::CompletionQueue* cq, void* tag) override; + ::grpc::ClientAsyncReader<::grpc::testing::Response>* + PrepareAsyncMethodA3Raw(::grpc::ClientContext* context, + const ::grpc::testing::Request& request, + ::grpc::CompletionQueue* cq) override; + ::grpc::ClientReaderWriter<::grpc::testing::Request, + ::grpc::testing::Response>* + MethodA4Raw(::grpc::ClientContext* context) override; + ::grpc::ClientAsyncReaderWriter<::grpc::testing::Request, + ::grpc::testing::Response>* + AsyncMethodA4Raw(::grpc::ClientContext* context, + ::grpc::CompletionQueue* cq, void* tag) override; + ::grpc::ClientAsyncReaderWriter<::grpc::testing::Request, + ::grpc::testing::Response>* + PrepareAsyncMethodA4Raw(::grpc::ClientContext* context, + ::grpc::CompletionQueue* cq) override; const ::grpc::internal::RpcMethod rpcmethod_MethodA1_; const ::grpc::internal::RpcMethod rpcmethod_MethodA2_; const ::grpc::internal::RpcMethod rpcmethod_MethodA3_; const ::grpc::internal::RpcMethod rpcmethod_MethodA4_; }; - static std::unique_ptr NewStub(const std::shared_ptr< ::grpc::ChannelInterface>& channel, const ::grpc::StubOptions& options = ::grpc::StubOptions()); + static std::unique_ptr NewStub( + const std::shared_ptr<::grpc::ChannelInterface>& channel, + const ::grpc::StubOptions& options = ::grpc::StubOptions()); class Service : public ::grpc::Service { public: Service(); virtual ~Service(); // MethodA1 leading comment 1 - virtual ::grpc::Status MethodA1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response); + virtual ::grpc::Status MethodA1(::grpc::ServerContext* context, + const ::grpc::testing::Request* request, + ::grpc::testing::Response* response); // MethodA1 trailing comment 1 // MethodA2 detached leading comment 1 // // Method A2 leading comment 1 // Method A2 leading comment 2 - virtual ::grpc::Status MethodA2(::grpc::ServerContext* context, ::grpc::ServerReader< ::grpc::testing::Request>* reader, ::grpc::testing::Response* response); + virtual ::grpc::Status MethodA2( + ::grpc::ServerContext* context, + ::grpc::ServerReader<::grpc::testing::Request>* reader, + ::grpc::testing::Response* response); // MethodA2 trailing comment 1 // Method A3 leading comment 1 - virtual ::grpc::Status MethodA3(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::ServerWriter< ::grpc::testing::Response>* writer); + virtual ::grpc::Status MethodA3( + ::grpc::ServerContext* context, const ::grpc::testing::Request* request, + ::grpc::ServerWriter<::grpc::testing::Response>* writer); // Method A3 trailing comment 1 // Method A4 leading comment 1 - virtual ::grpc::Status MethodA4(::grpc::ServerContext* context, ::grpc::ServerReaderWriter< ::grpc::testing::Response, ::grpc::testing::Request>* stream); + virtual ::grpc::Status MethodA4( + ::grpc::ServerContext* context, + ::grpc::ServerReaderWriter<::grpc::testing::Response, + ::grpc::testing::Request>* stream); // Method A4 trailing comment 1 }; template class WithAsyncMethod_MethodA1 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: - WithAsyncMethod_MethodA1() { - ::grpc::Service::MarkMethodAsync(0); - } + WithAsyncMethod_MethodA1() { ::grpc::Service::MarkMethodAsync(0); } ~WithAsyncMethod_MethodA1() override { BaseClassMustBeDerivedFromService(this); } // disable synchronous version of this method - ::grpc::Status MethodA1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response) override { + ::grpc::Status MethodA1(::grpc::ServerContext* context, + const ::grpc::testing::Request* request, + ::grpc::testing::Response* response) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } - void RequestMethodA1(::grpc::ServerContext* context, ::grpc::testing::Request* request, ::grpc::ServerAsyncResponseWriter< ::grpc::testing::Response>* response, ::grpc::CompletionQueue* new_call_cq, ::grpc::ServerCompletionQueue* notification_cq, void *tag) { - ::grpc::Service::RequestAsyncUnary(0, context, request, response, new_call_cq, notification_cq, tag); + void RequestMethodA1( + ::grpc::ServerContext* context, ::grpc::testing::Request* request, + ::grpc::ServerAsyncResponseWriter<::grpc::testing::Response>* response, + ::grpc::CompletionQueue* new_call_cq, + ::grpc::ServerCompletionQueue* notification_cq, void* tag) { + ::grpc::Service::RequestAsyncUnary(0, context, request, response, + new_call_cq, notification_cq, tag); } }; template class WithAsyncMethod_MethodA2 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: - WithAsyncMethod_MethodA2() { - ::grpc::Service::MarkMethodAsync(1); - } + WithAsyncMethod_MethodA2() { ::grpc::Service::MarkMethodAsync(1); } ~WithAsyncMethod_MethodA2() override { BaseClassMustBeDerivedFromService(this); } // disable synchronous version of this method - ::grpc::Status MethodA2(::grpc::ServerContext* context, ::grpc::ServerReader< ::grpc::testing::Request>* reader, ::grpc::testing::Response* response) override { + ::grpc::Status MethodA2( + ::grpc::ServerContext* context, + ::grpc::ServerReader<::grpc::testing::Request>* reader, + ::grpc::testing::Response* response) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } - void RequestMethodA2(::grpc::ServerContext* context, ::grpc::ServerAsyncReader< ::grpc::testing::Response, ::grpc::testing::Request>* reader, ::grpc::CompletionQueue* new_call_cq, ::grpc::ServerCompletionQueue* notification_cq, void *tag) { - ::grpc::Service::RequestAsyncClientStreaming(1, context, reader, new_call_cq, notification_cq, tag); + void RequestMethodA2( + ::grpc::ServerContext* context, + ::grpc::ServerAsyncReader<::grpc::testing::Response, + ::grpc::testing::Request>* reader, + ::grpc::CompletionQueue* new_call_cq, + ::grpc::ServerCompletionQueue* notification_cq, void* tag) { + ::grpc::Service::RequestAsyncClientStreaming( + 1, context, reader, new_call_cq, notification_cq, tag); } }; template class WithAsyncMethod_MethodA3 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: - WithAsyncMethod_MethodA3() { - ::grpc::Service::MarkMethodAsync(2); - } + WithAsyncMethod_MethodA3() { ::grpc::Service::MarkMethodAsync(2); } ~WithAsyncMethod_MethodA3() override { BaseClassMustBeDerivedFromService(this); } // disable synchronous version of this method - ::grpc::Status MethodA3(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::ServerWriter< ::grpc::testing::Response>* writer) override { + ::grpc::Status MethodA3( + ::grpc::ServerContext* context, const ::grpc::testing::Request* request, + ::grpc::ServerWriter<::grpc::testing::Response>* writer) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } - void RequestMethodA3(::grpc::ServerContext* context, ::grpc::testing::Request* request, ::grpc::ServerAsyncWriter< ::grpc::testing::Response>* writer, ::grpc::CompletionQueue* new_call_cq, ::grpc::ServerCompletionQueue* notification_cq, void *tag) { - ::grpc::Service::RequestAsyncServerStreaming(2, context, request, writer, new_call_cq, notification_cq, tag); + void RequestMethodA3( + ::grpc::ServerContext* context, ::grpc::testing::Request* request, + ::grpc::ServerAsyncWriter<::grpc::testing::Response>* writer, + ::grpc::CompletionQueue* new_call_cq, + ::grpc::ServerCompletionQueue* notification_cq, void* tag) { + ::grpc::Service::RequestAsyncServerStreaming( + 2, context, request, writer, new_call_cq, notification_cq, tag); } }; template class WithAsyncMethod_MethodA4 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: - WithAsyncMethod_MethodA4() { - ::grpc::Service::MarkMethodAsync(3); - } + WithAsyncMethod_MethodA4() { ::grpc::Service::MarkMethodAsync(3); } ~WithAsyncMethod_MethodA4() override { BaseClassMustBeDerivedFromService(this); } // disable synchronous version of this method - ::grpc::Status MethodA4(::grpc::ServerContext* context, ::grpc::ServerReaderWriter< ::grpc::testing::Response, ::grpc::testing::Request>* stream) override { + ::grpc::Status MethodA4( + ::grpc::ServerContext* context, + ::grpc::ServerReaderWriter<::grpc::testing::Response, + ::grpc::testing::Request>* stream) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } - void RequestMethodA4(::grpc::ServerContext* context, ::grpc::ServerAsyncReaderWriter< ::grpc::testing::Response, ::grpc::testing::Request>* stream, ::grpc::CompletionQueue* new_call_cq, ::grpc::ServerCompletionQueue* notification_cq, void *tag) { - ::grpc::Service::RequestAsyncBidiStreaming(3, context, stream, new_call_cq, notification_cq, tag); + void RequestMethodA4( + ::grpc::ServerContext* context, + ::grpc::ServerAsyncReaderWriter<::grpc::testing::Response, + ::grpc::testing::Request>* stream, + ::grpc::CompletionQueue* new_call_cq, + ::grpc::ServerCompletionQueue* notification_cq, void* tag) { + ::grpc::Service::RequestAsyncBidiStreaming( + 3, context, stream, new_call_cq, notification_cq, tag); } }; - typedef WithAsyncMethod_MethodA1 > > > AsyncService; + typedef WithAsyncMethod_MethodA1>>> + AsyncService; template class ExperimentalWithCallbackMethod_MethodA1 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: ExperimentalWithCallbackMethod_MethodA1() { - ::grpc::Service::experimental().MarkMethodCallback(0, - new ::grpc::internal::CallbackUnaryHandler< ExperimentalWithCallbackMethod_MethodA1, ::grpc::testing::Request, ::grpc::testing::Response>( - [this](::grpc::ServerContext* context, - const ::grpc::testing::Request* request, - ::grpc::testing::Response* response, - ::grpc::experimental::ServerCallbackRpcController* controller) { + ::grpc::Service::experimental().MarkMethodCallback( + 0, new ::grpc::internal::CallbackUnaryHandler< + ExperimentalWithCallbackMethod_MethodA1, + ::grpc::testing::Request, ::grpc::testing::Response>( + [this](::grpc::ServerContext* context, + const ::grpc::testing::Request* request, + ::grpc::testing::Response* response, + ::grpc::experimental::ServerCallbackRpcController* + controller) { this->MethodA1(context, request, response, controller); - }, this)); + }, + this)); } ~ExperimentalWithCallbackMethod_MethodA1() override { BaseClassMustBeDerivedFromService(this); } // disable synchronous version of this method - ::grpc::Status MethodA1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response) override { + ::grpc::Status MethodA1(::grpc::ServerContext* context, + const ::grpc::testing::Request* request, + ::grpc::testing::Response* response) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } - virtual void MethodA1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response, ::grpc::experimental::ServerCallbackRpcController* controller) { controller->Finish(::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "")); } + virtual void MethodA1( + ::grpc::ServerContext* context, const ::grpc::testing::Request* request, + ::grpc::testing::Response* response, + ::grpc::experimental::ServerCallbackRpcController* controller) { + controller->Finish(::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "")); + } }; template class ExperimentalWithCallbackMethod_MethodA2 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: - ExperimentalWithCallbackMethod_MethodA2() { - } + ExperimentalWithCallbackMethod_MethodA2() {} ~ExperimentalWithCallbackMethod_MethodA2() override { BaseClassMustBeDerivedFromService(this); } // disable synchronous version of this method - ::grpc::Status MethodA2(::grpc::ServerContext* context, ::grpc::ServerReader< ::grpc::testing::Request>* reader, ::grpc::testing::Response* response) override { + ::grpc::Status MethodA2( + ::grpc::ServerContext* context, + ::grpc::ServerReader<::grpc::testing::Request>* reader, + ::grpc::testing::Response* response) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } @@ -359,15 +621,17 @@ class ServiceA final { template class ExperimentalWithCallbackMethod_MethodA3 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: - ExperimentalWithCallbackMethod_MethodA3() { - } + ExperimentalWithCallbackMethod_MethodA3() {} ~ExperimentalWithCallbackMethod_MethodA3() override { BaseClassMustBeDerivedFromService(this); } // disable synchronous version of this method - ::grpc::Status MethodA3(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::ServerWriter< ::grpc::testing::Response>* writer) override { + ::grpc::Status MethodA3( + ::grpc::ServerContext* context, const ::grpc::testing::Request* request, + ::grpc::ServerWriter<::grpc::testing::Response>* writer) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } @@ -375,33 +639,41 @@ class ServiceA final { template class ExperimentalWithCallbackMethod_MethodA4 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: - ExperimentalWithCallbackMethod_MethodA4() { - } + ExperimentalWithCallbackMethod_MethodA4() {} ~ExperimentalWithCallbackMethod_MethodA4() override { BaseClassMustBeDerivedFromService(this); } // disable synchronous version of this method - ::grpc::Status MethodA4(::grpc::ServerContext* context, ::grpc::ServerReaderWriter< ::grpc::testing::Response, ::grpc::testing::Request>* stream) override { + ::grpc::Status MethodA4( + ::grpc::ServerContext* context, + ::grpc::ServerReaderWriter<::grpc::testing::Response, + ::grpc::testing::Request>* stream) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } }; - typedef ExperimentalWithCallbackMethod_MethodA1 > > > ExperimentalCallbackService; + typedef ExperimentalWithCallbackMethod_MethodA1< + ExperimentalWithCallbackMethod_MethodA2< + ExperimentalWithCallbackMethod_MethodA3< + ExperimentalWithCallbackMethod_MethodA4>>> + ExperimentalCallbackService; template class WithGenericMethod_MethodA1 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: - WithGenericMethod_MethodA1() { - ::grpc::Service::MarkMethodGeneric(0); - } + WithGenericMethod_MethodA1() { ::grpc::Service::MarkMethodGeneric(0); } ~WithGenericMethod_MethodA1() override { BaseClassMustBeDerivedFromService(this); } // disable synchronous version of this method - ::grpc::Status MethodA1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response) override { + ::grpc::Status MethodA1(::grpc::ServerContext* context, + const ::grpc::testing::Request* request, + ::grpc::testing::Response* response) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } @@ -409,16 +681,18 @@ class ServiceA final { template class WithGenericMethod_MethodA2 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: - WithGenericMethod_MethodA2() { - ::grpc::Service::MarkMethodGeneric(1); - } + WithGenericMethod_MethodA2() { ::grpc::Service::MarkMethodGeneric(1); } ~WithGenericMethod_MethodA2() override { BaseClassMustBeDerivedFromService(this); } // disable synchronous version of this method - ::grpc::Status MethodA2(::grpc::ServerContext* context, ::grpc::ServerReader< ::grpc::testing::Request>* reader, ::grpc::testing::Response* response) override { + ::grpc::Status MethodA2( + ::grpc::ServerContext* context, + ::grpc::ServerReader<::grpc::testing::Request>* reader, + ::grpc::testing::Response* response) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } @@ -426,16 +700,17 @@ class ServiceA final { template class WithGenericMethod_MethodA3 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: - WithGenericMethod_MethodA3() { - ::grpc::Service::MarkMethodGeneric(2); - } + WithGenericMethod_MethodA3() { ::grpc::Service::MarkMethodGeneric(2); } ~WithGenericMethod_MethodA3() override { BaseClassMustBeDerivedFromService(this); } // disable synchronous version of this method - ::grpc::Status MethodA3(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::ServerWriter< ::grpc::testing::Response>* writer) override { + ::grpc::Status MethodA3( + ::grpc::ServerContext* context, const ::grpc::testing::Request* request, + ::grpc::ServerWriter<::grpc::testing::Response>* writer) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } @@ -443,16 +718,18 @@ class ServiceA final { template class WithGenericMethod_MethodA4 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: - WithGenericMethod_MethodA4() { - ::grpc::Service::MarkMethodGeneric(3); - } + WithGenericMethod_MethodA4() { ::grpc::Service::MarkMethodGeneric(3); } ~WithGenericMethod_MethodA4() override { BaseClassMustBeDerivedFromService(this); } // disable synchronous version of this method - ::grpc::Status MethodA4(::grpc::ServerContext* context, ::grpc::ServerReaderWriter< ::grpc::testing::Response, ::grpc::testing::Request>* stream) override { + ::grpc::Status MethodA4( + ::grpc::ServerContext* context, + ::grpc::ServerReaderWriter<::grpc::testing::Response, + ::grpc::testing::Request>* stream) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } @@ -460,120 +737,164 @@ class ServiceA final { template class WithRawMethod_MethodA1 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: - WithRawMethod_MethodA1() { - ::grpc::Service::MarkMethodRaw(0); - } + WithRawMethod_MethodA1() { ::grpc::Service::MarkMethodRaw(0); } ~WithRawMethod_MethodA1() override { BaseClassMustBeDerivedFromService(this); } // disable synchronous version of this method - ::grpc::Status MethodA1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response) override { + ::grpc::Status MethodA1(::grpc::ServerContext* context, + const ::grpc::testing::Request* request, + ::grpc::testing::Response* response) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } - void RequestMethodA1(::grpc::ServerContext* context, ::grpc::ByteBuffer* request, ::grpc::ServerAsyncResponseWriter< ::grpc::ByteBuffer>* response, ::grpc::CompletionQueue* new_call_cq, ::grpc::ServerCompletionQueue* notification_cq, void *tag) { - ::grpc::Service::RequestAsyncUnary(0, context, request, response, new_call_cq, notification_cq, tag); + void RequestMethodA1( + ::grpc::ServerContext* context, ::grpc::ByteBuffer* request, + ::grpc::ServerAsyncResponseWriter<::grpc::ByteBuffer>* response, + ::grpc::CompletionQueue* new_call_cq, + ::grpc::ServerCompletionQueue* notification_cq, void* tag) { + ::grpc::Service::RequestAsyncUnary(0, context, request, response, + new_call_cq, notification_cq, tag); } }; template class WithRawMethod_MethodA2 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: - WithRawMethod_MethodA2() { - ::grpc::Service::MarkMethodRaw(1); - } + WithRawMethod_MethodA2() { ::grpc::Service::MarkMethodRaw(1); } ~WithRawMethod_MethodA2() override { BaseClassMustBeDerivedFromService(this); } // disable synchronous version of this method - ::grpc::Status MethodA2(::grpc::ServerContext* context, ::grpc::ServerReader< ::grpc::testing::Request>* reader, ::grpc::testing::Response* response) override { + ::grpc::Status MethodA2( + ::grpc::ServerContext* context, + ::grpc::ServerReader<::grpc::testing::Request>* reader, + ::grpc::testing::Response* response) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } - void RequestMethodA2(::grpc::ServerContext* context, ::grpc::ServerAsyncReader< ::grpc::ByteBuffer, ::grpc::ByteBuffer>* reader, ::grpc::CompletionQueue* new_call_cq, ::grpc::ServerCompletionQueue* notification_cq, void *tag) { - ::grpc::Service::RequestAsyncClientStreaming(1, context, reader, new_call_cq, notification_cq, tag); + void RequestMethodA2(::grpc::ServerContext* context, + ::grpc::ServerAsyncReader<::grpc::ByteBuffer, + ::grpc::ByteBuffer>* reader, + ::grpc::CompletionQueue* new_call_cq, + ::grpc::ServerCompletionQueue* notification_cq, + void* tag) { + ::grpc::Service::RequestAsyncClientStreaming( + 1, context, reader, new_call_cq, notification_cq, tag); } }; template class WithRawMethod_MethodA3 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: - WithRawMethod_MethodA3() { - ::grpc::Service::MarkMethodRaw(2); - } + WithRawMethod_MethodA3() { ::grpc::Service::MarkMethodRaw(2); } ~WithRawMethod_MethodA3() override { BaseClassMustBeDerivedFromService(this); } // disable synchronous version of this method - ::grpc::Status MethodA3(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::ServerWriter< ::grpc::testing::Response>* writer) override { + ::grpc::Status MethodA3( + ::grpc::ServerContext* context, const ::grpc::testing::Request* request, + ::grpc::ServerWriter<::grpc::testing::Response>* writer) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } - void RequestMethodA3(::grpc::ServerContext* context, ::grpc::ByteBuffer* request, ::grpc::ServerAsyncWriter< ::grpc::ByteBuffer>* writer, ::grpc::CompletionQueue* new_call_cq, ::grpc::ServerCompletionQueue* notification_cq, void *tag) { - ::grpc::Service::RequestAsyncServerStreaming(2, context, request, writer, new_call_cq, notification_cq, tag); + void RequestMethodA3(::grpc::ServerContext* context, + ::grpc::ByteBuffer* request, + ::grpc::ServerAsyncWriter<::grpc::ByteBuffer>* writer, + ::grpc::CompletionQueue* new_call_cq, + ::grpc::ServerCompletionQueue* notification_cq, + void* tag) { + ::grpc::Service::RequestAsyncServerStreaming( + 2, context, request, writer, new_call_cq, notification_cq, tag); } }; template class WithRawMethod_MethodA4 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: - WithRawMethod_MethodA4() { - ::grpc::Service::MarkMethodRaw(3); - } + WithRawMethod_MethodA4() { ::grpc::Service::MarkMethodRaw(3); } ~WithRawMethod_MethodA4() override { BaseClassMustBeDerivedFromService(this); } // disable synchronous version of this method - ::grpc::Status MethodA4(::grpc::ServerContext* context, ::grpc::ServerReaderWriter< ::grpc::testing::Response, ::grpc::testing::Request>* stream) override { + ::grpc::Status MethodA4( + ::grpc::ServerContext* context, + ::grpc::ServerReaderWriter<::grpc::testing::Response, + ::grpc::testing::Request>* stream) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } - void RequestMethodA4(::grpc::ServerContext* context, ::grpc::ServerAsyncReaderWriter< ::grpc::ByteBuffer, ::grpc::ByteBuffer>* stream, ::grpc::CompletionQueue* new_call_cq, ::grpc::ServerCompletionQueue* notification_cq, void *tag) { - ::grpc::Service::RequestAsyncBidiStreaming(3, context, stream, new_call_cq, notification_cq, tag); + void RequestMethodA4( + ::grpc::ServerContext* context, + ::grpc::ServerAsyncReaderWriter<::grpc::ByteBuffer, ::grpc::ByteBuffer>* + stream, + ::grpc::CompletionQueue* new_call_cq, + ::grpc::ServerCompletionQueue* notification_cq, void* tag) { + ::grpc::Service::RequestAsyncBidiStreaming( + 3, context, stream, new_call_cq, notification_cq, tag); } }; template class ExperimentalWithRawCallbackMethod_MethodA1 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: ExperimentalWithRawCallbackMethod_MethodA1() { - ::grpc::Service::experimental().MarkMethodRawCallback(0, - new ::grpc::internal::CallbackUnaryHandler< ExperimentalWithRawCallbackMethod_MethodA1, ::grpc::ByteBuffer, ::grpc::ByteBuffer>( - [this](::grpc::ServerContext* context, - const ::grpc::ByteBuffer* request, - ::grpc::ByteBuffer* response, - ::grpc::experimental::ServerCallbackRpcController* controller) { + ::grpc::Service::experimental().MarkMethodRawCallback( + 0, new ::grpc::internal::CallbackUnaryHandler< + ExperimentalWithRawCallbackMethod_MethodA1, + ::grpc::ByteBuffer, ::grpc::ByteBuffer>( + [this](::grpc::ServerContext* context, + const ::grpc::ByteBuffer* request, + ::grpc::ByteBuffer* response, + ::grpc::experimental::ServerCallbackRpcController* + controller) { this->MethodA1(context, request, response, controller); - }, this)); + }, + this)); } ~ExperimentalWithRawCallbackMethod_MethodA1() override { BaseClassMustBeDerivedFromService(this); } // disable synchronous version of this method - ::grpc::Status MethodA1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response) override { + ::grpc::Status MethodA1(::grpc::ServerContext* context, + const ::grpc::testing::Request* request, + ::grpc::testing::Response* response) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } - virtual void MethodA1(::grpc::ServerContext* context, const ::grpc::ByteBuffer* request, ::grpc::ByteBuffer* response, ::grpc::experimental::ServerCallbackRpcController* controller) { controller->Finish(::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "")); } + virtual void MethodA1( + ::grpc::ServerContext* context, const ::grpc::ByteBuffer* request, + ::grpc::ByteBuffer* response, + ::grpc::experimental::ServerCallbackRpcController* controller) { + controller->Finish(::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "")); + } }; template class ExperimentalWithRawCallbackMethod_MethodA2 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: - ExperimentalWithRawCallbackMethod_MethodA2() { - } + ExperimentalWithRawCallbackMethod_MethodA2() {} ~ExperimentalWithRawCallbackMethod_MethodA2() override { BaseClassMustBeDerivedFromService(this); } // disable synchronous version of this method - ::grpc::Status MethodA2(::grpc::ServerContext* context, ::grpc::ServerReader< ::grpc::testing::Request>* reader, ::grpc::testing::Response* response) override { + ::grpc::Status MethodA2( + ::grpc::ServerContext* context, + ::grpc::ServerReader<::grpc::testing::Request>* reader, + ::grpc::testing::Response* response) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } @@ -581,15 +902,17 @@ class ServiceA final { template class ExperimentalWithRawCallbackMethod_MethodA3 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: - ExperimentalWithRawCallbackMethod_MethodA3() { - } + ExperimentalWithRawCallbackMethod_MethodA3() {} ~ExperimentalWithRawCallbackMethod_MethodA3() override { BaseClassMustBeDerivedFromService(this); } // disable synchronous version of this method - ::grpc::Status MethodA3(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::ServerWriter< ::grpc::testing::Response>* writer) override { + ::grpc::Status MethodA3( + ::grpc::ServerContext* context, const ::grpc::testing::Request* request, + ::grpc::ServerWriter<::grpc::testing::Response>* writer) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } @@ -597,15 +920,18 @@ class ServiceA final { template class ExperimentalWithRawCallbackMethod_MethodA4 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: - ExperimentalWithRawCallbackMethod_MethodA4() { - } + ExperimentalWithRawCallbackMethod_MethodA4() {} ~ExperimentalWithRawCallbackMethod_MethodA4() override { BaseClassMustBeDerivedFromService(this); } // disable synchronous version of this method - ::grpc::Status MethodA4(::grpc::ServerContext* context, ::grpc::ServerReaderWriter< ::grpc::testing::Response, ::grpc::testing::Request>* stream) override { + ::grpc::Status MethodA4( + ::grpc::ServerContext* context, + ::grpc::ServerReaderWriter<::grpc::testing::Response, + ::grpc::testing::Request>* stream) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } @@ -613,46 +939,69 @@ class ServiceA final { template class WithStreamedUnaryMethod_MethodA1 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: WithStreamedUnaryMethod_MethodA1() { - ::grpc::Service::MarkMethodStreamed(0, - new ::grpc::internal::StreamedUnaryHandler< ::grpc::testing::Request, ::grpc::testing::Response>(std::bind(&WithStreamedUnaryMethod_MethodA1::StreamedMethodA1, this, std::placeholders::_1, std::placeholders::_2))); + ::grpc::Service::MarkMethodStreamed( + 0, new ::grpc::internal::StreamedUnaryHandler< + ::grpc::testing::Request, ::grpc::testing::Response>(std::bind( + &WithStreamedUnaryMethod_MethodA1::StreamedMethodA1, + this, std::placeholders::_1, std::placeholders::_2))); } ~WithStreamedUnaryMethod_MethodA1() override { BaseClassMustBeDerivedFromService(this); } // disable regular version of this method - ::grpc::Status MethodA1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response) override { + ::grpc::Status MethodA1(::grpc::ServerContext* context, + const ::grpc::testing::Request* request, + ::grpc::testing::Response* response) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } // replace default version of method with streamed unary - virtual ::grpc::Status StreamedMethodA1(::grpc::ServerContext* context, ::grpc::ServerUnaryStreamer< ::grpc::testing::Request,::grpc::testing::Response>* server_unary_streamer) = 0; + virtual ::grpc::Status StreamedMethodA1( + ::grpc::ServerContext* context, + ::grpc::ServerUnaryStreamer<::grpc::testing::Request, + ::grpc::testing::Response>* + server_unary_streamer) = 0; }; - typedef WithStreamedUnaryMethod_MethodA1 StreamedUnaryService; + typedef WithStreamedUnaryMethod_MethodA1 StreamedUnaryService; template class WithSplitStreamingMethod_MethodA3 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: WithSplitStreamingMethod_MethodA3() { - ::grpc::Service::MarkMethodStreamed(2, - new ::grpc::internal::SplitServerStreamingHandler< ::grpc::testing::Request, ::grpc::testing::Response>(std::bind(&WithSplitStreamingMethod_MethodA3::StreamedMethodA3, this, std::placeholders::_1, std::placeholders::_2))); + ::grpc::Service::MarkMethodStreamed( + 2, + new ::grpc::internal::SplitServerStreamingHandler< + ::grpc::testing::Request, ::grpc::testing::Response>(std::bind( + &WithSplitStreamingMethod_MethodA3::StreamedMethodA3, + this, std::placeholders::_1, std::placeholders::_2))); } ~WithSplitStreamingMethod_MethodA3() override { BaseClassMustBeDerivedFromService(this); } // disable regular version of this method - ::grpc::Status MethodA3(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::ServerWriter< ::grpc::testing::Response>* writer) override { + ::grpc::Status MethodA3( + ::grpc::ServerContext* context, const ::grpc::testing::Request* request, + ::grpc::ServerWriter<::grpc::testing::Response>* writer) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } // replace default version of method with split streamed - virtual ::grpc::Status StreamedMethodA3(::grpc::ServerContext* context, ::grpc::ServerSplitStreamer< ::grpc::testing::Request,::grpc::testing::Response>* server_split_streamer) = 0; + virtual ::grpc::Status StreamedMethodA3( + ::grpc::ServerContext* context, + ::grpc::ServerSplitStreamer<::grpc::testing::Request, + ::grpc::testing::Response>* + server_split_streamer) = 0; }; - typedef WithSplitStreamingMethod_MethodA3 SplitStreamedService; - typedef WithStreamedUnaryMethod_MethodA1 > StreamedService; + typedef WithSplitStreamingMethod_MethodA3 SplitStreamedService; + typedef WithStreamedUnaryMethod_MethodA1< + WithSplitStreamingMethod_MethodA3> + StreamedService; }; // ServiceB leading comment 1 @@ -665,125 +1014,204 @@ class ServiceB final { public: virtual ~StubInterface() {} // MethodB1 leading comment 1 - virtual ::grpc::Status MethodB1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::testing::Response* response) = 0; - std::unique_ptr< ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>> AsyncMethodB1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) { - return std::unique_ptr< ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>>(AsyncMethodB1Raw(context, request, cq)); - } - std::unique_ptr< ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>> PrepareAsyncMethodB1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) { - return std::unique_ptr< ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>>(PrepareAsyncMethodB1Raw(context, request, cq)); + virtual ::grpc::Status MethodB1(::grpc::ClientContext* context, + const ::grpc::testing::Request& request, + ::grpc::testing::Response* response) = 0; + std::unique_ptr< + ::grpc::ClientAsyncResponseReaderInterface<::grpc::testing::Response>> + AsyncMethodB1(::grpc::ClientContext* context, + const ::grpc::testing::Request& request, + ::grpc::CompletionQueue* cq) { + return std::unique_ptr<::grpc::ClientAsyncResponseReaderInterface< + ::grpc::testing::Response>>(AsyncMethodB1Raw(context, request, cq)); + } + std::unique_ptr< + ::grpc::ClientAsyncResponseReaderInterface<::grpc::testing::Response>> + PrepareAsyncMethodB1(::grpc::ClientContext* context, + const ::grpc::testing::Request& request, + ::grpc::CompletionQueue* cq) { + return std::unique_ptr<::grpc::ClientAsyncResponseReaderInterface< + ::grpc::testing::Response>>( + PrepareAsyncMethodB1Raw(context, request, cq)); } // MethodB1 trailing comment 1 class experimental_async_interface { public: virtual ~experimental_async_interface() {} // MethodB1 leading comment 1 - virtual void MethodB1(::grpc::ClientContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response, std::function) = 0; + virtual void MethodB1(::grpc::ClientContext* context, + const ::grpc::testing::Request* request, + ::grpc::testing::Response* response, + std::function) = 0; // MethodB1 trailing comment 1 }; - virtual class experimental_async_interface* experimental_async() { return nullptr; } - private: - virtual ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>* AsyncMethodB1Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) = 0; - virtual ::grpc::ClientAsyncResponseReaderInterface< ::grpc::testing::Response>* PrepareAsyncMethodB1Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) = 0; + virtual class experimental_async_interface* experimental_async() { + return nullptr; + } + + private: + virtual ::grpc::ClientAsyncResponseReaderInterface< + ::grpc::testing::Response>* + AsyncMethodB1Raw(::grpc::ClientContext* context, + const ::grpc::testing::Request& request, + ::grpc::CompletionQueue* cq) = 0; + virtual ::grpc::ClientAsyncResponseReaderInterface< + ::grpc::testing::Response>* + PrepareAsyncMethodB1Raw(::grpc::ClientContext* context, + const ::grpc::testing::Request& request, + ::grpc::CompletionQueue* cq) = 0; }; class Stub final : public StubInterface { public: - Stub(const std::shared_ptr< ::grpc::ChannelInterface>& channel); - ::grpc::Status MethodB1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::testing::Response* response) override; - std::unique_ptr< ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>> AsyncMethodB1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) { - return std::unique_ptr< ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>>(AsyncMethodB1Raw(context, request, cq)); - } - std::unique_ptr< ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>> PrepareAsyncMethodB1(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) { - return std::unique_ptr< ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>>(PrepareAsyncMethodB1Raw(context, request, cq)); - } - class experimental_async final : - public StubInterface::experimental_async_interface { + Stub(const std::shared_ptr<::grpc::ChannelInterface>& channel); + ::grpc::Status MethodB1(::grpc::ClientContext* context, + const ::grpc::testing::Request& request, + ::grpc::testing::Response* response) override; + std::unique_ptr< + ::grpc::ClientAsyncResponseReader<::grpc::testing::Response>> + AsyncMethodB1(::grpc::ClientContext* context, + const ::grpc::testing::Request& request, + ::grpc::CompletionQueue* cq) { + return std::unique_ptr< + ::grpc::ClientAsyncResponseReader<::grpc::testing::Response>>( + AsyncMethodB1Raw(context, request, cq)); + } + std::unique_ptr< + ::grpc::ClientAsyncResponseReader<::grpc::testing::Response>> + PrepareAsyncMethodB1(::grpc::ClientContext* context, + const ::grpc::testing::Request& request, + ::grpc::CompletionQueue* cq) { + return std::unique_ptr< + ::grpc::ClientAsyncResponseReader<::grpc::testing::Response>>( + PrepareAsyncMethodB1Raw(context, request, cq)); + } + class experimental_async final + : public StubInterface::experimental_async_interface { public: - void MethodB1(::grpc::ClientContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response, std::function) override; + void MethodB1(::grpc::ClientContext* context, + const ::grpc::testing::Request* request, + ::grpc::testing::Response* response, + std::function) override; + private: friend class Stub; - explicit experimental_async(Stub* stub): stub_(stub) { } + explicit experimental_async(Stub* stub) : stub_(stub) {} Stub* stub() { return stub_; } Stub* stub_; }; - class experimental_async_interface* experimental_async() override { return &async_stub_; } + class experimental_async_interface* experimental_async() override { + return &async_stub_; + } private: - std::shared_ptr< ::grpc::ChannelInterface> channel_; - class experimental_async async_stub_{this}; - ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>* AsyncMethodB1Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) override; - ::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>* PrepareAsyncMethodB1Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) override; + std::shared_ptr<::grpc::ChannelInterface> channel_; + class experimental_async async_stub_ { + this + }; + ::grpc::ClientAsyncResponseReader<::grpc::testing::Response>* + AsyncMethodB1Raw(::grpc::ClientContext* context, + const ::grpc::testing::Request& request, + ::grpc::CompletionQueue* cq) override; + ::grpc::ClientAsyncResponseReader<::grpc::testing::Response>* + PrepareAsyncMethodB1Raw(::grpc::ClientContext* context, + const ::grpc::testing::Request& request, + ::grpc::CompletionQueue* cq) override; const ::grpc::internal::RpcMethod rpcmethod_MethodB1_; }; - static std::unique_ptr NewStub(const std::shared_ptr< ::grpc::ChannelInterface>& channel, const ::grpc::StubOptions& options = ::grpc::StubOptions()); + static std::unique_ptr NewStub( + const std::shared_ptr<::grpc::ChannelInterface>& channel, + const ::grpc::StubOptions& options = ::grpc::StubOptions()); class Service : public ::grpc::Service { public: Service(); virtual ~Service(); // MethodB1 leading comment 1 - virtual ::grpc::Status MethodB1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response); + virtual ::grpc::Status MethodB1(::grpc::ServerContext* context, + const ::grpc::testing::Request* request, + ::grpc::testing::Response* response); // MethodB1 trailing comment 1 }; template class WithAsyncMethod_MethodB1 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: - WithAsyncMethod_MethodB1() { - ::grpc::Service::MarkMethodAsync(0); - } + WithAsyncMethod_MethodB1() { ::grpc::Service::MarkMethodAsync(0); } ~WithAsyncMethod_MethodB1() override { BaseClassMustBeDerivedFromService(this); } // disable synchronous version of this method - ::grpc::Status MethodB1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response) override { + ::grpc::Status MethodB1(::grpc::ServerContext* context, + const ::grpc::testing::Request* request, + ::grpc::testing::Response* response) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } - void RequestMethodB1(::grpc::ServerContext* context, ::grpc::testing::Request* request, ::grpc::ServerAsyncResponseWriter< ::grpc::testing::Response>* response, ::grpc::CompletionQueue* new_call_cq, ::grpc::ServerCompletionQueue* notification_cq, void *tag) { - ::grpc::Service::RequestAsyncUnary(0, context, request, response, new_call_cq, notification_cq, tag); + void RequestMethodB1( + ::grpc::ServerContext* context, ::grpc::testing::Request* request, + ::grpc::ServerAsyncResponseWriter<::grpc::testing::Response>* response, + ::grpc::CompletionQueue* new_call_cq, + ::grpc::ServerCompletionQueue* notification_cq, void* tag) { + ::grpc::Service::RequestAsyncUnary(0, context, request, response, + new_call_cq, notification_cq, tag); } }; - typedef WithAsyncMethod_MethodB1 AsyncService; + typedef WithAsyncMethod_MethodB1 AsyncService; template class ExperimentalWithCallbackMethod_MethodB1 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: ExperimentalWithCallbackMethod_MethodB1() { - ::grpc::Service::experimental().MarkMethodCallback(0, - new ::grpc::internal::CallbackUnaryHandler< ExperimentalWithCallbackMethod_MethodB1, ::grpc::testing::Request, ::grpc::testing::Response>( - [this](::grpc::ServerContext* context, - const ::grpc::testing::Request* request, - ::grpc::testing::Response* response, - ::grpc::experimental::ServerCallbackRpcController* controller) { + ::grpc::Service::experimental().MarkMethodCallback( + 0, new ::grpc::internal::CallbackUnaryHandler< + ExperimentalWithCallbackMethod_MethodB1, + ::grpc::testing::Request, ::grpc::testing::Response>( + [this](::grpc::ServerContext* context, + const ::grpc::testing::Request* request, + ::grpc::testing::Response* response, + ::grpc::experimental::ServerCallbackRpcController* + controller) { this->MethodB1(context, request, response, controller); - }, this)); + }, + this)); } ~ExperimentalWithCallbackMethod_MethodB1() override { BaseClassMustBeDerivedFromService(this); } // disable synchronous version of this method - ::grpc::Status MethodB1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response) override { + ::grpc::Status MethodB1(::grpc::ServerContext* context, + const ::grpc::testing::Request* request, + ::grpc::testing::Response* response) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } - virtual void MethodB1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response, ::grpc::experimental::ServerCallbackRpcController* controller) { controller->Finish(::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "")); } + virtual void MethodB1( + ::grpc::ServerContext* context, const ::grpc::testing::Request* request, + ::grpc::testing::Response* response, + ::grpc::experimental::ServerCallbackRpcController* controller) { + controller->Finish(::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "")); + } }; - typedef ExperimentalWithCallbackMethod_MethodB1 ExperimentalCallbackService; + typedef ExperimentalWithCallbackMethod_MethodB1 + ExperimentalCallbackService; template class WithGenericMethod_MethodB1 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: - WithGenericMethod_MethodB1() { - ::grpc::Service::MarkMethodGeneric(0); - } + WithGenericMethod_MethodB1() { ::grpc::Service::MarkMethodGeneric(0); } ~WithGenericMethod_MethodB1() override { BaseClassMustBeDerivedFromService(this); } // disable synchronous version of this method - ::grpc::Status MethodB1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response) override { + ::grpc::Status MethodB1(::grpc::ServerContext* context, + const ::grpc::testing::Request* request, + ::grpc::testing::Response* response) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } @@ -791,76 +1219,103 @@ class ServiceB final { template class WithRawMethod_MethodB1 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: - WithRawMethod_MethodB1() { - ::grpc::Service::MarkMethodRaw(0); - } + WithRawMethod_MethodB1() { ::grpc::Service::MarkMethodRaw(0); } ~WithRawMethod_MethodB1() override { BaseClassMustBeDerivedFromService(this); } // disable synchronous version of this method - ::grpc::Status MethodB1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response) override { + ::grpc::Status MethodB1(::grpc::ServerContext* context, + const ::grpc::testing::Request* request, + ::grpc::testing::Response* response) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } - void RequestMethodB1(::grpc::ServerContext* context, ::grpc::ByteBuffer* request, ::grpc::ServerAsyncResponseWriter< ::grpc::ByteBuffer>* response, ::grpc::CompletionQueue* new_call_cq, ::grpc::ServerCompletionQueue* notification_cq, void *tag) { - ::grpc::Service::RequestAsyncUnary(0, context, request, response, new_call_cq, notification_cq, tag); + void RequestMethodB1( + ::grpc::ServerContext* context, ::grpc::ByteBuffer* request, + ::grpc::ServerAsyncResponseWriter<::grpc::ByteBuffer>* response, + ::grpc::CompletionQueue* new_call_cq, + ::grpc::ServerCompletionQueue* notification_cq, void* tag) { + ::grpc::Service::RequestAsyncUnary(0, context, request, response, + new_call_cq, notification_cq, tag); } }; template class ExperimentalWithRawCallbackMethod_MethodB1 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: ExperimentalWithRawCallbackMethod_MethodB1() { - ::grpc::Service::experimental().MarkMethodRawCallback(0, - new ::grpc::internal::CallbackUnaryHandler< ExperimentalWithRawCallbackMethod_MethodB1, ::grpc::ByteBuffer, ::grpc::ByteBuffer>( - [this](::grpc::ServerContext* context, - const ::grpc::ByteBuffer* request, - ::grpc::ByteBuffer* response, - ::grpc::experimental::ServerCallbackRpcController* controller) { + ::grpc::Service::experimental().MarkMethodRawCallback( + 0, new ::grpc::internal::CallbackUnaryHandler< + ExperimentalWithRawCallbackMethod_MethodB1, + ::grpc::ByteBuffer, ::grpc::ByteBuffer>( + [this](::grpc::ServerContext* context, + const ::grpc::ByteBuffer* request, + ::grpc::ByteBuffer* response, + ::grpc::experimental::ServerCallbackRpcController* + controller) { this->MethodB1(context, request, response, controller); - }, this)); + }, + this)); } ~ExperimentalWithRawCallbackMethod_MethodB1() override { BaseClassMustBeDerivedFromService(this); } // disable synchronous version of this method - ::grpc::Status MethodB1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response) override { + ::grpc::Status MethodB1(::grpc::ServerContext* context, + const ::grpc::testing::Request* request, + ::grpc::testing::Response* response) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } - virtual void MethodB1(::grpc::ServerContext* context, const ::grpc::ByteBuffer* request, ::grpc::ByteBuffer* response, ::grpc::experimental::ServerCallbackRpcController* controller) { controller->Finish(::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "")); } + virtual void MethodB1( + ::grpc::ServerContext* context, const ::grpc::ByteBuffer* request, + ::grpc::ByteBuffer* response, + ::grpc::experimental::ServerCallbackRpcController* controller) { + controller->Finish(::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "")); + } }; template class WithStreamedUnaryMethod_MethodB1 : public BaseClass { private: - void BaseClassMustBeDerivedFromService(const Service *service) {} + void BaseClassMustBeDerivedFromService(const Service* service) {} + public: WithStreamedUnaryMethod_MethodB1() { - ::grpc::Service::MarkMethodStreamed(0, - new ::grpc::internal::StreamedUnaryHandler< ::grpc::testing::Request, ::grpc::testing::Response>(std::bind(&WithStreamedUnaryMethod_MethodB1::StreamedMethodB1, this, std::placeholders::_1, std::placeholders::_2))); + ::grpc::Service::MarkMethodStreamed( + 0, new ::grpc::internal::StreamedUnaryHandler< + ::grpc::testing::Request, ::grpc::testing::Response>(std::bind( + &WithStreamedUnaryMethod_MethodB1::StreamedMethodB1, + this, std::placeholders::_1, std::placeholders::_2))); } ~WithStreamedUnaryMethod_MethodB1() override { BaseClassMustBeDerivedFromService(this); } // disable regular version of this method - ::grpc::Status MethodB1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response) override { + ::grpc::Status MethodB1(::grpc::ServerContext* context, + const ::grpc::testing::Request* request, + ::grpc::testing::Response* response) override { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } // replace default version of method with streamed unary - virtual ::grpc::Status StreamedMethodB1(::grpc::ServerContext* context, ::grpc::ServerUnaryStreamer< ::grpc::testing::Request,::grpc::testing::Response>* server_unary_streamer) = 0; + virtual ::grpc::Status StreamedMethodB1( + ::grpc::ServerContext* context, + ::grpc::ServerUnaryStreamer<::grpc::testing::Request, + ::grpc::testing::Response>* + server_unary_streamer) = 0; }; - typedef WithStreamedUnaryMethod_MethodB1 StreamedUnaryService; + typedef WithStreamedUnaryMethod_MethodB1 StreamedUnaryService; typedef Service SplitStreamedService; - typedef WithStreamedUnaryMethod_MethodB1 StreamedService; + typedef WithStreamedUnaryMethod_MethodB1 StreamedService; }; // ServiceB trailing comment 1 } // namespace testing } // namespace grpc - #endif // GRPC_src_2fproto_2fgrpc_2ftesting_2fcompiler_5ftest_2eproto__INCLUDED diff --git a/test/cpp/end2end/client_callback_end2end_test.cc b/test/cpp/end2end/client_callback_end2end_test.cc index 57c3cb87f2..a1fe199b54 100644 --- a/test/cpp/end2end/client_callback_end2end_test.cc +++ b/test/cpp/end2end/client_callback_end2end_test.cc @@ -187,20 +187,20 @@ class ClientCallbackEnd2endTest grpc::string test_string(""); for (int i = 0; i < num_rpcs; i++) { test_string += "Hello world. "; - class Client : public grpc::experimental::ClientBidiReactor { + class Client : public grpc::experimental::ClientBidiReactor { public: Client(ClientCallbackEnd2endTest* test, const grpc::string& method_name, const grpc::string& test_str) { - stream_ = - test->generic_stub_->experimental().PrepareBidiStreamingCall( - &cli_ctx_, method_name, this); + test->generic_stub_->experimental().PrepareBidiStreamingCall( + &cli_ctx_, method_name, this); request_.set_message(test_str); send_buf_ = SerializeToByteBuffer(&request_); - stream_->Write(send_buf_.get()); - stream_->Read(&recv_buf_); - stream_->StartCall(); + StartWrite(send_buf_.get()); + StartRead(&recv_buf_); + StartCall(); } - void OnWriteDone(bool ok) override { stream_->WritesDone(); } + void OnWriteDone(bool ok) override { StartWritesDone(); } void OnReadDone(bool ok) override { EchoResponse response; EXPECT_TRUE(ParseFromByteBuffer(&recv_buf_, &response)); @@ -223,8 +223,6 @@ class ClientCallbackEnd2endTest std::unique_ptr send_buf_; ByteBuffer recv_buf_; ClientContext cli_ctx_; - experimental::ClientCallbackReaderWriter* - stream_; std::mutex mu_; std::condition_variable cv_; bool done_ = false; @@ -330,22 +328,21 @@ TEST_P(ClientCallbackEnd2endTest, RequestStream) { } ResetStub(); - class Client : public grpc::experimental::ClientWriteReactor { + class Client : public grpc::experimental::ClientWriteReactor { public: explicit Client(grpc::testing::EchoTestService::Stub* stub) { context_.set_initial_metadata_corked(true); - stream_ = stub->experimental_async()->RequestStream(&context_, &response_, - this); - stream_->StartCall(); + stub->experimental_async()->RequestStream(&context_, &response_, this); + StartCall(); request_.set_message("Hello server."); - stream_->Write(&request_); + StartWrite(&request_); } void OnWriteDone(bool ok) override { writes_left_--; if (writes_left_ > 1) { - stream_->Write(&request_); + StartWrite(&request_); } else if (writes_left_ == 1) { - stream_->WriteLast(&request_, WriteOptions()); + StartWriteLast(&request_, WriteOptions()); } } void OnDone(Status s) override { @@ -363,7 +360,6 @@ TEST_P(ClientCallbackEnd2endTest, RequestStream) { } private: - ::grpc::experimental::ClientCallbackWriter* stream_; EchoRequest request_; EchoResponse response_; ClientContext context_; @@ -383,14 +379,13 @@ TEST_P(ClientCallbackEnd2endTest, ResponseStream) { } ResetStub(); - class Client : public grpc::experimental::ClientReadReactor { + class Client : public grpc::experimental::ClientReadReactor { public: explicit Client(grpc::testing::EchoTestService::Stub* stub) { request_.set_message("Hello client "); - stream_ = stub->experimental_async()->ResponseStream(&context_, &request_, - this); - stream_->StartCall(); - stream_->Read(&response_); + stub->experimental_async()->ResponseStream(&context_, &request_, this); + StartCall(); + StartRead(&response_); } void OnReadDone(bool ok) override { if (!ok) { @@ -400,7 +395,7 @@ TEST_P(ClientCallbackEnd2endTest, ResponseStream) { EXPECT_EQ(response_.message(), request_.message() + grpc::to_string(reads_complete_)); reads_complete_++; - stream_->Read(&response_); + StartRead(&response_); } } void OnDone(Status s) override { @@ -417,7 +412,6 @@ TEST_P(ClientCallbackEnd2endTest, ResponseStream) { } private: - ::grpc::experimental::ClientCallbackReader* stream_; EchoRequest request_; EchoResponse response_; ClientContext context_; @@ -436,14 +430,15 @@ TEST_P(ClientCallbackEnd2endTest, BidiStream) { return; } ResetStub(); - class Client : public grpc::experimental::ClientBidiReactor { + class Client : public grpc::experimental::ClientBidiReactor { public: explicit Client(grpc::testing::EchoTestService::Stub* stub) { request_.set_message("Hello fren "); - stream_ = stub->experimental_async()->BidiStream(&context_, this); - stream_->StartCall(); - stream_->Read(&response_); - stream_->Write(&request_); + stub->experimental_async()->BidiStream(&context_, this); + StartCall(); + StartRead(&response_); + StartWrite(&request_); } void OnReadDone(bool ok) override { if (!ok) { @@ -452,15 +447,15 @@ TEST_P(ClientCallbackEnd2endTest, BidiStream) { EXPECT_LE(reads_complete_, kServerDefaultResponseStreamsToSend); EXPECT_EQ(response_.message(), request_.message()); reads_complete_++; - stream_->Read(&response_); + StartRead(&response_); } } void OnWriteDone(bool ok) override { EXPECT_TRUE(ok); if (++writes_complete_ == kServerDefaultResponseStreamsToSend) { - stream_->WritesDone(); + StartWritesDone(); } else { - stream_->Write(&request_); + StartWrite(&request_); } } void OnDone(Status s) override { @@ -477,8 +472,6 @@ TEST_P(ClientCallbackEnd2endTest, BidiStream) { } private: - ::grpc::experimental::ClientCallbackReaderWriter* - stream_; EchoRequest request_; EchoResponse response_; ClientContext context_; -- cgit v1.2.3 From 2a0c0d7ad6a1256ef6c0398e1900eca2be077a51 Mon Sep 17 00:00:00 2001 From: Vijay Pai Date: Mon, 5 Nov 2018 14:39:44 -0800 Subject: Streaming API for callback servers --- include/grpcpp/impl/codegen/byte_buffer.h | 8 +- include/grpcpp/impl/codegen/callback_common.h | 19 +- include/grpcpp/impl/codegen/server_callback.h | 774 ++++++++++++++++++++++++-- include/grpcpp/impl/codegen/server_context.h | 21 +- src/compiler/cpp_generator.cc | 69 ++- src/cpp/server/server_cc.cc | 7 +- src/cpp/server/server_context.cc | 47 +- test/cpp/codegen/compiler_test_golden | 56 +- test/cpp/end2end/end2end_test.cc | 90 ++- test/cpp/end2end/test_service_impl.cc | 314 +++++++++-- test/cpp/end2end/test_service_impl.h | 21 +- 11 files changed, 1261 insertions(+), 165 deletions(-) (limited to 'src/compiler') diff --git a/include/grpcpp/impl/codegen/byte_buffer.h b/include/grpcpp/impl/codegen/byte_buffer.h index abba5549b8..53ecb53371 100644 --- a/include/grpcpp/impl/codegen/byte_buffer.h +++ b/include/grpcpp/impl/codegen/byte_buffer.h @@ -45,8 +45,10 @@ template class RpcMethodHandler; template class ServerStreamingHandler; -template +template class CallbackUnaryHandler; +template +class CallbackServerStreamingHandler; template class ErrorMethodHandler; template @@ -156,8 +158,10 @@ class ByteBuffer final { friend class internal::RpcMethodHandler; template friend class internal::ServerStreamingHandler; - template + template friend class internal::CallbackUnaryHandler; + template + friend class ::grpc::internal::CallbackServerStreamingHandler; template friend class internal::ErrorMethodHandler; template diff --git a/include/grpcpp/impl/codegen/callback_common.h b/include/grpcpp/impl/codegen/callback_common.h index f7a24204dc..a3c8c41246 100644 --- a/include/grpcpp/impl/codegen/callback_common.h +++ b/include/grpcpp/impl/codegen/callback_common.h @@ -32,6 +32,8 @@ namespace grpc { namespace internal { /// An exception-safe way of invoking a user-specified callback function +// TODO(vjpai): decide whether it is better for this to take a const lvalue +// parameter or an rvalue parameter, or if it even matters template void CatchingCallback(Func&& func, Args&&... args) { #if GRPC_ALLOW_EXCEPTIONS @@ -45,6 +47,20 @@ void CatchingCallback(Func&& func, Args&&... args) { #endif // GRPC_ALLOW_EXCEPTIONS } +template +ReturnType* CatchingReactorCreator(Func&& func, Args&&... args) { +#if GRPC_ALLOW_EXCEPTIONS + try { + return func(std::forward(args)...); + } catch (...) { + // fail the RPC, don't crash the library + return nullptr; + } +#else // GRPC_ALLOW_EXCEPTIONS + return func(std::forward(args)...); +#endif // GRPC_ALLOW_EXCEPTIONS +} + // The contract on these tags is that they are single-shot. They must be // constructed and then fired at exactly one point. There is no expectation // that they can be reused without reconstruction. @@ -185,8 +201,9 @@ class CallbackWithSuccessTag void* ignored = ops_; // Allow a "false" return value from FinalizeResult to silence the // callback, just as it silences a CQ tag in the async cases + auto* ops = ops_; bool do_callback = ops_->FinalizeResult(&ignored, &ok); - GPR_CODEGEN_ASSERT(ignored == ops_); + GPR_CODEGEN_ASSERT(ignored == ops); if (do_callback) { CatchingCallback(func_, ok); diff --git a/include/grpcpp/impl/codegen/server_callback.h b/include/grpcpp/impl/codegen/server_callback.h index b866fc16dc..1854f6ef2f 100644 --- a/include/grpcpp/impl/codegen/server_callback.h +++ b/include/grpcpp/impl/codegen/server_callback.h @@ -19,7 +19,9 @@ #ifndef GRPCPP_IMPL_CODEGEN_SERVER_CALLBACK_H #define GRPCPP_IMPL_CODEGEN_SERVER_CALLBACK_H +#include #include +#include #include #include @@ -32,19 +34,33 @@ namespace grpc { -// forward declarations +// Declare base class of all reactors as internal namespace internal { -template -class CallbackUnaryHandler; + +class ServerReactor { + public: + virtual ~ServerReactor() = default; + virtual void OnDone() {} + virtual void OnCancel() {} +}; + } // namespace internal namespace experimental { +// Forward declarations +template +class ServerReadReactor; +template +class ServerWriteReactor; +template +class ServerBidiReactor; + // For unary RPCs, the exposed controller class is only an interface // and the actual implementation is an internal class. class ServerCallbackRpcController { public: - virtual ~ServerCallbackRpcController() {} + virtual ~ServerCallbackRpcController() = default; // The method handler must call this function when it is done so that // the library knows to free its resources @@ -55,18 +71,193 @@ class ServerCallbackRpcController { virtual void SendInitialMetadata(std::function) = 0; }; +// NOTE: The actual streaming object classes are provided +// as API only to support mocking. There are no implementations of +// these class interfaces in the API. +template +class ServerCallbackReader { + public: + virtual ~ServerCallbackReader() {} + virtual void Finish(Status s) = 0; + virtual void SendInitialMetadata() = 0; + virtual void Read(Request* msg) = 0; + + protected: + template + void BindReactor(ServerReadReactor* reactor) { + reactor->BindReader(this); + } +}; + +template +class ServerCallbackWriter { + public: + virtual ~ServerCallbackWriter() {} + + virtual void Finish(Status s) = 0; + virtual void SendInitialMetadata() = 0; + virtual void Write(const Response* msg, WriteOptions options) = 0; + virtual void WriteAndFinish(const Response* msg, WriteOptions options, + Status s) { + // Default implementation that can/should be overridden + Write(msg, std::move(options)); + Finish(std::move(s)); + }; + + protected: + template + void BindReactor(ServerWriteReactor* reactor) { + reactor->BindWriter(this); + } +}; + +template +class ServerCallbackReaderWriter { + public: + virtual ~ServerCallbackReaderWriter() {} + + virtual void Finish(Status s) = 0; + virtual void SendInitialMetadata() = 0; + virtual void Read(Request* msg) = 0; + virtual void Write(const Response* msg, WriteOptions options) = 0; + virtual void WriteAndFinish(const Response* msg, WriteOptions options, + Status s) { + // Default implementation that can/should be overridden + Write(msg, std::move(options)); + Finish(std::move(s)); + }; + + protected: + void BindReactor(ServerBidiReactor* reactor) { + reactor->BindStream(this); + } +}; + +// The following classes are reactors that are to be implemented +// by the user, returned as the result of the method handler for +// a callback method, and activated by the call to OnStarted +template +class ServerBidiReactor : public internal::ServerReactor { + public: + ~ServerBidiReactor() = default; + virtual void OnStarted(ServerContext*) {} + virtual void OnSendInitialMetadataDone(bool ok) {} + virtual void OnReadDone(bool ok) {} + virtual void OnWriteDone(bool ok) {} + + void StartSendInitialMetadata() { stream_->SendInitialMetadata(); } + void StartRead(Request* msg) { stream_->Read(msg); } + void StartWrite(const Response* msg) { StartWrite(msg, WriteOptions()); } + void StartWrite(const Response* msg, WriteOptions options) { + stream_->Write(msg, std::move(options)); + } + void StartWriteAndFinish(const Response* msg, WriteOptions options, + Status s) { + stream_->WriteAndFinish(msg, std::move(options), std::move(s)); + } + void StartWriteLast(const Response* msg, WriteOptions options) { + StartWrite(msg, std::move(options.set_last_message())); + } + void Finish(Status s) { stream_->Finish(std::move(s)); } + + private: + friend class ServerCallbackReaderWriter; + void BindStream(ServerCallbackReaderWriter* stream) { + stream_ = stream; + } + + ServerCallbackReaderWriter* stream_; +}; + +template +class ServerReadReactor : public internal::ServerReactor { + public: + ~ServerReadReactor() = default; + virtual void OnStarted(ServerContext*, Response* resp) {} + virtual void OnSendInitialMetadataDone(bool ok) {} + virtual void OnReadDone(bool ok) {} + + void StartSendInitialMetadata() { reader_->SendInitialMetadata(); } + void StartRead(Request* msg) { reader_->Read(msg); } + void Finish(Status s) { reader_->Finish(std::move(s)); } + + private: + friend class ServerCallbackReader; + void BindReader(ServerCallbackReader* reader) { reader_ = reader; } + + ServerCallbackReader* reader_; +}; + +template +class ServerWriteReactor : public internal::ServerReactor { + public: + ~ServerWriteReactor() = default; + virtual void OnStarted(ServerContext*, const Request* req) {} + virtual void OnSendInitialMetadataDone(bool ok) {} + virtual void OnWriteDone(bool ok) {} + + void StartSendInitialMetadata() { writer_->SendInitialMetadata(); } + void StartWrite(const Response* msg) { StartWrite(msg, WriteOptions()); } + void StartWrite(const Response* msg, WriteOptions options) { + writer_->Write(msg, std::move(options)); + } + void StartWriteAndFinish(const Response* msg, WriteOptions options, + Status s) { + writer_->WriteAndFinish(msg, std::move(options), std::move(s)); + } + void StartWriteLast(const Response* msg, WriteOptions options) { + StartWrite(msg, std::move(options.set_last_message())); + } + void Finish(Status s) { writer_->Finish(std::move(s)); } + + private: + friend class ServerCallbackWriter; + void BindWriter(ServerCallbackWriter* writer) { writer_ = writer; } + + ServerCallbackWriter* writer_; +}; + } // namespace experimental namespace internal { -template +template +class UnimplementedReadReactor + : public experimental::ServerReadReactor { + public: + void OnDone() override { delete this; } + void OnStarted(ServerContext*, Response*) override { + this->Finish(Status(StatusCode::UNIMPLEMENTED, "")); + } +}; + +template +class UnimplementedWriteReactor + : public experimental::ServerWriteReactor { + public: + void OnDone() override { delete this; } + void OnStarted(ServerContext*, const Request*) override { + this->Finish(Status(StatusCode::UNIMPLEMENTED, "")); + } +}; + +template +class UnimplementedBidiReactor + : public experimental::ServerBidiReactor { + public: + void OnDone() override { delete this; } + void OnStarted(ServerContext*) override { + this->Finish(Status(StatusCode::UNIMPLEMENTED, "")); + } +}; + +template class CallbackUnaryHandler : public MethodHandler { public: CallbackUnaryHandler( std::function - func, - ServiceType* service) + func) : func_(func) {} void RunHandler(const HandlerParameter& param) final { // Arena allocate a controller structure (that includes request/response) @@ -81,9 +272,8 @@ class CallbackUnaryHandler : public MethodHandler { if (status.ok()) { // Call the actual function handler and expect the user to call finish - CatchingCallback(std::move(func_), param.server_context, - controller->request(), controller->response(), - controller); + CatchingCallback(func_, param.server_context, controller->request(), + controller->response(), controller); } else { // if deserialization failed, we need to fail the call controller->Finish(status); @@ -117,79 +307,579 @@ class CallbackUnaryHandler : public MethodHandler { : public experimental::ServerCallbackRpcController { public: void Finish(Status s) override { - finish_tag_.Set( - call_.call(), - [this](bool) { - grpc_call* call = call_.call(); - auto call_requester = std::move(call_requester_); - this->~ServerCallbackRpcControllerImpl(); // explicitly call - // destructor - g_core_codegen_interface->grpc_call_unref(call); - call_requester(); - }, - &finish_buf_); + finish_tag_.Set(call_.call(), [this](bool) { MaybeDone(); }, + &finish_ops_); if (!ctx_->sent_initial_metadata_) { - finish_buf_.SendInitialMetadata(&ctx_->initial_metadata_, + finish_ops_.SendInitialMetadata(&ctx_->initial_metadata_, ctx_->initial_metadata_flags()); if (ctx_->compression_level_set()) { - finish_buf_.set_compression_level(ctx_->compression_level()); + finish_ops_.set_compression_level(ctx_->compression_level()); } ctx_->sent_initial_metadata_ = true; } // The response is dropped if the status is not OK. if (s.ok()) { - finish_buf_.ServerSendStatus(&ctx_->trailing_metadata_, - finish_buf_.SendMessage(resp_)); + finish_ops_.ServerSendStatus(&ctx_->trailing_metadata_, + finish_ops_.SendMessage(resp_)); } else { - finish_buf_.ServerSendStatus(&ctx_->trailing_metadata_, s); + finish_ops_.ServerSendStatus(&ctx_->trailing_metadata_, s); } - finish_buf_.set_core_cq_tag(&finish_tag_); - call_.PerformOps(&finish_buf_); + finish_ops_.set_core_cq_tag(&finish_tag_); + call_.PerformOps(&finish_ops_); } void SendInitialMetadata(std::function f) override { GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_); - - meta_tag_.Set(call_.call(), std::move(f), &meta_buf_); - meta_buf_.SendInitialMetadata(&ctx_->initial_metadata_, + callbacks_outstanding_++; + // TODO(vjpai): Consider taking f as a move-capture if we adopt C++14 + // and if performance of this operation matters + meta_tag_.Set(call_.call(), + [this, f](bool ok) { + f(ok); + MaybeDone(); + }, + &meta_ops_); + meta_ops_.SendInitialMetadata(&ctx_->initial_metadata_, ctx_->initial_metadata_flags()); if (ctx_->compression_level_set()) { - meta_buf_.set_compression_level(ctx_->compression_level()); + meta_ops_.set_compression_level(ctx_->compression_level()); } ctx_->sent_initial_metadata_ = true; - meta_buf_.set_core_cq_tag(&meta_tag_); - call_.PerformOps(&meta_buf_); + meta_ops_.set_core_cq_tag(&meta_tag_); + call_.PerformOps(&meta_ops_); } private: - template - friend class CallbackUnaryHandler; + friend class CallbackUnaryHandler; ServerCallbackRpcControllerImpl(ServerContext* ctx, Call* call, - RequestType* req, + const RequestType* req, std::function call_requester) : ctx_(ctx), call_(*call), req_(req), - call_requester_(std::move(call_requester)) {} + call_requester_(std::move(call_requester)) { + ctx_->BeginCompletionOp(call, [this](bool) { MaybeDone(); }, nullptr); + } ~ServerCallbackRpcControllerImpl() { req_->~RequestType(); } - RequestType* request() { return req_; } + const RequestType* request() { return req_; } ResponseType* response() { return &resp_; } - CallOpSet meta_buf_; + void MaybeDone() { + if (--callbacks_outstanding_ == 0) { + grpc_call* call = call_.call(); + auto call_requester = std::move(call_requester_); + this->~ServerCallbackRpcControllerImpl(); // explicitly call destructor + g_core_codegen_interface->grpc_call_unref(call); + call_requester(); + } + } + + CallOpSet meta_ops_; CallbackWithSuccessTag meta_tag_; CallOpSet - finish_buf_; + finish_ops_; CallbackWithSuccessTag finish_tag_; ServerContext* ctx_; Call call_; - RequestType* req_; + const RequestType* req_; ResponseType resp_; std::function call_requester_; + std::atomic_int callbacks_outstanding_{ + 2}; // reserve for Finish and CompletionOp + }; +}; + +template +class CallbackClientStreamingHandler : public MethodHandler { + public: + CallbackClientStreamingHandler( + std::function< + experimental::ServerReadReactor*()> + func) + : func_(std::move(func)) {} + void RunHandler(const HandlerParameter& param) final { + // Arena allocate a reader structure (that includes response) + g_core_codegen_interface->grpc_call_ref(param.call->call()); + + experimental::ServerReadReactor* reactor = + param.status.ok() + ? CatchingReactorCreator< + experimental::ServerReadReactor>( + func_) + : nullptr; + + if (reactor == nullptr) { + // if deserialization or reactor creator failed, we need to fail the call + reactor = new UnimplementedReadReactor; + } + + auto* reader = new (g_core_codegen_interface->grpc_call_arena_alloc( + param.call->call(), sizeof(ServerCallbackReaderImpl))) + ServerCallbackReaderImpl(param.server_context, param.call, + std::move(param.call_requester), reactor); + + reader->BindReactor(reactor); + reactor->OnStarted(param.server_context, reader->response()); + reader->MaybeDone(); + } + + private: + std::function*()> + func_; + + class ServerCallbackReaderImpl + : public experimental::ServerCallbackReader { + public: + void Finish(Status s) override { + finish_tag_.Set(call_.call(), [this](bool) { MaybeDone(); }, + &finish_ops_); + if (!ctx_->sent_initial_metadata_) { + finish_ops_.SendInitialMetadata(&ctx_->initial_metadata_, + ctx_->initial_metadata_flags()); + if (ctx_->compression_level_set()) { + finish_ops_.set_compression_level(ctx_->compression_level()); + } + ctx_->sent_initial_metadata_ = true; + } + // The response is dropped if the status is not OK. + if (s.ok()) { + finish_ops_.ServerSendStatus(&ctx_->trailing_metadata_, + finish_ops_.SendMessage(resp_)); + } else { + finish_ops_.ServerSendStatus(&ctx_->trailing_metadata_, s); + } + finish_ops_.set_core_cq_tag(&finish_tag_); + call_.PerformOps(&finish_ops_); + } + + void SendInitialMetadata() override { + GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_); + callbacks_outstanding_++; + meta_tag_.Set(call_.call(), + [this](bool ok) { + reactor_->OnSendInitialMetadataDone(ok); + MaybeDone(); + }, + &meta_ops_); + meta_ops_.SendInitialMetadata(&ctx_->initial_metadata_, + ctx_->initial_metadata_flags()); + if (ctx_->compression_level_set()) { + meta_ops_.set_compression_level(ctx_->compression_level()); + } + ctx_->sent_initial_metadata_ = true; + meta_ops_.set_core_cq_tag(&meta_tag_); + call_.PerformOps(&meta_ops_); + } + + void Read(RequestType* req) override { + callbacks_outstanding_++; + read_ops_.RecvMessage(req); + call_.PerformOps(&read_ops_); + } + + private: + friend class CallbackClientStreamingHandler; + + ServerCallbackReaderImpl( + ServerContext* ctx, Call* call, std::function call_requester, + experimental::ServerReadReactor* reactor) + : ctx_(ctx), + call_(*call), + call_requester_(std::move(call_requester)), + reactor_(reactor) { + ctx_->BeginCompletionOp(call, [this](bool) { MaybeDone(); }, reactor); + read_tag_.Set(call_.call(), + [this](bool ok) { + reactor_->OnReadDone(ok); + MaybeDone(); + }, + &read_ops_); + read_ops_.set_core_cq_tag(&read_tag_); + } + + ~ServerCallbackReaderImpl() {} + + ResponseType* response() { return &resp_; } + + void MaybeDone() { + if (--callbacks_outstanding_ == 0) { + reactor_->OnDone(); + grpc_call* call = call_.call(); + auto call_requester = std::move(call_requester_); + this->~ServerCallbackReaderImpl(); // explicitly call destructor + g_core_codegen_interface->grpc_call_unref(call); + call_requester(); + } + } + + CallOpSet meta_ops_; + CallbackWithSuccessTag meta_tag_; + CallOpSet + finish_ops_; + CallbackWithSuccessTag finish_tag_; + CallOpSet> read_ops_; + CallbackWithSuccessTag read_tag_; + + ServerContext* ctx_; + Call call_; + ResponseType resp_; + std::function call_requester_; + experimental::ServerReadReactor* reactor_; + std::atomic_int callbacks_outstanding_{ + 3}; // reserve for OnStarted, Finish, and CompletionOp + }; +}; + +template +class CallbackServerStreamingHandler : public MethodHandler { + public: + CallbackServerStreamingHandler( + std::function< + experimental::ServerWriteReactor*()> + func) + : func_(std::move(func)) {} + void RunHandler(const HandlerParameter& param) final { + // Arena allocate a writer structure + g_core_codegen_interface->grpc_call_ref(param.call->call()); + + experimental::ServerWriteReactor* reactor = + param.status.ok() + ? CatchingReactorCreator< + experimental::ServerWriteReactor>( + func_) + : nullptr; + + if (reactor == nullptr) { + // if deserialization or reactor creator failed, we need to fail the call + reactor = new UnimplementedWriteReactor; + } + + auto* writer = new (g_core_codegen_interface->grpc_call_arena_alloc( + param.call->call(), sizeof(ServerCallbackWriterImpl))) + ServerCallbackWriterImpl(param.server_context, param.call, + static_cast(param.request), + std::move(param.call_requester), reactor); + writer->BindReactor(reactor); + reactor->OnStarted(param.server_context, writer->request()); + writer->MaybeDone(); + } + + void* Deserialize(grpc_call* call, grpc_byte_buffer* req, + Status* status) final { + ByteBuffer buf; + buf.set_buffer(req); + auto* request = new (g_core_codegen_interface->grpc_call_arena_alloc( + call, sizeof(RequestType))) RequestType(); + *status = SerializationTraits::Deserialize(&buf, request); + buf.Release(); + if (status->ok()) { + return request; + } + request->~RequestType(); + return nullptr; + } + + private: + std::function*()> + func_; + + class ServerCallbackWriterImpl + : public experimental::ServerCallbackWriter { + public: + void Finish(Status s) override { + finish_tag_.Set(call_.call(), [this](bool) { MaybeDone(); }, + &finish_ops_); + finish_ops_.set_core_cq_tag(&finish_tag_); + + if (!ctx_->sent_initial_metadata_) { + finish_ops_.SendInitialMetadata(&ctx_->initial_metadata_, + ctx_->initial_metadata_flags()); + if (ctx_->compression_level_set()) { + finish_ops_.set_compression_level(ctx_->compression_level()); + } + ctx_->sent_initial_metadata_ = true; + } + finish_ops_.ServerSendStatus(&ctx_->trailing_metadata_, s); + call_.PerformOps(&finish_ops_); + } + + void SendInitialMetadata() override { + GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_); + callbacks_outstanding_++; + meta_tag_.Set(call_.call(), + [this](bool ok) { + reactor_->OnSendInitialMetadataDone(ok); + MaybeDone(); + }, + &meta_ops_); + meta_ops_.SendInitialMetadata(&ctx_->initial_metadata_, + ctx_->initial_metadata_flags()); + if (ctx_->compression_level_set()) { + meta_ops_.set_compression_level(ctx_->compression_level()); + } + ctx_->sent_initial_metadata_ = true; + meta_ops_.set_core_cq_tag(&meta_tag_); + call_.PerformOps(&meta_ops_); + } + + void Write(const ResponseType* resp, WriteOptions options) override { + callbacks_outstanding_++; + if (options.is_last_message()) { + options.set_buffer_hint(); + } + if (!ctx_->sent_initial_metadata_) { + write_ops_.SendInitialMetadata(&ctx_->initial_metadata_, + ctx_->initial_metadata_flags()); + if (ctx_->compression_level_set()) { + write_ops_.set_compression_level(ctx_->compression_level()); + } + ctx_->sent_initial_metadata_ = true; + } + // TODO(vjpai): don't assert + GPR_CODEGEN_ASSERT(write_ops_.SendMessage(*resp, options).ok()); + call_.PerformOps(&write_ops_); + } + + void WriteAndFinish(const ResponseType* resp, WriteOptions options, + Status s) override { + // This combines the write into the finish callback + // Don't send any message if the status is bad + if (s.ok()) { + // TODO(vjpai): don't assert + GPR_CODEGEN_ASSERT(finish_ops_.SendMessage(*resp, options).ok()); + } + Finish(std::move(s)); + } + + private: + friend class CallbackServerStreamingHandler; + + ServerCallbackWriterImpl( + ServerContext* ctx, Call* call, const RequestType* req, + std::function call_requester, + experimental::ServerWriteReactor* reactor) + : ctx_(ctx), + call_(*call), + req_(req), + call_requester_(std::move(call_requester)), + reactor_(reactor) { + ctx_->BeginCompletionOp(call, [this](bool) { MaybeDone(); }, reactor); + write_tag_.Set(call_.call(), + [this](bool ok) { + reactor_->OnWriteDone(ok); + MaybeDone(); + }, + &write_ops_); + write_ops_.set_core_cq_tag(&write_tag_); + } + ~ServerCallbackWriterImpl() { req_->~RequestType(); } + + const RequestType* request() { return req_; } + + void MaybeDone() { + if (--callbacks_outstanding_ == 0) { + reactor_->OnDone(); + grpc_call* call = call_.call(); + auto call_requester = std::move(call_requester_); + this->~ServerCallbackWriterImpl(); // explicitly call destructor + g_core_codegen_interface->grpc_call_unref(call); + call_requester(); + } + } + + CallOpSet meta_ops_; + CallbackWithSuccessTag meta_tag_; + CallOpSet + finish_ops_; + CallbackWithSuccessTag finish_tag_; + CallOpSet write_ops_; + CallbackWithSuccessTag write_tag_; + + ServerContext* ctx_; + Call call_; + const RequestType* req_; + std::function call_requester_; + experimental::ServerWriteReactor* reactor_; + std::atomic_int callbacks_outstanding_{ + 3}; // reserve for OnStarted, Finish, and CompletionOp + }; +}; + +template +class CallbackBidiHandler : public MethodHandler { + public: + CallbackBidiHandler( + std::function< + experimental::ServerBidiReactor*()> + func) + : func_(std::move(func)) {} + void RunHandler(const HandlerParameter& param) final { + g_core_codegen_interface->grpc_call_ref(param.call->call()); + + experimental::ServerBidiReactor* reactor = + param.status.ok() + ? CatchingReactorCreator< + experimental::ServerBidiReactor>( + func_) + : nullptr; + + if (reactor == nullptr) { + // if deserialization or reactor creator failed, we need to fail the call + reactor = new UnimplementedBidiReactor; + } + + auto* stream = new (g_core_codegen_interface->grpc_call_arena_alloc( + param.call->call(), sizeof(ServerCallbackReaderWriterImpl))) + ServerCallbackReaderWriterImpl(param.server_context, param.call, + std::move(param.call_requester), + reactor); + + stream->BindReactor(reactor); + reactor->OnStarted(param.server_context); + stream->MaybeDone(); + } + + private: + std::function*()> + func_; + + class ServerCallbackReaderWriterImpl + : public experimental::ServerCallbackReaderWriter { + public: + void Finish(Status s) override { + finish_tag_.Set(call_.call(), [this](bool) { MaybeDone(); }, + &finish_ops_); + finish_ops_.set_core_cq_tag(&finish_tag_); + + if (!ctx_->sent_initial_metadata_) { + finish_ops_.SendInitialMetadata(&ctx_->initial_metadata_, + ctx_->initial_metadata_flags()); + if (ctx_->compression_level_set()) { + finish_ops_.set_compression_level(ctx_->compression_level()); + } + ctx_->sent_initial_metadata_ = true; + } + finish_ops_.ServerSendStatus(&ctx_->trailing_metadata_, s); + call_.PerformOps(&finish_ops_); + } + + void SendInitialMetadata() override { + GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_); + callbacks_outstanding_++; + meta_tag_.Set(call_.call(), + [this](bool ok) { + reactor_->OnSendInitialMetadataDone(ok); + MaybeDone(); + }, + &meta_ops_); + meta_ops_.SendInitialMetadata(&ctx_->initial_metadata_, + ctx_->initial_metadata_flags()); + if (ctx_->compression_level_set()) { + meta_ops_.set_compression_level(ctx_->compression_level()); + } + ctx_->sent_initial_metadata_ = true; + meta_ops_.set_core_cq_tag(&meta_tag_); + call_.PerformOps(&meta_ops_); + } + + void Write(const ResponseType* resp, WriteOptions options) override { + callbacks_outstanding_++; + if (options.is_last_message()) { + options.set_buffer_hint(); + } + if (!ctx_->sent_initial_metadata_) { + write_ops_.SendInitialMetadata(&ctx_->initial_metadata_, + ctx_->initial_metadata_flags()); + if (ctx_->compression_level_set()) { + write_ops_.set_compression_level(ctx_->compression_level()); + } + ctx_->sent_initial_metadata_ = true; + } + // TODO(vjpai): don't assert + GPR_CODEGEN_ASSERT(write_ops_.SendMessage(*resp, options).ok()); + call_.PerformOps(&write_ops_); + } + + void WriteAndFinish(const ResponseType* resp, WriteOptions options, + Status s) override { + // Don't send any message if the status is bad + if (s.ok()) { + // TODO(vjpai): don't assert + GPR_CODEGEN_ASSERT(finish_ops_.SendMessage(*resp, options).ok()); + } + Finish(std::move(s)); + } + + void Read(RequestType* req) override { + callbacks_outstanding_++; + read_ops_.RecvMessage(req); + call_.PerformOps(&read_ops_); + } + + private: + friend class CallbackBidiHandler; + + ServerCallbackReaderWriterImpl( + ServerContext* ctx, Call* call, std::function call_requester, + experimental::ServerBidiReactor* reactor) + : ctx_(ctx), + call_(*call), + call_requester_(std::move(call_requester)), + reactor_(reactor) { + ctx_->BeginCompletionOp(call, [this](bool) { MaybeDone(); }, reactor); + write_tag_.Set(call_.call(), + [this](bool ok) { + reactor_->OnWriteDone(ok); + MaybeDone(); + }, + &write_ops_); + write_ops_.set_core_cq_tag(&write_tag_); + read_tag_.Set(call_.call(), + [this](bool ok) { + reactor_->OnReadDone(ok); + MaybeDone(); + }, + &read_ops_); + read_ops_.set_core_cq_tag(&read_tag_); + } + ~ServerCallbackReaderWriterImpl() {} + + void MaybeDone() { + if (--callbacks_outstanding_ == 0) { + reactor_->OnDone(); + grpc_call* call = call_.call(); + auto call_requester = std::move(call_requester_); + this->~ServerCallbackReaderWriterImpl(); // explicitly call destructor + g_core_codegen_interface->grpc_call_unref(call); + call_requester(); + } + } + + CallOpSet meta_ops_; + CallbackWithSuccessTag meta_tag_; + CallOpSet + finish_ops_; + CallbackWithSuccessTag finish_tag_; + CallOpSet write_ops_; + CallbackWithSuccessTag write_tag_; + CallOpSet> read_ops_; + CallbackWithSuccessTag read_tag_; + + ServerContext* ctx_; + Call call_; + std::function call_requester_; + experimental::ServerBidiReactor* reactor_; + std::atomic_int callbacks_outstanding_{ + 3}; // reserve for OnStarted, Finish, and CompletionOp }; }; diff --git a/include/grpcpp/impl/codegen/server_context.h b/include/grpcpp/impl/codegen/server_context.h index 82ee862f61..4a5f9e2dd9 100644 --- a/include/grpcpp/impl/codegen/server_context.h +++ b/include/grpcpp/impl/codegen/server_context.h @@ -66,13 +66,20 @@ template class ServerStreamingHandler; template class BidiStreamingHandler; -template +template class CallbackUnaryHandler; +template +class CallbackClientStreamingHandler; +template +class CallbackServerStreamingHandler; +template +class CallbackBidiHandler; template class TemplatedBidiStreamingHandler; template class ErrorMethodHandler; class Call; +class ServerReactor; } // namespace internal class CompletionQueue; @@ -270,8 +277,14 @@ class ServerContext { friend class ::grpc::internal::ServerStreamingHandler; template friend class ::grpc::internal::TemplatedBidiStreamingHandler; - template + template friend class ::grpc::internal::CallbackUnaryHandler; + template + friend class ::grpc::internal::CallbackClientStreamingHandler; + template + friend class ::grpc::internal::CallbackServerStreamingHandler; + template + friend class ::grpc::internal::CallbackBidiHandler; template friend class internal::ErrorMethodHandler; friend class ::grpc::ClientContext; @@ -282,7 +295,9 @@ class ServerContext { class CompletionOp; - void BeginCompletionOp(internal::Call* call, bool callback); + void BeginCompletionOp(internal::Call* call, + std::function callback, + internal::ServerReactor* reactor); /// Return the tag queued by BeginCompletionOp() internal::CompletionQueueTag* GetCompletionOpTag(); diff --git a/src/compiler/cpp_generator.cc b/src/compiler/cpp_generator.cc index a368b47f01..b004687250 100644 --- a/src/compiler/cpp_generator.cc +++ b/src/compiler/cpp_generator.cc @@ -889,6 +889,11 @@ void PrintHeaderServerCallbackMethodsHelper( " abort();\n" " return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, \"\");\n" "}\n"); + printer->Print(*vars, + "virtual ::grpc::experimental::ServerReadReactor< " + "$RealRequest$, $RealResponse$>* $Method$() {\n" + " return new ::grpc::internal::UnimplementedReadReactor<\n" + " $RealRequest$, $RealResponse$>;}\n"); } else if (ServerOnlyStreaming(method)) { printer->Print( *vars, @@ -900,6 +905,11 @@ void PrintHeaderServerCallbackMethodsHelper( " abort();\n" " return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, \"\");\n" "}\n"); + printer->Print(*vars, + "virtual ::grpc::experimental::ServerWriteReactor< " + "$RealRequest$, $RealResponse$>* $Method$() {\n" + " return new ::grpc::internal::UnimplementedWriteReactor<\n" + " $RealRequest$, $RealResponse$>;}\n"); } else if (method->BidiStreaming()) { printer->Print( *vars, @@ -911,6 +921,11 @@ void PrintHeaderServerCallbackMethodsHelper( " abort();\n" " return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, \"\");\n" "}\n"); + printer->Print(*vars, + "virtual ::grpc::experimental::ServerBidiReactor< " + "$RealRequest$, $RealResponse$>* $Method$() {\n" + " return new ::grpc::internal::UnimplementedBidiReactor<\n" + " $RealRequest$, $RealResponse$>;}\n"); } } @@ -939,22 +954,36 @@ void PrintHeaderServerMethodCallback( *vars, " ::grpc::Service::experimental().MarkMethodCallback($Idx$,\n" " new ::grpc::internal::CallbackUnaryHandler< " - "ExperimentalWithCallbackMethod_$Method$, $RealRequest$, " - "$RealResponse$>(\n" + "$RealRequest$, $RealResponse$>(\n" " [this](::grpc::ServerContext* context,\n" " const $RealRequest$* request,\n" " $RealResponse$* response,\n" " ::grpc::experimental::ServerCallbackRpcController* " "controller) {\n" - " this->$" + " return this->$" "Method$(context, request, response, controller);\n" - " }, this));\n"); + " }));\n"); } else if (ClientOnlyStreaming(method)) { - // TODO(vjpai): Add in code generation for all streaming methods + printer->Print( + *vars, + " ::grpc::Service::experimental().MarkMethodCallback($Idx$,\n" + " new ::grpc::internal::CallbackClientStreamingHandler< " + "$RealRequest$, $RealResponse$>(\n" + " [this] { return this->$Method$(); }));\n"); } else if (ServerOnlyStreaming(method)) { - // TODO(vjpai): Add in code generation for all streaming methods + printer->Print( + *vars, + " ::grpc::Service::experimental().MarkMethodCallback($Idx$,\n" + " new ::grpc::internal::CallbackServerStreamingHandler< " + "$RealRequest$, $RealResponse$>(\n" + " [this] { return this->$Method$(); }));\n"); } else if (method->BidiStreaming()) { - // TODO(vjpai): Add in code generation for all streaming methods + printer->Print( + *vars, + " ::grpc::Service::experimental().MarkMethodCallback($Idx$,\n" + " new ::grpc::internal::CallbackBidiHandler< " + "$RealRequest$, $RealResponse$>(\n" + " [this] { return this->$Method$(); }));\n"); } printer->Print(*vars, "}\n"); printer->Print(*vars, @@ -991,8 +1020,7 @@ void PrintHeaderServerMethodRawCallback( *vars, " ::grpc::Service::experimental().MarkMethodRawCallback($Idx$,\n" " new ::grpc::internal::CallbackUnaryHandler< " - "ExperimentalWithRawCallbackMethod_$Method$, $RealRequest$, " - "$RealResponse$>(\n" + "$RealRequest$, $RealResponse$>(\n" " [this](::grpc::ServerContext* context,\n" " const $RealRequest$* request,\n" " $RealResponse$* response,\n" @@ -1000,13 +1028,28 @@ void PrintHeaderServerMethodRawCallback( "controller) {\n" " this->$" "Method$(context, request, response, controller);\n" - " }, this));\n"); + " }));\n"); } else if (ClientOnlyStreaming(method)) { - // TODO(vjpai): Add in code generation for all streaming methods + printer->Print( + *vars, + " ::grpc::Service::experimental().MarkMethodRawCallback($Idx$,\n" + " new ::grpc::internal::CallbackClientStreamingHandler< " + "$RealRequest$, $RealResponse$>(\n" + " [this] { return this->$Method$(); }));\n"); } else if (ServerOnlyStreaming(method)) { - // TODO(vjpai): Add in code generation for all streaming methods + printer->Print( + *vars, + " ::grpc::Service::experimental().MarkMethodRawCallback($Idx$,\n" + " new ::grpc::internal::CallbackServerStreamingHandler< " + "$RealRequest$, $RealResponse$>(\n" + " [this] { return this->$Method$(); }));\n"); } else if (method->BidiStreaming()) { - // TODO(vjpai): Add in code generation for all streaming methods + printer->Print( + *vars, + " ::grpc::Service::experimental().MarkMethodRawCallback($Idx$,\n" + " new ::grpc::internal::CallbackBidiHandler< " + "$RealRequest$, $RealResponse$>(\n" + " [this] { return this->$Method$(); }));\n"); } printer->Print(*vars, "}\n"); printer->Print(*vars, diff --git a/src/cpp/server/server_cc.cc b/src/cpp/server/server_cc.cc index 0a51cf5626..69af43a656 100644 --- a/src/cpp/server/server_cc.cc +++ b/src/cpp/server/server_cc.cc @@ -291,7 +291,7 @@ class Server::SyncRequest final : public internal::CompletionQueueTag { void ContinueRunAfterInterception() { { - ctx_.BeginCompletionOp(&call_, false); + ctx_.BeginCompletionOp(&call_, nullptr, nullptr); global_callbacks_->PreSynchronousRequest(&ctx_); auto* handler = resources_ ? method_->handler() : server_->resource_exhausted_handler_.get(); @@ -456,7 +456,6 @@ class Server::CallbackRequest final : public internal::CompletionQueueTag { } } void ContinueRunAfterInterception() { - req_->ctx_.BeginCompletionOp(call_, true); req_->method_->handler()->RunHandler( internal::MethodHandler::HandlerParameter( call_, &req_->ctx_, req_->request_, req_->request_status_, @@ -1018,7 +1017,7 @@ bool ServerInterface::BaseAsyncRequest::FinalizeResult(void** tag, } } if (*status && call_) { - context_->BeginCompletionOp(&call_wrapper_, false); + context_->BeginCompletionOp(&call_wrapper_, nullptr, nullptr); } *tag = tag_; if (delete_on_finalize_) { @@ -1029,7 +1028,7 @@ bool ServerInterface::BaseAsyncRequest::FinalizeResult(void** tag, void ServerInterface::BaseAsyncRequest:: ContinueFinalizeResultAfterInterception() { - context_->BeginCompletionOp(&call_wrapper_, false); + context_->BeginCompletionOp(&call_wrapper_, nullptr, nullptr); // Queue a tag which will be returned immediately grpc_core::ExecCtx exec_ctx; grpc_cq_begin_op(notification_cq_->cq(), this); diff --git a/src/cpp/server/server_context.cc b/src/cpp/server/server_context.cc index 9c01f896e6..1b524bc3e8 100644 --- a/src/cpp/server/server_context.cc +++ b/src/cpp/server/server_context.cc @@ -17,6 +17,7 @@ */ #include +#include #include #include @@ -41,8 +42,9 @@ class ServerContext::CompletionOp final : public internal::CallOpSetInterface { public: // initial refs: one in the server context, one in the cq // must ref the call before calling constructor and after deleting this - CompletionOp(internal::Call* call) + CompletionOp(internal::Call* call, internal::ServerReactor* reactor) : call_(*call), + reactor_(reactor), has_tag_(false), tag_(nullptr), core_cq_tag_(this), @@ -124,9 +126,9 @@ class ServerContext::CompletionOp final : public internal::CallOpSetInterface { return; } /* Start a dummy op so that we can return the tag */ - GPR_CODEGEN_ASSERT(GRPC_CALL_OK == - g_core_codegen_interface->grpc_call_start_batch( - call_.call(), nullptr, 0, this, nullptr)); + GPR_CODEGEN_ASSERT( + GRPC_CALL_OK == + grpc_call_start_batch(call_.call(), nullptr, 0, core_cq_tag_, nullptr)); } private: @@ -136,13 +138,14 @@ class ServerContext::CompletionOp final : public internal::CallOpSetInterface { } internal::Call call_; + internal::ServerReactor* reactor_; bool has_tag_; void* tag_; void* core_cq_tag_; std::mutex mu_; int refs_; bool finalized_; - int cancelled_; + int cancelled_; // This is an int (not bool) because it is passed to core bool done_intercepting_; internal::InterceptorBatchMethodsImpl interceptor_methods_; }; @@ -190,7 +193,16 @@ bool ServerContext::CompletionOp::FinalizeResult(void** tag, bool* status) { } finalized_ = true; - if (!*status) cancelled_ = 1; + // If for some reason the incoming status is false, mark that as a + // cancellation. + // TODO(vjpai): does this ever happen? + if (!*status) { + cancelled_ = 1; + } + + if (cancelled_ && (reactor_ != nullptr)) { + reactor_->OnCancel(); + } /* Release the lock since we are going to be running through interceptors now */ lock.unlock(); @@ -251,21 +263,25 @@ void ServerContext::Clear() { initial_metadata_.clear(); trailing_metadata_.clear(); client_metadata_.Reset(); - if (call_) { - grpc_call_unref(call_); - } if (completion_op_) { completion_op_->Unref(); + completion_op_ = nullptr; completion_tag_.Clear(); } if (rpc_info_) { rpc_info_->Unref(); + rpc_info_ = nullptr; + } + if (call_) { + auto* call = call_; + call_ = nullptr; + grpc_call_unref(call); } - // Don't need to clear out call_, completion_op_, or rpc_info_ because this is - // either called from destructor or just before Setup } -void ServerContext::BeginCompletionOp(internal::Call* call, bool callback) { +void ServerContext::BeginCompletionOp(internal::Call* call, + std::function callback, + internal::ServerReactor* reactor) { GPR_ASSERT(!completion_op_); if (rpc_info_) { rpc_info_->Ref(); @@ -273,10 +289,11 @@ void ServerContext::BeginCompletionOp(internal::Call* call, bool callback) { grpc_call_ref(call->call()); completion_op_ = new (grpc_call_arena_alloc(call->call(), sizeof(CompletionOp))) - CompletionOp(call); - if (callback) { - completion_tag_.Set(call->call(), nullptr, completion_op_); + CompletionOp(call, reactor); + if (callback != nullptr) { + completion_tag_.Set(call->call(), std::move(callback), completion_op_); completion_op_->set_core_cq_tag(&completion_tag_); + completion_op_->set_tag(completion_op_); } else if (has_notify_when_done_tag_) { completion_op_->set_tag(async_notify_when_done_tag_); } diff --git a/test/cpp/codegen/compiler_test_golden b/test/cpp/codegen/compiler_test_golden index 5f0eb6c35c..1871e1375e 100644 --- a/test/cpp/codegen/compiler_test_golden +++ b/test/cpp/codegen/compiler_test_golden @@ -322,13 +322,13 @@ class ServiceA final { public: ExperimentalWithCallbackMethod_MethodA1() { ::grpc::Service::experimental().MarkMethodCallback(0, - new ::grpc::internal::CallbackUnaryHandler< ExperimentalWithCallbackMethod_MethodA1, ::grpc::testing::Request, ::grpc::testing::Response>( + new ::grpc::internal::CallbackUnaryHandler< ::grpc::testing::Request, ::grpc::testing::Response>( [this](::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response, ::grpc::experimental::ServerCallbackRpcController* controller) { - this->MethodA1(context, request, response, controller); - }, this)); + return this->MethodA1(context, request, response, controller); + })); } ~ExperimentalWithCallbackMethod_MethodA1() override { BaseClassMustBeDerivedFromService(this); @@ -346,6 +346,9 @@ class ServiceA final { void BaseClassMustBeDerivedFromService(const Service *service) {} public: ExperimentalWithCallbackMethod_MethodA2() { + ::grpc::Service::experimental().MarkMethodCallback(1, + new ::grpc::internal::CallbackClientStreamingHandler< ::grpc::testing::Request, ::grpc::testing::Response>( + [this] { return this->MethodA2(); })); } ~ExperimentalWithCallbackMethod_MethodA2() override { BaseClassMustBeDerivedFromService(this); @@ -355,6 +358,9 @@ class ServiceA final { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } + virtual ::grpc::experimental::ServerReadReactor< ::grpc::testing::Request, ::grpc::testing::Response>* MethodA2() { + return new ::grpc::internal::UnimplementedReadReactor< + ::grpc::testing::Request, ::grpc::testing::Response>;} }; template class ExperimentalWithCallbackMethod_MethodA3 : public BaseClass { @@ -362,6 +368,9 @@ class ServiceA final { void BaseClassMustBeDerivedFromService(const Service *service) {} public: ExperimentalWithCallbackMethod_MethodA3() { + ::grpc::Service::experimental().MarkMethodCallback(2, + new ::grpc::internal::CallbackServerStreamingHandler< ::grpc::testing::Request, ::grpc::testing::Response>( + [this] { return this->MethodA3(); })); } ~ExperimentalWithCallbackMethod_MethodA3() override { BaseClassMustBeDerivedFromService(this); @@ -371,6 +380,9 @@ class ServiceA final { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } + virtual ::grpc::experimental::ServerWriteReactor< ::grpc::testing::Request, ::grpc::testing::Response>* MethodA3() { + return new ::grpc::internal::UnimplementedWriteReactor< + ::grpc::testing::Request, ::grpc::testing::Response>;} }; template class ExperimentalWithCallbackMethod_MethodA4 : public BaseClass { @@ -378,6 +390,9 @@ class ServiceA final { void BaseClassMustBeDerivedFromService(const Service *service) {} public: ExperimentalWithCallbackMethod_MethodA4() { + ::grpc::Service::experimental().MarkMethodCallback(3, + new ::grpc::internal::CallbackBidiHandler< ::grpc::testing::Request, ::grpc::testing::Response>( + [this] { return this->MethodA4(); })); } ~ExperimentalWithCallbackMethod_MethodA4() override { BaseClassMustBeDerivedFromService(this); @@ -387,6 +402,9 @@ class ServiceA final { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } + virtual ::grpc::experimental::ServerBidiReactor< ::grpc::testing::Request, ::grpc::testing::Response>* MethodA4() { + return new ::grpc::internal::UnimplementedBidiReactor< + ::grpc::testing::Request, ::grpc::testing::Response>;} }; typedef ExperimentalWithCallbackMethod_MethodA1 > > > ExperimentalCallbackService; template @@ -544,13 +562,13 @@ class ServiceA final { public: ExperimentalWithRawCallbackMethod_MethodA1() { ::grpc::Service::experimental().MarkMethodRawCallback(0, - new ::grpc::internal::CallbackUnaryHandler< ExperimentalWithRawCallbackMethod_MethodA1, ::grpc::ByteBuffer, ::grpc::ByteBuffer>( + new ::grpc::internal::CallbackUnaryHandler< ::grpc::ByteBuffer, ::grpc::ByteBuffer>( [this](::grpc::ServerContext* context, const ::grpc::ByteBuffer* request, ::grpc::ByteBuffer* response, ::grpc::experimental::ServerCallbackRpcController* controller) { this->MethodA1(context, request, response, controller); - }, this)); + })); } ~ExperimentalWithRawCallbackMethod_MethodA1() override { BaseClassMustBeDerivedFromService(this); @@ -568,6 +586,9 @@ class ServiceA final { void BaseClassMustBeDerivedFromService(const Service *service) {} public: ExperimentalWithRawCallbackMethod_MethodA2() { + ::grpc::Service::experimental().MarkMethodRawCallback(1, + new ::grpc::internal::CallbackClientStreamingHandler< ::grpc::ByteBuffer, ::grpc::ByteBuffer>( + [this] { return this->MethodA2(); })); } ~ExperimentalWithRawCallbackMethod_MethodA2() override { BaseClassMustBeDerivedFromService(this); @@ -577,6 +598,9 @@ class ServiceA final { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } + virtual ::grpc::experimental::ServerReadReactor< ::grpc::ByteBuffer, ::grpc::ByteBuffer>* MethodA2() { + return new ::grpc::internal::UnimplementedReadReactor< + ::grpc::ByteBuffer, ::grpc::ByteBuffer>;} }; template class ExperimentalWithRawCallbackMethod_MethodA3 : public BaseClass { @@ -584,6 +608,9 @@ class ServiceA final { void BaseClassMustBeDerivedFromService(const Service *service) {} public: ExperimentalWithRawCallbackMethod_MethodA3() { + ::grpc::Service::experimental().MarkMethodRawCallback(2, + new ::grpc::internal::CallbackServerStreamingHandler< ::grpc::ByteBuffer, ::grpc::ByteBuffer>( + [this] { return this->MethodA3(); })); } ~ExperimentalWithRawCallbackMethod_MethodA3() override { BaseClassMustBeDerivedFromService(this); @@ -593,6 +620,9 @@ class ServiceA final { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } + virtual ::grpc::experimental::ServerWriteReactor< ::grpc::ByteBuffer, ::grpc::ByteBuffer>* MethodA3() { + return new ::grpc::internal::UnimplementedWriteReactor< + ::grpc::ByteBuffer, ::grpc::ByteBuffer>;} }; template class ExperimentalWithRawCallbackMethod_MethodA4 : public BaseClass { @@ -600,6 +630,9 @@ class ServiceA final { void BaseClassMustBeDerivedFromService(const Service *service) {} public: ExperimentalWithRawCallbackMethod_MethodA4() { + ::grpc::Service::experimental().MarkMethodRawCallback(3, + new ::grpc::internal::CallbackBidiHandler< ::grpc::ByteBuffer, ::grpc::ByteBuffer>( + [this] { return this->MethodA4(); })); } ~ExperimentalWithRawCallbackMethod_MethodA4() override { BaseClassMustBeDerivedFromService(this); @@ -609,6 +642,9 @@ class ServiceA final { abort(); return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } + virtual ::grpc::experimental::ServerBidiReactor< ::grpc::ByteBuffer, ::grpc::ByteBuffer>* MethodA4() { + return new ::grpc::internal::UnimplementedBidiReactor< + ::grpc::ByteBuffer, ::grpc::ByteBuffer>;} }; template class WithStreamedUnaryMethod_MethodA1 : public BaseClass { @@ -752,13 +788,13 @@ class ServiceB final { public: ExperimentalWithCallbackMethod_MethodB1() { ::grpc::Service::experimental().MarkMethodCallback(0, - new ::grpc::internal::CallbackUnaryHandler< ExperimentalWithCallbackMethod_MethodB1, ::grpc::testing::Request, ::grpc::testing::Response>( + new ::grpc::internal::CallbackUnaryHandler< ::grpc::testing::Request, ::grpc::testing::Response>( [this](::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response, ::grpc::experimental::ServerCallbackRpcController* controller) { - this->MethodB1(context, request, response, controller); - }, this)); + return this->MethodB1(context, request, response, controller); + })); } ~ExperimentalWithCallbackMethod_MethodB1() override { BaseClassMustBeDerivedFromService(this); @@ -815,13 +851,13 @@ class ServiceB final { public: ExperimentalWithRawCallbackMethod_MethodB1() { ::grpc::Service::experimental().MarkMethodRawCallback(0, - new ::grpc::internal::CallbackUnaryHandler< ExperimentalWithRawCallbackMethod_MethodB1, ::grpc::ByteBuffer, ::grpc::ByteBuffer>( + new ::grpc::internal::CallbackUnaryHandler< ::grpc::ByteBuffer, ::grpc::ByteBuffer>( [this](::grpc::ServerContext* context, const ::grpc::ByteBuffer* request, ::grpc::ByteBuffer* response, ::grpc::experimental::ServerCallbackRpcController* controller) { this->MethodB1(context, request, response, controller); - }, this)); + })); } ~ExperimentalWithRawCallbackMethod_MethodB1() override { BaseClassMustBeDerivedFromService(this); diff --git a/test/cpp/end2end/end2end_test.cc b/test/cpp/end2end/end2end_test.cc index 03291e1785..4d37bae217 100644 --- a/test/cpp/end2end/end2end_test.cc +++ b/test/cpp/end2end/end2end_test.cc @@ -196,16 +196,18 @@ class TestServiceImplDupPkg class TestScenario { public: TestScenario(bool interceptors, bool proxy, bool inproc_stub, - const grpc::string& creds_type) + const grpc::string& creds_type, bool use_callback_server) : use_interceptors(interceptors), use_proxy(proxy), inproc(inproc_stub), - credentials_type(creds_type) {} + credentials_type(creds_type), + callback_server(use_callback_server) {} void Log() const; bool use_interceptors; bool use_proxy; bool inproc; const grpc::string credentials_type; + bool callback_server; }; static std::ostream& operator<<(std::ostream& out, @@ -214,6 +216,8 @@ static std::ostream& operator<<(std::ostream& out, << (scenario.use_interceptors ? "true" : "false") << ", use_proxy=" << (scenario.use_proxy ? "true" : "false") << ", inproc=" << (scenario.inproc ? "true" : "false") + << ", server_type=" + << (scenario.callback_server ? "callback" : "sync") << ", credentials='" << scenario.credentials_type << "'}"; } @@ -280,7 +284,11 @@ class End2endTest : public ::testing::TestWithParam { builder.experimental().SetInterceptorCreators(std::move(creators)); } builder.AddListeningPort(server_address_.str(), server_creds); - builder.RegisterService(&service_); + if (!GetParam().callback_server) { + builder.RegisterService(&service_); + } else { + builder.RegisterService(&callback_service_); + } builder.RegisterService("foo.test.youtube.com", &special_service_); builder.RegisterService(&dup_pkg_service_); @@ -362,6 +370,7 @@ class End2endTest : public ::testing::TestWithParam { std::ostringstream server_address_; const int kMaxMessageSize_; TestServiceImpl service_; + CallbackTestServiceImpl callback_service_; TestServiceImpl special_service_; TestServiceImplDupPkg dup_pkg_service_; grpc::string user_agent_prefix_; @@ -1016,7 +1025,8 @@ TEST_P(End2endTest, DiffPackageServices) { EXPECT_TRUE(s.ok()); } -void CancelRpc(ClientContext* context, int delay_us, TestServiceImpl* service) { +template +void CancelRpc(ClientContext* context, int delay_us, ServiceType* service) { gpr_sleep_until(gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), gpr_time_from_micros(delay_us, GPR_TIMESPAN))); while (!service->signal_client()) { @@ -1446,7 +1456,24 @@ TEST_P(ProxyEnd2endTest, ClientCancelsRpc) { request.mutable_param()->set_client_cancel_after_us(kCancelDelayUs); ClientContext context; - std::thread cancel_thread(CancelRpc, &context, kCancelDelayUs, &service_); + std::thread cancel_thread; + if (!GetParam().callback_server) { + cancel_thread = std::thread( + [&context, this](int delay) { CancelRpc(&context, delay, &service_); }, + kCancelDelayUs); + // Note: the unusual pattern above (and below) is caused by a conflict + // between two sets of compiler expectations. clang allows const to be + // captured without mention, so there is no need to capture kCancelDelayUs + // (and indeed clang-tidy complains if you do so). OTOH, a Windows compiler + // in our tests requires an explicit capture even for const. We square this + // circle by passing the const value in as an argument to the lambda. + } else { + cancel_thread = std::thread( + [&context, this](int delay) { + CancelRpc(&context, delay, &callback_service_); + }, + kCancelDelayUs); + } Status s = stub_->Echo(&context, request, &response); cancel_thread.join(); EXPECT_EQ(StatusCode::CANCELLED, s.error_code()); @@ -1838,10 +1865,12 @@ TEST_P(ResourceQuotaEnd2endTest, SimpleRequest) { EXPECT_TRUE(s.ok()); } +// TODO(vjpai): refactor arguments into a struct if it makes sense std::vector CreateTestScenarios(bool use_proxy, bool test_insecure, bool test_secure, - bool test_inproc) { + bool test_inproc, + bool test_callback_server) { std::vector scenarios; std::vector credentials_types; if (test_secure) { @@ -1857,41 +1886,48 @@ std::vector CreateTestScenarios(bool use_proxy, if (test_insecure && insec_ok()) { credentials_types.push_back(kInsecureCredentialsType); } + + // For now test callback server only with inproc GPR_ASSERT(!credentials_types.empty()); for (const auto& cred : credentials_types) { - scenarios.emplace_back(false, false, false, cred); - scenarios.emplace_back(true, false, false, cred); + scenarios.emplace_back(false, false, false, cred, false); + scenarios.emplace_back(true, false, false, cred, false); if (use_proxy) { - scenarios.emplace_back(false, true, false, cred); - scenarios.emplace_back(true, true, false, cred); + scenarios.emplace_back(false, true, false, cred, false); + scenarios.emplace_back(true, true, false, cred, false); } } if (test_inproc && insec_ok()) { - scenarios.emplace_back(false, false, true, kInsecureCredentialsType); - scenarios.emplace_back(true, false, true, kInsecureCredentialsType); + scenarios.emplace_back(false, false, true, kInsecureCredentialsType, false); + scenarios.emplace_back(true, false, true, kInsecureCredentialsType, false); + if (test_callback_server) { + scenarios.emplace_back(false, false, true, kInsecureCredentialsType, + true); + scenarios.emplace_back(true, false, true, kInsecureCredentialsType, true); + } } return scenarios; } -INSTANTIATE_TEST_CASE_P(End2end, End2endTest, - ::testing::ValuesIn(CreateTestScenarios(false, true, - true, true))); +INSTANTIATE_TEST_CASE_P( + End2end, End2endTest, + ::testing::ValuesIn(CreateTestScenarios(false, true, true, true, true))); -INSTANTIATE_TEST_CASE_P(End2endServerTryCancel, End2endServerTryCancelTest, - ::testing::ValuesIn(CreateTestScenarios(false, true, - true, true))); +INSTANTIATE_TEST_CASE_P( + End2endServerTryCancel, End2endServerTryCancelTest, + ::testing::ValuesIn(CreateTestScenarios(false, true, true, true, true))); -INSTANTIATE_TEST_CASE_P(ProxyEnd2end, ProxyEnd2endTest, - ::testing::ValuesIn(CreateTestScenarios(true, true, - true, true))); +INSTANTIATE_TEST_CASE_P( + ProxyEnd2end, ProxyEnd2endTest, + ::testing::ValuesIn(CreateTestScenarios(true, true, true, true, false))); -INSTANTIATE_TEST_CASE_P(SecureEnd2end, SecureEnd2endTest, - ::testing::ValuesIn(CreateTestScenarios(false, false, - true, false))); +INSTANTIATE_TEST_CASE_P( + SecureEnd2end, SecureEnd2endTest, + ::testing::ValuesIn(CreateTestScenarios(false, false, true, false, true))); -INSTANTIATE_TEST_CASE_P(ResourceQuotaEnd2end, ResourceQuotaEnd2endTest, - ::testing::ValuesIn(CreateTestScenarios(false, true, - true, true))); +INSTANTIATE_TEST_CASE_P( + ResourceQuotaEnd2end, ResourceQuotaEnd2endTest, + ::testing::ValuesIn(CreateTestScenarios(false, true, true, true, false))); } // namespace } // namespace testing diff --git a/test/cpp/end2end/test_service_impl.cc b/test/cpp/end2end/test_service_impl.cc index 1726e87ea6..9d9c01cade 100644 --- a/test/cpp/end2end/test_service_impl.cc +++ b/test/cpp/end2end/test_service_impl.cc @@ -71,6 +71,46 @@ void CheckServerAuthContext( } } // namespace +namespace { +int GetIntValueFromMetadataHelper( + const char* key, + const std::multimap& metadata, + int default_value) { + if (metadata.find(key) != metadata.end()) { + std::istringstream iss(ToString(metadata.find(key)->second)); + iss >> default_value; + gpr_log(GPR_INFO, "%s : %d", key, default_value); + } + + return default_value; +} + +int GetIntValueFromMetadata( + const char* key, + const std::multimap& metadata, + int default_value) { + return GetIntValueFromMetadataHelper(key, metadata, default_value); +} + +void ServerTryCancel(ServerContext* context) { + EXPECT_FALSE(context->IsCancelled()); + context->TryCancel(); + gpr_log(GPR_INFO, "Server called TryCancel() to cancel the request"); + // Now wait until it's really canceled + while (!context->IsCancelled()) { + gpr_sleep_until(gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), + gpr_time_from_micros(1000, GPR_TIMESPAN))); + } +} + +void ServerTryCancelNonblocking(ServerContext* context) { + EXPECT_FALSE(context->IsCancelled()); + context->TryCancel(); + gpr_log(GPR_INFO, "Server called TryCancel() to cancel the request"); +} + +} // namespace + Status TestServiceImpl::Echo(ServerContext* context, const EchoRequest* request, EchoResponse* response) { // A bit of sleep to make sure that short deadline tests fail @@ -195,6 +235,7 @@ void CallbackTestServiceImpl::EchoNonDelayed( controller->Finish(Status(static_cast(error.code()), error.error_message(), error.binary_error_details())); + return; } int server_try_cancel = GetIntValueFromMetadata( kServerTryCancelRequest, context->client_metadata(), DO_NOT_CANCEL); @@ -254,7 +295,7 @@ void CallbackTestServiceImpl::EchoNonDelayed( alarm_.experimental().Set( gpr_time_add( gpr_now(GPR_CLOCK_REALTIME), - gpr_time_from_micros(request->param().client_cancel_after_us(), + gpr_time_from_micros(request->param().server_cancel_after_us(), GPR_TIMESPAN)), [controller](bool) { controller->Finish(Status::CANCELLED); }); return; @@ -279,6 +320,7 @@ void CallbackTestServiceImpl::EchoNonDelayed( request->param().debug_info().SerializeAsString(); context->AddTrailingMetadata(kDebugInfoTrailerKey, serialized_debug_info); controller->Finish(Status::CANCELLED); + return; } } if (request->has_param() && @@ -325,7 +367,7 @@ Status TestServiceImpl::RequestStream(ServerContext* context, std::thread* server_try_cancel_thd = nullptr; if (server_try_cancel == CANCEL_DURING_PROCESSING) { server_try_cancel_thd = - new std::thread(&TestServiceImpl::ServerTryCancel, this, context); + new std::thread([context] { ServerTryCancel(context); }); } int num_msgs_read = 0; @@ -380,7 +422,7 @@ Status TestServiceImpl::ResponseStream(ServerContext* context, std::thread* server_try_cancel_thd = nullptr; if (server_try_cancel == CANCEL_DURING_PROCESSING) { server_try_cancel_thd = - new std::thread(&TestServiceImpl::ServerTryCancel, this, context); + new std::thread([context] { ServerTryCancel(context); }); } for (int i = 0; i < server_responses_to_send; i++) { @@ -431,7 +473,7 @@ Status TestServiceImpl::BidiStream( std::thread* server_try_cancel_thd = nullptr; if (server_try_cancel == CANCEL_DURING_PROCESSING) { server_try_cancel_thd = - new std::thread(&TestServiceImpl::ServerTryCancel, this, context); + new std::thread([context] { ServerTryCancel(context); }); } // kServerFinishAfterNReads suggests after how many reads, the server should @@ -465,44 +507,244 @@ Status TestServiceImpl::BidiStream( return Status::OK; } -namespace { -int GetIntValueFromMetadataHelper( - const char* key, - const std::multimap& metadata, - int default_value) { - if (metadata.find(key) != metadata.end()) { - std::istringstream iss(ToString(metadata.find(key)->second)); - iss >> default_value; - gpr_log(GPR_INFO, "%s : %d", key, default_value); - } +experimental::ServerReadReactor* +CallbackTestServiceImpl::RequestStream() { + class Reactor : public ::grpc::experimental::ServerReadReactor { + public: + Reactor() {} + void OnStarted(ServerContext* context, EchoResponse* response) override { + ctx_ = context; + response_ = response; + // If 'server_try_cancel' is set in the metadata, the RPC is cancelled by + // the server by calling ServerContext::TryCancel() depending on the + // value: + // CANCEL_BEFORE_PROCESSING: The RPC is cancelled before the server + // reads any message from the client CANCEL_DURING_PROCESSING: The RPC + // is cancelled while the server is reading messages from the client + // CANCEL_AFTER_PROCESSING: The RPC is cancelled after the server reads + // all the messages from the client + server_try_cancel_ = GetIntValueFromMetadata( + kServerTryCancelRequest, context->client_metadata(), DO_NOT_CANCEL); + + response_->set_message(""); + + if (server_try_cancel_ == CANCEL_BEFORE_PROCESSING) { + ServerTryCancelNonblocking(ctx_); + return; + } - return default_value; -} -}; // namespace + if (server_try_cancel_ == CANCEL_DURING_PROCESSING) { + ctx_->TryCancel(); + // Don't wait for it here + } -int TestServiceImpl::GetIntValueFromMetadata( - const char* key, - const std::multimap& metadata, - int default_value) { - return GetIntValueFromMetadataHelper(key, metadata, default_value); + StartRead(&request_); + } + void OnDone() override { delete this; } + void OnCancel() override { FinishOnce(Status::CANCELLED); } + void OnReadDone(bool ok) override { + if (ok) { + response_->mutable_message()->append(request_.message()); + num_msgs_read_++; + StartRead(&request_); + } else { + gpr_log(GPR_INFO, "Read: %d messages", num_msgs_read_); + + if (server_try_cancel_ == CANCEL_DURING_PROCESSING) { + // Let OnCancel recover this + return; + } + if (server_try_cancel_ == CANCEL_AFTER_PROCESSING) { + ServerTryCancelNonblocking(ctx_); + return; + } + FinishOnce(Status::OK); + } + } + + private: + void FinishOnce(const Status& s) { + std::lock_guard l(finish_mu_); + if (!finished_) { + Finish(s); + finished_ = true; + } + } + + ServerContext* ctx_; + EchoResponse* response_; + EchoRequest request_; + int num_msgs_read_{0}; + int server_try_cancel_; + std::mutex finish_mu_; + bool finished_{false}; + }; + + return new Reactor; } -int CallbackTestServiceImpl::GetIntValueFromMetadata( - const char* key, - const std::multimap& metadata, - int default_value) { - return GetIntValueFromMetadataHelper(key, metadata, default_value); +// Return 'kNumResponseStreamMsgs' messages. +// TODO(yangg) make it generic by adding a parameter into EchoRequest +experimental::ServerWriteReactor* +CallbackTestServiceImpl::ResponseStream() { + class Reactor + : public ::grpc::experimental::ServerWriteReactor { + public: + Reactor() {} + void OnStarted(ServerContext* context, + const EchoRequest* request) override { + ctx_ = context; + request_ = request; + // If 'server_try_cancel' is set in the metadata, the RPC is cancelled by + // the server by calling ServerContext::TryCancel() depending on the + // value: + // CANCEL_BEFORE_PROCESSING: The RPC is cancelled before the server + // reads any message from the client CANCEL_DURING_PROCESSING: The RPC + // is cancelled while the server is reading messages from the client + // CANCEL_AFTER_PROCESSING: The RPC is cancelled after the server reads + // all the messages from the client + server_try_cancel_ = GetIntValueFromMetadata( + kServerTryCancelRequest, context->client_metadata(), DO_NOT_CANCEL); + server_coalescing_api_ = GetIntValueFromMetadata( + kServerUseCoalescingApi, context->client_metadata(), 0); + server_responses_to_send_ = GetIntValueFromMetadata( + kServerResponseStreamsToSend, context->client_metadata(), + kServerDefaultResponseStreamsToSend); + if (server_try_cancel_ == CANCEL_BEFORE_PROCESSING) { + ServerTryCancelNonblocking(ctx_); + return; + } + + if (server_try_cancel_ == CANCEL_DURING_PROCESSING) { + ctx_->TryCancel(); + } + if (num_msgs_sent_ < server_responses_to_send_) { + NextWrite(); + } + } + void OnDone() override { delete this; } + void OnCancel() override { FinishOnce(Status::CANCELLED); } + void OnWriteDone(bool ok) override { + if (num_msgs_sent_ < server_responses_to_send_) { + NextWrite(); + } else if (server_try_cancel_ == CANCEL_DURING_PROCESSING) { + // Let OnCancel recover this + } else if (server_try_cancel_ == CANCEL_AFTER_PROCESSING) { + ServerTryCancelNonblocking(ctx_); + } else { + FinishOnce(Status::OK); + } + } + + private: + void FinishOnce(const Status& s) { + std::lock_guard l(finish_mu_); + if (!finished_) { + Finish(s); + finished_ = true; + } + } + + void NextWrite() { + response_.set_message(request_->message() + + grpc::to_string(num_msgs_sent_)); + if (num_msgs_sent_ == server_responses_to_send_ - 1 && + server_coalescing_api_ != 0) { + num_msgs_sent_++; + StartWriteLast(&response_, WriteOptions()); + } else { + num_msgs_sent_++; + StartWrite(&response_); + } + } + ServerContext* ctx_; + const EchoRequest* request_; + EchoResponse response_; + int num_msgs_sent_{0}; + int server_try_cancel_; + int server_coalescing_api_; + int server_responses_to_send_; + std::mutex finish_mu_; + bool finished_{false}; + }; + return new Reactor; } -void TestServiceImpl::ServerTryCancel(ServerContext* context) { - EXPECT_FALSE(context->IsCancelled()); - context->TryCancel(); - gpr_log(GPR_INFO, "Server called TryCancel() to cancel the request"); - // Now wait until it's really canceled - while (!context->IsCancelled()) { - gpr_sleep_until(gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), - gpr_time_from_micros(1000, GPR_TIMESPAN))); - } +experimental::ServerBidiReactor* +CallbackTestServiceImpl::BidiStream() { + class Reactor : public ::grpc::experimental::ServerBidiReactor { + public: + Reactor() {} + void OnStarted(ServerContext* context) override { + ctx_ = context; + // If 'server_try_cancel' is set in the metadata, the RPC is cancelled by + // the server by calling ServerContext::TryCancel() depending on the + // value: + // CANCEL_BEFORE_PROCESSING: The RPC is cancelled before the server + // reads any message from the client CANCEL_DURING_PROCESSING: The RPC + // is cancelled while the server is reading messages from the client + // CANCEL_AFTER_PROCESSING: The RPC is cancelled after the server reads + // all the messages from the client + server_try_cancel_ = GetIntValueFromMetadata( + kServerTryCancelRequest, context->client_metadata(), DO_NOT_CANCEL); + server_write_last_ = GetIntValueFromMetadata( + kServerFinishAfterNReads, context->client_metadata(), 0); + if (server_try_cancel_ == CANCEL_BEFORE_PROCESSING) { + ServerTryCancelNonblocking(ctx_); + return; + } + + if (server_try_cancel_ == CANCEL_DURING_PROCESSING) { + ctx_->TryCancel(); + } + + StartRead(&request_); + } + void OnDone() override { delete this; } + void OnCancel() override { FinishOnce(Status::CANCELLED); } + void OnReadDone(bool ok) override { + if (ok) { + num_msgs_read_++; + gpr_log(GPR_INFO, "recv msg %s", request_.message().c_str()); + response_.set_message(request_.message()); + if (num_msgs_read_ == server_write_last_) { + StartWriteLast(&response_, WriteOptions()); + } else { + StartWrite(&response_); + } + } else if (server_try_cancel_ == CANCEL_DURING_PROCESSING) { + // Let OnCancel handle this + } else if (server_try_cancel_ == CANCEL_AFTER_PROCESSING) { + ServerTryCancelNonblocking(ctx_); + } else { + FinishOnce(Status::OK); + } + } + void OnWriteDone(bool ok) override { StartRead(&request_); } + + private: + void FinishOnce(const Status& s) { + std::lock_guard l(finish_mu_); + if (!finished_) { + Finish(s); + finished_ = true; + } + } + + ServerContext* ctx_; + EchoRequest request_; + EchoResponse response_; + int num_msgs_read_{0}; + int server_try_cancel_; + int server_write_last_; + std::mutex finish_mu_; + bool finished_{false}; + }; + + return new Reactor; } } // namespace testing diff --git a/test/cpp/end2end/test_service_impl.h b/test/cpp/end2end/test_service_impl.h index ddfe94487e..fad7768b87 100644 --- a/test/cpp/end2end/test_service_impl.h +++ b/test/cpp/end2end/test_service_impl.h @@ -72,13 +72,6 @@ class TestServiceImpl : public ::grpc::testing::EchoTestService::Service { } private: - int GetIntValueFromMetadata( - const char* key, - const std::multimap& metadata, - int default_value); - - void ServerTryCancel(ServerContext* context); - bool signal_client_; std::mutex mu_; std::unique_ptr host_; @@ -95,6 +88,15 @@ class CallbackTestServiceImpl EchoResponse* response, experimental::ServerCallbackRpcController* controller) override; + experimental::ServerReadReactor* RequestStream() + override; + + experimental::ServerWriteReactor* ResponseStream() + override; + + experimental::ServerBidiReactor* BidiStream() + override; + // Unimplemented is left unimplemented to test the returned error. bool signal_client() { std::unique_lock lock(mu_); @@ -106,11 +108,6 @@ class CallbackTestServiceImpl EchoResponse* response, experimental::ServerCallbackRpcController* controller); - int GetIntValueFromMetadata( - const char* key, - const std::multimap& metadata, - int default_value); - Alarm alarm_; bool signal_client_; std::mutex mu_; -- cgit v1.2.3