aboutsummaryrefslogtreecommitdiffhomepage
path: root/src
diff options
context:
space:
mode:
authorGravatar Vijay Pai <vpai@google.com>2018-09-20 16:35:33 -0700
committerGravatar GitHub <noreply@github.com>2018-09-20 16:35:33 -0700
commit3ee2919623dfcc11ac58e3e2a69c8986a2dd90eb (patch)
tree4614ea685c719cdf47b7ff8834545c3db7d1a4f4 /src
parent07308653a8ed2ec0ff7ef3a35197bf5b6caaee8b (diff)
parentda1b75b5d5a9e9d437886d324f61241808fe05ab (diff)
Merge pull request #16646 from vjpai/callback_codegen_client_unary
EXPERIMENTAL: Codegen for callback client unary calls
Diffstat (limited to 'src')
-rw-r--r--src/compiler/cpp_generator.cc148
-rw-r--r--src/core/lib/surface/completion_queue.cc34
-rw-r--r--src/core/lib/surface/completion_queue.h19
-rw-r--r--src/core/lib/surface/completion_queue_factory.cc6
-rw-r--r--src/cpp/client/channel_cc.cc10
-rw-r--r--src/cpp/common/callback_common.cc149
-rw-r--r--src/ruby/ext/grpc/rb_grpc_imports.generated.h2
7 files changed, 177 insertions, 191 deletions
diff --git a/src/compiler/cpp_generator.cc b/src/compiler/cpp_generator.cc
index 1e0c36451b..56716493dc 100644
--- a/src/compiler/cpp_generator.cc
+++ b/src/compiler/cpp_generator.cc
@@ -128,6 +128,7 @@ grpc::string GetHeaderIncludes(grpc_generator::File* file,
"");
}
static const char* headers_strs[] = {
+ "functional",
"grpcpp/impl/codegen/async_generic_service.h",
"grpcpp/impl/codegen/async_stream.h",
"grpcpp/impl/codegen/async_unary_call.h",
@@ -547,6 +548,116 @@ void PrintHeaderClientMethod(grpc_generator::Printer* printer,
}
}
+void PrintHeaderClientMethodCallbackInterfacesStart(
+ grpc_generator::Printer* printer,
+ std::map<grpc::string, grpc::string>* vars) {
+ // This declares the interface for the callback-based API. The components
+ // are pure; even though this is new (post-1.0) API, it can be pure because
+ // it is an entirely new interface that happens to be scoped within
+ // StubInterface, not new additions to StubInterface itself
+ printer->Print("class experimental_async_interface {\n");
+ // All methods in this new interface are public. There is no need for private
+ // "Raw" methods since the callback-based API returns unowned raw pointers
+ printer->Print(" public:\n");
+ printer->Indent();
+ printer->Print("virtual ~experimental_async_interface() {}\n");
+}
+
+void PrintHeaderClientMethodCallbackInterfaces(
+ grpc_generator::Printer* printer, const grpc_generator::Method* method,
+ std::map<grpc::string, grpc::string>* vars, bool is_public) {
+ // Reserve is_public for future expansion
+ assert(is_public);
+
+ (*vars)["Method"] = method->name();
+ (*vars)["Request"] = method->input_type_name();
+ (*vars)["Response"] = method->output_type_name();
+
+ if (method->NoStreaming()) {
+ printer->Print(*vars,
+ "virtual void $Method$(::grpc::ClientContext* context, "
+ "const $Request$* request, $Response$* response, "
+ "std::function<void(::grpc::Status)>) = 0;\n");
+ } else if (ClientOnlyStreaming(method)) {
+ // TODO(vjpai): Add support for client-side streaming
+ } else if (ServerOnlyStreaming(method)) {
+ // TODO(vjpai): Add support for server-side streaming
+ } else if (method->BidiStreaming()) {
+ // TODO(vjpai): Add support for bidi streaming
+ }
+}
+
+void PrintHeaderClientMethodCallbackInterfacesEnd(
+ grpc_generator::Printer* printer,
+ std::map<grpc::string, grpc::string>* vars) {
+ printer->Outdent();
+ printer->Print("};\n");
+
+ // Declare a function to give the async stub contents. It can't be pure
+ // since this is a new API in StubInterface, but it is meaningless by default
+ // (since any stub that wants to use it must have its own implementation of
+ // the callback functions therein), so make the default return value nullptr.
+ // Intentionally include the word "class" to avoid possible shadowing.
+ printer->Print(
+ "virtual class experimental_async_interface* experimental_async() { "
+ "return nullptr; }\n");
+}
+
+void PrintHeaderClientMethodCallbackStart(
+ grpc_generator::Printer* printer,
+ std::map<grpc::string, grpc::string>* vars) {
+ // This declares the stub entry for the callback-based API.
+ printer->Print("class experimental_async final :\n");
+ printer->Print(" public StubInterface::experimental_async_interface {\n");
+ printer->Print(" public:\n");
+ printer->Indent();
+}
+
+void PrintHeaderClientMethodCallback(grpc_generator::Printer* printer,
+ const grpc_generator::Method* method,
+ std::map<grpc::string, grpc::string>* vars,
+ bool is_public) {
+ // Reserve is_public for future expansion
+ assert(is_public);
+
+ (*vars)["Method"] = method->name();
+ (*vars)["Request"] = method->input_type_name();
+ (*vars)["Response"] = method->output_type_name();
+
+ if (method->NoStreaming()) {
+ printer->Print(*vars,
+ "void $Method$(::grpc::ClientContext* context, "
+ "const $Request$* request, $Response$* response, "
+ "std::function<void(::grpc::Status)>) override;\n");
+ } else if (ClientOnlyStreaming(method)) {
+ // TODO(vjpai): Add support for client-side streaming
+ } else if (ServerOnlyStreaming(method)) {
+ // TODO(vjpai): Add support for server-side streaming
+ } else if (method->BidiStreaming()) {
+ // TODO(vjpai): Add support for bidi streaming
+ }
+}
+
+void PrintHeaderClientMethodCallbackEnd(
+ grpc_generator::Printer* printer,
+ std::map<grpc::string, grpc::string>* vars) {
+ printer->Outdent();
+ printer->Print(" private:\n");
+ printer->Indent();
+ printer->Print("friend class Stub;\n");
+ printer->Print("explicit experimental_async(Stub* stub): stub_(stub) { }\n");
+ // include a function with a dummy use of stub_ to avoid an unused
+ // private member warning for service with no methods
+ printer->Print("Stub* stub() { return stub_; }\n");
+ printer->Print("Stub* stub_;\n");
+ printer->Outdent();
+ printer->Print("};\n");
+
+ printer->Print(
+ "class experimental_async_interface* experimental_async() override { "
+ "return &async_stub_; }\n");
+}
+
void PrintHeaderClientMethodData(grpc_generator::Printer* printer,
const grpc_generator::Method* method,
std::map<grpc::string, grpc::string>* vars) {
@@ -951,6 +1062,14 @@ void PrintHeaderService(grpc_generator::Printer* printer,
true);
printer->Print(service->method(i)->GetTrailingComments("//").c_str());
}
+ PrintHeaderClientMethodCallbackInterfacesStart(printer, vars);
+ for (int i = 0; i < service->method_count(); ++i) {
+ printer->Print(service->method(i)->GetLeadingComments("//").c_str());
+ PrintHeaderClientMethodCallbackInterfaces(printer, service->method(i).get(),
+ vars, true);
+ printer->Print(service->method(i)->GetTrailingComments("//").c_str());
+ }
+ PrintHeaderClientMethodCallbackInterfacesEnd(printer, vars);
printer->Outdent();
printer->Print("private:\n");
printer->Indent();
@@ -970,10 +1089,17 @@ void PrintHeaderService(grpc_generator::Printer* printer,
for (int i = 0; i < service->method_count(); ++i) {
PrintHeaderClientMethod(printer, service->method(i).get(), vars, true);
}
+ PrintHeaderClientMethodCallbackStart(printer, vars);
+ for (int i = 0; i < service->method_count(); ++i) {
+ PrintHeaderClientMethodCallback(printer, service->method(i).get(), vars,
+ true);
+ }
+ PrintHeaderClientMethodCallbackEnd(printer, vars);
printer->Outdent();
printer->Print("\n private:\n");
printer->Indent();
printer->Print("std::shared_ptr< ::grpc::ChannelInterface> channel_;\n");
+ printer->Print("class experimental_async async_stub_{this};\n");
for (int i = 0; i < service->method_count(); ++i) {
PrintHeaderClientMethod(printer, service->method(i).get(), vars, false);
}
@@ -1199,10 +1325,12 @@ grpc::string GetSourceIncludes(grpc_generator::File* file,
std::map<grpc::string, grpc::string> vars;
static const char* headers_strs[] = {
+ "functional",
"grpcpp/impl/codegen/async_stream.h",
"grpcpp/impl/codegen/async_unary_call.h",
"grpcpp/impl/codegen/channel_interface.h",
"grpcpp/impl/codegen/client_unary_call.h",
+ "grpcpp/impl/codegen/client_callback.h",
"grpcpp/impl/codegen/method_handler_impl.h",
"grpcpp/impl/codegen/rpc_service_method.h",
"grpcpp/impl/codegen/service_type.h",
@@ -1247,6 +1375,17 @@ void PrintSourceClientMethod(grpc_generator::Printer* printer,
" return ::grpc::internal::BlockingUnaryCall"
"(channel_.get(), rpcmethod_$Method$_, "
"context, request, response);\n}\n\n");
+
+ printer->Print(*vars,
+ "void $ns$$Service$::Stub::experimental_async::$Method$("
+ "::grpc::ClientContext* context, "
+ "const $Request$* request, $Response$* response, "
+ "std::function<void(::grpc::Status)> f) {\n");
+ printer->Print(*vars,
+ " return ::grpc::internal::CallbackUnaryCall"
+ "(stub_->channel_.get(), stub_->rpcmethod_$Method$_, "
+ "context, request, response, std::move(f));\n}\n\n");
+
for (auto async_prefix : async_prefixes) {
(*vars)["AsyncPrefix"] = async_prefix.prefix;
(*vars)["AsyncStart"] = async_prefix.start;
@@ -1277,6 +1416,9 @@ void PrintSourceClientMethod(grpc_generator::Printer* printer,
"rpcmethod_$Method$_, "
"context, response);\n"
"}\n\n");
+
+ // TODO(vjpai): Add callback version
+
for (auto async_prefix : async_prefixes) {
(*vars)["AsyncPrefix"] = async_prefix.prefix;
(*vars)["AsyncStart"] = async_prefix.start;
@@ -1308,6 +1450,9 @@ void PrintSourceClientMethod(grpc_generator::Printer* printer,
"rpcmethod_$Method$_, "
"context, request);\n"
"}\n\n");
+
+ // TODO(vjpai): Add callback version
+
for (auto async_prefix : async_prefixes) {
(*vars)["AsyncPrefix"] = async_prefix.prefix;
(*vars)["AsyncStart"] = async_prefix.start;
@@ -1339,6 +1484,9 @@ void PrintSourceClientMethod(grpc_generator::Printer* printer,
"rpcmethod_$Method$_, "
"context);\n"
"}\n\n");
+
+ // TODO(vjpai): Add callback version
+
for (auto async_prefix : async_prefixes) {
(*vars)["AsyncPrefix"] = async_prefix.prefix;
(*vars)["AsyncStart"] = async_prefix.start;
diff --git a/src/core/lib/surface/completion_queue.cc b/src/core/lib/surface/completion_queue.cc
index c2cf450e94..5dc9991f70 100644
--- a/src/core/lib/surface/completion_queue.cc
+++ b/src/core/lib/surface/completion_queue.cc
@@ -184,7 +184,8 @@ static const cq_poller_vtable g_poller_vtable_by_poller_type[] = {
typedef struct cq_vtable {
grpc_cq_completion_type cq_completion_type;
size_t data_size;
- void (*init)(void* data, grpc_core::CQCallbackInterface* shutdown_callback);
+ void (*init)(void* data,
+ grpc_experimental_completion_queue_functor* shutdown_callback);
void (*shutdown)(grpc_completion_queue* cq);
void (*destroy)(void* data);
bool (*begin_op)(grpc_completion_queue* cq, void* tag);
@@ -267,7 +268,7 @@ typedef struct cq_callback_data {
bool shutdown_called;
/** A callback that gets invoked when the CQ completes shutdown */
- grpc_core::CQCallbackInterface* shutdown_callback;
+ grpc_experimental_completion_queue_functor* shutdown_callback;
} cq_callback_data;
/* Completion queue structure */
@@ -333,12 +334,12 @@ static grpc_event cq_pluck(grpc_completion_queue* cq, void* tag,
gpr_timespec deadline, void* reserved);
// Note that cq_init_next and cq_init_pluck do not use the shutdown_callback
-static void cq_init_next(void* data,
- grpc_core::CQCallbackInterface* shutdown_callback);
-static void cq_init_pluck(void* data,
- grpc_core::CQCallbackInterface* shutdown_callback);
-static void cq_init_callback(void* data,
- grpc_core::CQCallbackInterface* shutdown_callback);
+static void cq_init_next(
+ void* data, grpc_experimental_completion_queue_functor* shutdown_callback);
+static void cq_init_pluck(
+ void* data, grpc_experimental_completion_queue_functor* shutdown_callback);
+static void cq_init_callback(
+ void* data, grpc_experimental_completion_queue_functor* shutdown_callback);
static void cq_destroy_next(void* data);
static void cq_destroy_pluck(void* data);
static void cq_destroy_callback(void* data);
@@ -462,7 +463,7 @@ static long cq_event_queue_num_items(grpc_cq_event_queue* q) {
grpc_completion_queue* grpc_completion_queue_create_internal(
grpc_cq_completion_type completion_type, grpc_cq_polling_type polling_type,
- grpc_core::CQCallbackInterface* shutdown_callback) {
+ grpc_experimental_completion_queue_functor* shutdown_callback) {
GPR_TIMER_SCOPE("grpc_completion_queue_create_internal", 0);
grpc_completion_queue* cq;
@@ -497,8 +498,8 @@ grpc_completion_queue* grpc_completion_queue_create_internal(
return cq;
}
-static void cq_init_next(void* data,
- grpc_core::CQCallbackInterface* shutdown_callback) {
+static void cq_init_next(
+ void* data, grpc_experimental_completion_queue_functor* shutdown_callback) {
cq_next_data* cqd = static_cast<cq_next_data*>(data);
/* Initial count is dropped by grpc_completion_queue_shutdown */
gpr_atm_no_barrier_store(&cqd->pending_events, 1);
@@ -513,8 +514,8 @@ static void cq_destroy_next(void* data) {
cq_event_queue_destroy(&cqd->queue);
}
-static void cq_init_pluck(void* data,
- grpc_core::CQCallbackInterface* shutdown_callback) {
+static void cq_init_pluck(
+ void* data, grpc_experimental_completion_queue_functor* shutdown_callback) {
cq_pluck_data* cqd = static_cast<cq_pluck_data*>(data);
/* Initial count is dropped by grpc_completion_queue_shutdown */
gpr_atm_no_barrier_store(&cqd->pending_events, 1);
@@ -532,7 +533,7 @@ static void cq_destroy_pluck(void* data) {
}
static void cq_init_callback(
- void* data, grpc_core::CQCallbackInterface* shutdown_callback) {
+ void* data, grpc_experimental_completion_queue_functor* shutdown_callback) {
cq_callback_data* cqd = static_cast<cq_callback_data*>(data);
/* Initial count is dropped by grpc_completion_queue_shutdown */
gpr_atm_no_barrier_store(&cqd->pending_events, 1);
@@ -859,7 +860,8 @@ static void cq_end_op_for_callback(
GRPC_ERROR_UNREF(error);
- (static_cast<grpc_core::CQCallbackInterface*>(tag))->Run(is_success);
+ auto* functor = static_cast<grpc_experimental_completion_queue_functor*>(tag);
+ (*functor->functor_run)(functor, is_success);
}
void grpc_cq_end_op(grpc_completion_queue* cq, void* tag, grpc_error* error,
@@ -1343,7 +1345,7 @@ static void cq_finish_shutdown_callback(grpc_completion_queue* cq) {
GPR_ASSERT(cqd->shutdown_called);
cq->poller_vtable->shutdown(POLLSET_FROM_CQ(cq), &cq->pollset_shutdown_done);
- callback->Run(true);
+ (*callback->functor_run)(callback, true);
}
static void cq_shutdown_callback(grpc_completion_queue* cq) {
diff --git a/src/core/lib/surface/completion_queue.h b/src/core/lib/surface/completion_queue.h
index a7c524d8e8..d60fe6d6ef 100644
--- a/src/core/lib/surface/completion_queue.h
+++ b/src/core/lib/surface/completion_queue.h
@@ -48,23 +48,6 @@ typedef struct grpc_cq_completion {
uintptr_t next;
} grpc_cq_completion;
-/// For callback CQs, the tag that is passed in for an operation must
-/// actually be a pointer to an implementation of the following class.
-/// When the operation completes, the tag will be typecasted from void*
-/// to grpc_core::CQCallbackInterface* and then the Run method will be
-/// invoked on it. In practice, the language binding (e.g., C++ API
-/// implementation) is responsible for providing and using an implementation
-/// of this abstract base class.
-namespace grpc_core {
-class CQCallbackInterface {
- public:
- virtual ~CQCallbackInterface() {}
- virtual void Run(bool) GRPC_ABSTRACT;
-
- GRPC_ABSTRACT_BASE_CLASS
-};
-} // namespace grpc_core
-
#ifndef NDEBUG
void grpc_cq_internal_ref(grpc_completion_queue* cc, const char* reason,
const char* file, int line);
@@ -106,6 +89,6 @@ int grpc_get_cq_poll_num(grpc_completion_queue* cc);
grpc_completion_queue* grpc_completion_queue_create_internal(
grpc_cq_completion_type completion_type, grpc_cq_polling_type polling_type,
- grpc_core::CQCallbackInterface* shutdown_callback);
+ grpc_experimental_completion_queue_functor* shutdown_callback);
#endif /* GRPC_CORE_LIB_SURFACE_COMPLETION_QUEUE_H */
diff --git a/src/core/lib/surface/completion_queue_factory.cc b/src/core/lib/surface/completion_queue_factory.cc
index ed92dd7eba..2616c156e4 100644
--- a/src/core/lib/surface/completion_queue_factory.cc
+++ b/src/core/lib/surface/completion_queue_factory.cc
@@ -31,8 +31,7 @@ static grpc_completion_queue* default_create(
const grpc_completion_queue_factory* factory,
const grpc_completion_queue_attributes* attr) {
return grpc_completion_queue_create_internal(
- attr->cq_completion_type, attr->cq_polling_type,
- static_cast<grpc_core::CQCallbackInterface*>(attr->cq_shutdown_cb));
+ attr->cq_completion_type, attr->cq_polling_type, attr->cq_shutdown_cb);
}
static grpc_completion_queue_factory_vtable default_vtable = {default_create};
@@ -73,7 +72,8 @@ grpc_completion_queue* grpc_completion_queue_create_for_pluck(void* reserved) {
}
grpc_completion_queue* grpc_completion_queue_create_for_callback(
- void* shutdown_callback, void* reserved) {
+ grpc_experimental_completion_queue_functor* shutdown_callback,
+ void* reserved) {
GPR_ASSERT(!reserved);
grpc_completion_queue_attributes attr = {
2, GRPC_CQ_CALLBACK, GRPC_CQ_DEFAULT_POLLING, shutdown_callback};
diff --git a/src/cpp/client/channel_cc.cc b/src/cpp/client/channel_cc.cc
index ad71286e05..c59059f045 100644
--- a/src/cpp/client/channel_cc.cc
+++ b/src/cpp/client/channel_cc.cc
@@ -193,17 +193,19 @@ bool Channel::WaitForStateChangeImpl(grpc_connectivity_state last_observed,
}
namespace {
-class ShutdownCallback : public grpc_core::CQCallbackInterface {
+class ShutdownCallback : public grpc_experimental_completion_queue_functor {
public:
+ ShutdownCallback() { functor_run = &ShutdownCallback::Run; }
// TakeCQ takes ownership of the cq into the shutdown callback
// so that the shutdown callback will be responsible for destroying it
void TakeCQ(CompletionQueue* cq) { cq_ = cq; }
// The Run function will get invoked by the completion queue library
// when the shutdown is actually complete
- void Run(bool) override {
- delete cq_;
- grpc_core::Delete(this);
+ static void Run(grpc_experimental_completion_queue_functor* cb, int) {
+ auto* callback = static_cast<ShutdownCallback*>(cb);
+ delete callback->cq_;
+ grpc_core::Delete(callback);
}
private:
diff --git a/src/cpp/common/callback_common.cc b/src/cpp/common/callback_common.cc
deleted file mode 100644
index a0c8eeb516..0000000000
--- a/src/cpp/common/callback_common.cc
+++ /dev/null
@@ -1,149 +0,0 @@
-/*
- *
- * Copyright 2018 gRPC authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-#include <functional>
-
-#include <grpcpp/impl/codegen/callback_common.h>
-#include <grpcpp/impl/codegen/status.h>
-
-#include "src/core/lib/gprpp/memory.h"
-#include "src/core/lib/surface/completion_queue.h"
-
-namespace grpc {
-namespace internal {
-namespace {
-
-template <class Func, class Arg>
-void CatchingCallback(Func&& func, Arg&& arg) {
-#if GRPC_ALLOW_EXCEPTIONS
- try {
- func(arg);
- } catch (...) {
- // nothing to return or change here, just don't crash the library
- }
-#else // GRPC_ALLOW_EXCEPTIONS
- func(arg);
-#endif // GRPC_ALLOW_EXCEPTIONS
-}
-
-class CallbackWithSuccessImpl : public grpc_core::CQCallbackInterface {
- public:
- static void operator delete(void* ptr, std::size_t size) {
- assert(size == sizeof(CallbackWithSuccessImpl));
- }
-
- // 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); }
-
- CallbackWithSuccessImpl(grpc_call* call, CallbackWithSuccessTag* parent,
- std::function<void(bool)> f)
- : call_(call), parent_(parent), func_(std::move(f)) {
- grpc_call_ref(call);
- }
-
- void Run(bool ok) override {
- void* ignored = parent_->ops();
- bool new_ok = ok;
- GPR_ASSERT(parent_->ops()->FinalizeResult(&ignored, &new_ok));
- GPR_ASSERT(ignored == parent_->ops());
-
- // Last use of func_, so ok to move it out for rvalue call above
- CatchingCallback(std::move(func_), ok);
-
- func_ = nullptr; // reset to clear this out for sure
- grpc_call_unref(call_);
- }
-
- private:
- grpc_call* call_;
- CallbackWithSuccessTag* parent_;
- std::function<void(bool)> func_;
-};
-
-class CallbackWithStatusImpl : public grpc_core::CQCallbackInterface {
- public:
- static void operator delete(void* ptr, std::size_t size) {
- assert(size == sizeof(CallbackWithStatusImpl));
- }
-
- // 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); }
-
- CallbackWithStatusImpl(grpc_call* call, CallbackWithStatusTag* parent,
- std::function<void(Status)> f)
- : call_(call), parent_(parent), func_(std::move(f)), status_() {
- grpc_call_ref(call);
- }
-
- void Run(bool ok) override {
- void* ignored = parent_->ops();
-
- GPR_ASSERT(parent_->ops()->FinalizeResult(&ignored, &ok));
- GPR_ASSERT(ignored == parent_->ops());
-
- // Last use of func_ or status_, so ok to move them out
- CatchingCallback(std::move(func_), std::move(status_));
-
- func_ = nullptr; // reset to clear this out for sure
- grpc_call_unref(call_);
- }
- Status* status_ptr() { return &status_; }
-
- private:
- grpc_call* call_;
- CallbackWithStatusTag* parent_;
- std::function<void(Status)> func_;
- Status status_;
-};
-
-} // namespace
-
-CallbackWithSuccessTag::CallbackWithSuccessTag(grpc_call* call,
- std::function<void(bool)> f,
- CompletionQueueTag* ops)
- : impl_(new (grpc_call_arena_alloc(call, sizeof(CallbackWithSuccessImpl)))
- CallbackWithSuccessImpl(call, this, std::move(f))),
- ops_(ops) {}
-
-void CallbackWithSuccessTag::force_run(bool ok) { impl_->Run(ok); }
-
-CallbackWithStatusTag::CallbackWithStatusTag(grpc_call* call,
- std::function<void(Status)> f,
- CompletionQueueTag* ops)
- : ops_(ops) {
- auto* impl = new (grpc_call_arena_alloc(call, sizeof(CallbackWithStatusImpl)))
- CallbackWithStatusImpl(call, this, std::move(f));
- impl_ = impl;
- status_ = impl->status_ptr();
-}
-
-void CallbackWithStatusTag::force_run(Status s) {
- *status_ = std::move(s);
- impl_->Run(true);
-}
-
-} // namespace internal
-} // namespace grpc
diff --git a/src/ruby/ext/grpc/rb_grpc_imports.generated.h b/src/ruby/ext/grpc/rb_grpc_imports.generated.h
index d00e75c326..e25d953a18 100644
--- a/src/ruby/ext/grpc/rb_grpc_imports.generated.h
+++ b/src/ruby/ext/grpc/rb_grpc_imports.generated.h
@@ -107,7 +107,7 @@ extern grpc_completion_queue_create_for_next_type grpc_completion_queue_create_f
typedef grpc_completion_queue*(*grpc_completion_queue_create_for_pluck_type)(void* reserved);
extern grpc_completion_queue_create_for_pluck_type grpc_completion_queue_create_for_pluck_import;
#define grpc_completion_queue_create_for_pluck grpc_completion_queue_create_for_pluck_import
-typedef grpc_completion_queue*(*grpc_completion_queue_create_for_callback_type)(void* shutdown_callback, void* reserved);
+typedef grpc_completion_queue*(*grpc_completion_queue_create_for_callback_type)(grpc_experimental_completion_queue_functor* shutdown_callback, void* reserved);
extern grpc_completion_queue_create_for_callback_type grpc_completion_queue_create_for_callback_import;
#define grpc_completion_queue_create_for_callback grpc_completion_queue_create_for_callback_import
typedef grpc_completion_queue*(*grpc_completion_queue_create_type)(const grpc_completion_queue_factory* factory, const grpc_completion_queue_attributes* attributes, void* reserved);