move logic to a superclass

include/grpcpp/impl/server_callback_handlers.h

@@ -187,9 +187,7 @@ class CallbackUnaryHandler : public grpc::internal::MethodHandler {
       ctx_->set_message_allocator_state(allocator_state);
     }
 
-    void RunAsync(absl::AnyInvocable<void()> cb) override {
-      grpc_call_run_in_event_engine(call_.call(), std::move(cb));
-    }
+    grpc_call* call() override { return call_.call(); }
 
     /// SetupReactor binds the reactor (which also releases any queued
     /// operations), maybe calls OnCancel if possible/needed, and maybe marks
@@ -375,9 +373,7 @@ class CallbackClientStreamingHandler : public grpc::internal::MethodHandler {
                              std::function<void()> call_requester)
         : ctx_(ctx), call_(*call), call_requester_(std::move(call_requester)) {}
 
-    void RunAsync(absl::AnyInvocable<void()> cb) override {
-      grpc_call_run_in_event_engine(call_.call(), std::move(cb));
-    }
+    grpc_call* call() override { return call_.call(); }
 
     void SetupReactor(ServerReadReactor<RequestType>* reactor) {
       reactor_.store(reactor, std::memory_order_relaxed);
@@ -604,9 +600,7 @@ class CallbackServerStreamingHandler : public grpc::internal::MethodHandler {
           req_(req),
           call_requester_(std::move(call_requester)) {}
 
-    void RunAsync(absl::AnyInvocable<void()> cb) override {
-      grpc_call_run_in_event_engine(call_.call(), std::move(cb));
-    }
+    grpc_call* call() override { return call_.call(); }
 
     void SetupReactor(ServerWriteReactor<ResponseType>* reactor) {
       reactor_.store(reactor, std::memory_order_relaxed);
@@ -820,9 +814,7 @@ class CallbackBidiHandler : public grpc::internal::MethodHandler {
                                    std::function<void()> call_requester)
         : ctx_(ctx), call_(*call), call_requester_(std::move(call_requester)) {}
 
-    void RunAsync(absl::AnyInvocable<void()> cb) override {
-      grpc_call_run_in_event_engine(call_.call(), std::move(cb));
-    }
+    grpc_call* call() override { return call_.call(); }
 
     void SetupReactor(ServerBidiReactor<RequestType, ResponseType>* reactor) {
       reactor_.store(reactor, std::memory_order_relaxed);

include/grpcpp/server_context.h

@@ -540,9 +540,8 @@ class ServerContextBase {
 
    private:
     void CallOnDone() override {}
-    void RunAsync(absl::AnyInvocable<void()> cb) override {
-      grpc_call_run_in_event_engine(call_, std::move(cb));
-    }
+
+    grpc_call* call() override { return call_; }
 
     grpc::internal::ServerReactor* reactor() override { return reactor_; }
 

include/grpcpp/support/server_callback.h

@@ -25,6 +25,7 @@
 
 #include "absl/functional/any_invocable.h"
 
+#include <grpc/impl/call.h>
 #include <grpcpp/impl/call.h>
 #include <grpcpp/impl/call_op_set.h>
 #include <grpcpp/impl/sync.h>
@@ -128,7 +129,12 @@ class ServerCallbackCall {
 
  private:
   virtual ServerReactor* reactor() = 0;
-  virtual void RunAsync(absl::AnyInvocable<void()> cb) = 0;
+
+  virtual grpc_call* call() = 0;
+
+  virtual void RunAsync(absl::AnyInvocable<void()> cb) {
+    grpc_call_run_in_event_engine(call(), std::move(cb));
+  }
 
   // CallOnDone performs the work required at completion of the RPC: invoking
   // the OnDone function and doing all necessary cleanup. This function is only