diff --git a/BUILD b/BUILD index 37a61285e3b..9c37cdd6b00 100644 --- a/BUILD +++ b/BUILD @@ -2065,7 +2065,7 @@ grpc_cc_library( "include/grpcpp/impl/codegen/byte_buffer.h", "include/grpcpp/impl/codegen/call.h", "include/grpcpp/impl/codegen/call_hook.h", - "include/grpcpp/impl/codegen/call_wrapper.h", + "include/grpcpp/impl/codegen/call_op_set.h", "include/grpcpp/impl/codegen/callback_common.h", "include/grpcpp/impl/codegen/channel_interface.h", "include/grpcpp/impl/codegen/client_callback.h", @@ -2107,7 +2107,6 @@ grpc_cc_library( name = "grpc++_codegen_base_src", srcs = [ "src/cpp/codegen/codegen_init.cc", - "src/cpp/codegen/call_wrapper.cc", ], language = "c++", deps = [ diff --git a/CMakeLists.txt b/CMakeLists.txt index faa1e7d1d6f..9dca88c63d9 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -2895,7 +2895,6 @@ add_library(grpc++ third_party/nanopb/pb_common.c third_party/nanopb/pb_decode.c third_party/nanopb/pb_encode.c - src/cpp/codegen/call_wrapper.cc src/cpp/codegen/codegen_init.cc ) @@ -3114,7 +3113,7 @@ foreach(_hdr include/grpcpp/impl/codegen/byte_buffer.h include/grpcpp/impl/codegen/call.h include/grpcpp/impl/codegen/call_hook.h - include/grpcpp/impl/codegen/call_wrapper.h + include/grpcpp/impl/codegen/call_op_set.h include/grpcpp/impl/codegen/callback_common.h include/grpcpp/impl/codegen/channel_interface.h include/grpcpp/impl/codegen/client_callback.h @@ -3263,7 +3262,6 @@ add_library(grpc++_cronet third_party/nanopb/pb_common.c third_party/nanopb/pb_decode.c third_party/nanopb/pb_encode.c - src/cpp/codegen/call_wrapper.cc src/cpp/codegen/codegen_init.cc src/core/ext/transport/chttp2/client/insecure/channel_create.cc src/core/ext/transport/chttp2/client/insecure/channel_create_posix.cc @@ -3693,7 +3691,7 @@ foreach(_hdr include/grpcpp/impl/codegen/byte_buffer.h include/grpcpp/impl/codegen/call.h include/grpcpp/impl/codegen/call_hook.h - include/grpcpp/impl/codegen/call_wrapper.h + include/grpcpp/impl/codegen/call_op_set.h include/grpcpp/impl/codegen/callback_common.h include/grpcpp/impl/codegen/channel_interface.h include/grpcpp/impl/codegen/client_callback.h @@ -4013,7 +4011,6 @@ add_library(grpc++_test_util test/cpp/util/string_ref_helper.cc test/cpp/util/subprocess.cc test/cpp/util/test_credentials_provider.cc - src/cpp/codegen/call_wrapper.cc src/cpp/codegen/codegen_init.cc ) @@ -4106,7 +4103,7 @@ foreach(_hdr include/grpcpp/impl/codegen/byte_buffer.h include/grpcpp/impl/codegen/call.h include/grpcpp/impl/codegen/call_hook.h - include/grpcpp/impl/codegen/call_wrapper.h + include/grpcpp/impl/codegen/call_op_set.h include/grpcpp/impl/codegen/callback_common.h include/grpcpp/impl/codegen/channel_interface.h include/grpcpp/impl/codegen/client_callback.h @@ -4201,7 +4198,6 @@ add_library(grpc++_test_util_unsecure test/cpp/util/byte_buffer_proto_helper.cc test/cpp/util/string_ref_helper.cc test/cpp/util/subprocess.cc - src/cpp/codegen/call_wrapper.cc src/cpp/codegen/codegen_init.cc ) @@ -4291,7 +4287,7 @@ foreach(_hdr include/grpcpp/impl/codegen/byte_buffer.h include/grpcpp/impl/codegen/call.h include/grpcpp/impl/codegen/call_hook.h - include/grpcpp/impl/codegen/call_wrapper.h + include/grpcpp/impl/codegen/call_op_set.h include/grpcpp/impl/codegen/callback_common.h include/grpcpp/impl/codegen/channel_interface.h include/grpcpp/impl/codegen/client_callback.h @@ -4402,7 +4398,6 @@ add_library(grpc++_unsecure third_party/nanopb/pb_common.c third_party/nanopb/pb_decode.c third_party/nanopb/pb_encode.c - src/cpp/codegen/call_wrapper.cc src/cpp/codegen/codegen_init.cc ) @@ -4620,7 +4615,7 @@ foreach(_hdr include/grpcpp/impl/codegen/byte_buffer.h include/grpcpp/impl/codegen/call.h include/grpcpp/impl/codegen/call_hook.h - include/grpcpp/impl/codegen/call_wrapper.h + include/grpcpp/impl/codegen/call_op_set.h include/grpcpp/impl/codegen/callback_common.h include/grpcpp/impl/codegen/channel_interface.h include/grpcpp/impl/codegen/client_callback.h @@ -12626,7 +12621,6 @@ add_executable(codegen_test_minimal ${_gRPC_PROTO_GENS_DIR}/src/proto/grpc/testing/stats.pb.h ${_gRPC_PROTO_GENS_DIR}/src/proto/grpc/testing/stats.grpc.pb.h test/cpp/codegen/codegen_test_minimal.cc - src/cpp/codegen/call_wrapper.cc src/cpp/codegen/codegen_init.cc third_party/googletest/googletest/src/gtest-all.cc third_party/googletest/googlemock/src/gmock-all.cc diff --git a/Makefile b/Makefile index e1ca8857e95..7781b3d5772 100644 --- a/Makefile +++ b/Makefile @@ -5277,7 +5277,6 @@ LIBGRPC++_SRC = \ third_party/nanopb/pb_common.c \ third_party/nanopb/pb_decode.c \ third_party/nanopb/pb_encode.c \ - src/cpp/codegen/call_wrapper.cc \ src/cpp/codegen/codegen_init.cc \ PUBLIC_HEADERS_CXX += \ @@ -5461,7 +5460,7 @@ PUBLIC_HEADERS_CXX += \ include/grpcpp/impl/codegen/byte_buffer.h \ include/grpcpp/impl/codegen/call.h \ include/grpcpp/impl/codegen/call_hook.h \ - include/grpcpp/impl/codegen/call_wrapper.h \ + include/grpcpp/impl/codegen/call_op_set.h \ include/grpcpp/impl/codegen/callback_common.h \ include/grpcpp/impl/codegen/channel_interface.h \ include/grpcpp/impl/codegen/client_callback.h \ @@ -5655,7 +5654,6 @@ LIBGRPC++_CRONET_SRC = \ third_party/nanopb/pb_common.c \ third_party/nanopb/pb_decode.c \ third_party/nanopb/pb_encode.c \ - src/cpp/codegen/call_wrapper.cc \ src/cpp/codegen/codegen_init.cc \ src/core/ext/transport/chttp2/client/insecure/channel_create.cc \ src/core/ext/transport/chttp2/client/insecure/channel_create_posix.cc \ @@ -6049,7 +6047,7 @@ PUBLIC_HEADERS_CXX += \ include/grpcpp/impl/codegen/byte_buffer.h \ include/grpcpp/impl/codegen/call.h \ include/grpcpp/impl/codegen/call_hook.h \ - include/grpcpp/impl/codegen/call_wrapper.h \ + include/grpcpp/impl/codegen/call_op_set.h \ include/grpcpp/impl/codegen/callback_common.h \ include/grpcpp/impl/codegen/channel_interface.h \ include/grpcpp/impl/codegen/client_callback.h \ @@ -6407,7 +6405,6 @@ LIBGRPC++_TEST_UTIL_SRC = \ test/cpp/util/string_ref_helper.cc \ test/cpp/util/subprocess.cc \ test/cpp/util/test_credentials_provider.cc \ - src/cpp/codegen/call_wrapper.cc \ src/cpp/codegen/codegen_init.cc \ PUBLIC_HEADERS_CXX += \ @@ -6447,7 +6444,7 @@ PUBLIC_HEADERS_CXX += \ include/grpcpp/impl/codegen/byte_buffer.h \ include/grpcpp/impl/codegen/call.h \ include/grpcpp/impl/codegen/call_hook.h \ - include/grpcpp/impl/codegen/call_wrapper.h \ + include/grpcpp/impl/codegen/call_op_set.h \ include/grpcpp/impl/codegen/callback_common.h \ include/grpcpp/impl/codegen/channel_interface.h \ include/grpcpp/impl/codegen/client_callback.h \ @@ -6557,7 +6554,6 @@ $(OBJDIR)/$(CONFIG)/test/cpp/util/create_test_channel.o: $(GENDIR)/src/proto/grp $(OBJDIR)/$(CONFIG)/test/cpp/util/string_ref_helper.o: $(GENDIR)/src/proto/grpc/channelz/channelz.pb.cc $(GENDIR)/src/proto/grpc/channelz/channelz.grpc.pb.cc $(GENDIR)/src/proto/grpc/health/v1/health.pb.cc $(GENDIR)/src/proto/grpc/health/v1/health.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/echo_messages.pb.cc $(GENDIR)/src/proto/grpc/testing/echo_messages.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/echo.pb.cc $(GENDIR)/src/proto/grpc/testing/echo.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/duplicate/echo_duplicate.pb.cc $(GENDIR)/src/proto/grpc/testing/duplicate/echo_duplicate.grpc.pb.cc $(OBJDIR)/$(CONFIG)/test/cpp/util/subprocess.o: $(GENDIR)/src/proto/grpc/channelz/channelz.pb.cc $(GENDIR)/src/proto/grpc/channelz/channelz.grpc.pb.cc $(GENDIR)/src/proto/grpc/health/v1/health.pb.cc $(GENDIR)/src/proto/grpc/health/v1/health.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/echo_messages.pb.cc $(GENDIR)/src/proto/grpc/testing/echo_messages.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/echo.pb.cc $(GENDIR)/src/proto/grpc/testing/echo.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/duplicate/echo_duplicate.pb.cc $(GENDIR)/src/proto/grpc/testing/duplicate/echo_duplicate.grpc.pb.cc $(OBJDIR)/$(CONFIG)/test/cpp/util/test_credentials_provider.o: $(GENDIR)/src/proto/grpc/channelz/channelz.pb.cc $(GENDIR)/src/proto/grpc/channelz/channelz.grpc.pb.cc $(GENDIR)/src/proto/grpc/health/v1/health.pb.cc $(GENDIR)/src/proto/grpc/health/v1/health.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/echo_messages.pb.cc $(GENDIR)/src/proto/grpc/testing/echo_messages.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/echo.pb.cc $(GENDIR)/src/proto/grpc/testing/echo.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/duplicate/echo_duplicate.pb.cc $(GENDIR)/src/proto/grpc/testing/duplicate/echo_duplicate.grpc.pb.cc -$(OBJDIR)/$(CONFIG)/src/cpp/codegen/call_wrapper.o: $(GENDIR)/src/proto/grpc/channelz/channelz.pb.cc $(GENDIR)/src/proto/grpc/channelz/channelz.grpc.pb.cc $(GENDIR)/src/proto/grpc/health/v1/health.pb.cc $(GENDIR)/src/proto/grpc/health/v1/health.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/echo_messages.pb.cc $(GENDIR)/src/proto/grpc/testing/echo_messages.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/echo.pb.cc $(GENDIR)/src/proto/grpc/testing/echo.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/duplicate/echo_duplicate.pb.cc $(GENDIR)/src/proto/grpc/testing/duplicate/echo_duplicate.grpc.pb.cc $(OBJDIR)/$(CONFIG)/src/cpp/codegen/codegen_init.o: $(GENDIR)/src/proto/grpc/channelz/channelz.pb.cc $(GENDIR)/src/proto/grpc/channelz/channelz.grpc.pb.cc $(GENDIR)/src/proto/grpc/health/v1/health.pb.cc $(GENDIR)/src/proto/grpc/health/v1/health.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/echo_messages.pb.cc $(GENDIR)/src/proto/grpc/testing/echo_messages.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/echo.pb.cc $(GENDIR)/src/proto/grpc/testing/echo.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/duplicate/echo_duplicate.pb.cc $(GENDIR)/src/proto/grpc/testing/duplicate/echo_duplicate.grpc.pb.cc @@ -6570,7 +6566,6 @@ LIBGRPC++_TEST_UTIL_UNSECURE_SRC = \ test/cpp/util/byte_buffer_proto_helper.cc \ test/cpp/util/string_ref_helper.cc \ test/cpp/util/subprocess.cc \ - src/cpp/codegen/call_wrapper.cc \ src/cpp/codegen/codegen_init.cc \ PUBLIC_HEADERS_CXX += \ @@ -6610,7 +6605,7 @@ PUBLIC_HEADERS_CXX += \ include/grpcpp/impl/codegen/byte_buffer.h \ include/grpcpp/impl/codegen/call.h \ include/grpcpp/impl/codegen/call_hook.h \ - include/grpcpp/impl/codegen/call_wrapper.h \ + include/grpcpp/impl/codegen/call_op_set.h \ include/grpcpp/impl/codegen/callback_common.h \ include/grpcpp/impl/codegen/channel_interface.h \ include/grpcpp/impl/codegen/client_callback.h \ @@ -6717,7 +6712,6 @@ $(OBJDIR)/$(CONFIG)/test/cpp/end2end/test_service_impl.o: $(GENDIR)/src/proto/gr $(OBJDIR)/$(CONFIG)/test/cpp/util/byte_buffer_proto_helper.o: $(GENDIR)/src/proto/grpc/health/v1/health.pb.cc $(GENDIR)/src/proto/grpc/health/v1/health.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/echo_messages.pb.cc $(GENDIR)/src/proto/grpc/testing/echo_messages.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/echo.pb.cc $(GENDIR)/src/proto/grpc/testing/echo.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/duplicate/echo_duplicate.pb.cc $(GENDIR)/src/proto/grpc/testing/duplicate/echo_duplicate.grpc.pb.cc $(OBJDIR)/$(CONFIG)/test/cpp/util/string_ref_helper.o: $(GENDIR)/src/proto/grpc/health/v1/health.pb.cc $(GENDIR)/src/proto/grpc/health/v1/health.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/echo_messages.pb.cc $(GENDIR)/src/proto/grpc/testing/echo_messages.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/echo.pb.cc $(GENDIR)/src/proto/grpc/testing/echo.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/duplicate/echo_duplicate.pb.cc $(GENDIR)/src/proto/grpc/testing/duplicate/echo_duplicate.grpc.pb.cc $(OBJDIR)/$(CONFIG)/test/cpp/util/subprocess.o: $(GENDIR)/src/proto/grpc/health/v1/health.pb.cc $(GENDIR)/src/proto/grpc/health/v1/health.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/echo_messages.pb.cc $(GENDIR)/src/proto/grpc/testing/echo_messages.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/echo.pb.cc $(GENDIR)/src/proto/grpc/testing/echo.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/duplicate/echo_duplicate.pb.cc $(GENDIR)/src/proto/grpc/testing/duplicate/echo_duplicate.grpc.pb.cc -$(OBJDIR)/$(CONFIG)/src/cpp/codegen/call_wrapper.o: $(GENDIR)/src/proto/grpc/health/v1/health.pb.cc $(GENDIR)/src/proto/grpc/health/v1/health.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/echo_messages.pb.cc $(GENDIR)/src/proto/grpc/testing/echo_messages.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/echo.pb.cc $(GENDIR)/src/proto/grpc/testing/echo.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/duplicate/echo_duplicate.pb.cc $(GENDIR)/src/proto/grpc/testing/duplicate/echo_duplicate.grpc.pb.cc $(OBJDIR)/$(CONFIG)/src/cpp/codegen/codegen_init.o: $(GENDIR)/src/proto/grpc/health/v1/health.pb.cc $(GENDIR)/src/proto/grpc/health/v1/health.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/echo_messages.pb.cc $(GENDIR)/src/proto/grpc/testing/echo_messages.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/echo.pb.cc $(GENDIR)/src/proto/grpc/testing/echo.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/duplicate/echo_duplicate.pb.cc $(GENDIR)/src/proto/grpc/testing/duplicate/echo_duplicate.grpc.pb.cc @@ -6761,7 +6755,6 @@ LIBGRPC++_UNSECURE_SRC = \ third_party/nanopb/pb_common.c \ third_party/nanopb/pb_decode.c \ third_party/nanopb/pb_encode.c \ - src/cpp/codegen/call_wrapper.cc \ src/cpp/codegen/codegen_init.cc \ PUBLIC_HEADERS_CXX += \ @@ -6945,7 +6938,7 @@ PUBLIC_HEADERS_CXX += \ include/grpcpp/impl/codegen/byte_buffer.h \ include/grpcpp/impl/codegen/call.h \ include/grpcpp/impl/codegen/call_hook.h \ - include/grpcpp/impl/codegen/call_wrapper.h \ + include/grpcpp/impl/codegen/call_op_set.h \ include/grpcpp/impl/codegen/callback_common.h \ include/grpcpp/impl/codegen/channel_interface.h \ include/grpcpp/impl/codegen/client_callback.h \ @@ -17466,7 +17459,6 @@ CODEGEN_TEST_MINIMAL_SRC = \ $(GENDIR)/src/proto/grpc/testing/worker_service.pb.cc $(GENDIR)/src/proto/grpc/testing/worker_service.grpc.pb.cc \ $(GENDIR)/src/proto/grpc/testing/stats.pb.cc $(GENDIR)/src/proto/grpc/testing/stats.grpc.pb.cc \ test/cpp/codegen/codegen_test_minimal.cc \ - src/cpp/codegen/call_wrapper.cc \ src/cpp/codegen/codegen_init.cc \ CODEGEN_TEST_MINIMAL_OBJS = $(addprefix $(OBJDIR)/$(CONFIG)/, $(addsuffix .o, $(basename $(CODEGEN_TEST_MINIMAL_SRC)))) @@ -17514,8 +17506,6 @@ $(OBJDIR)/$(CONFIG)/src/proto/grpc/testing/stats.o: $(LIBDIR)/$(CONFIG)/libgrpc $(OBJDIR)/$(CONFIG)/test/cpp/codegen/codegen_test_minimal.o: $(LIBDIR)/$(CONFIG)/libgrpc++_core_stats.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr.a -$(OBJDIR)/$(CONFIG)/src/cpp/codegen/call_wrapper.o: $(LIBDIR)/$(CONFIG)/libgrpc++_core_stats.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr.a - $(OBJDIR)/$(CONFIG)/src/cpp/codegen/codegen_init.o: $(LIBDIR)/$(CONFIG)/libgrpc++_core_stats.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr.a deps_codegen_test_minimal: $(CODEGEN_TEST_MINIMAL_OBJS:.o=.dep) @@ -17526,7 +17516,6 @@ ifneq ($(NO_DEPS),true) endif endif $(OBJDIR)/$(CONFIG)/test/cpp/codegen/codegen_test_minimal.o: $(GENDIR)/src/proto/grpc/testing/control.pb.cc $(GENDIR)/src/proto/grpc/testing/control.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/messages.pb.cc $(GENDIR)/src/proto/grpc/testing/messages.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/payloads.pb.cc $(GENDIR)/src/proto/grpc/testing/payloads.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/benchmark_service.pb.cc $(GENDIR)/src/proto/grpc/testing/benchmark_service.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/report_qps_scenario_service.pb.cc $(GENDIR)/src/proto/grpc/testing/report_qps_scenario_service.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/worker_service.pb.cc $(GENDIR)/src/proto/grpc/testing/worker_service.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/stats.pb.cc $(GENDIR)/src/proto/grpc/testing/stats.grpc.pb.cc -$(OBJDIR)/$(CONFIG)/src/cpp/codegen/call_wrapper.o: $(GENDIR)/src/proto/grpc/testing/control.pb.cc $(GENDIR)/src/proto/grpc/testing/control.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/messages.pb.cc $(GENDIR)/src/proto/grpc/testing/messages.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/payloads.pb.cc $(GENDIR)/src/proto/grpc/testing/payloads.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/benchmark_service.pb.cc $(GENDIR)/src/proto/grpc/testing/benchmark_service.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/report_qps_scenario_service.pb.cc $(GENDIR)/src/proto/grpc/testing/report_qps_scenario_service.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/worker_service.pb.cc $(GENDIR)/src/proto/grpc/testing/worker_service.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/stats.pb.cc $(GENDIR)/src/proto/grpc/testing/stats.grpc.pb.cc $(OBJDIR)/$(CONFIG)/src/cpp/codegen/codegen_init.o: $(GENDIR)/src/proto/grpc/testing/control.pb.cc $(GENDIR)/src/proto/grpc/testing/control.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/messages.pb.cc $(GENDIR)/src/proto/grpc/testing/messages.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/payloads.pb.cc $(GENDIR)/src/proto/grpc/testing/payloads.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/benchmark_service.pb.cc $(GENDIR)/src/proto/grpc/testing/benchmark_service.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/report_qps_scenario_service.pb.cc $(GENDIR)/src/proto/grpc/testing/report_qps_scenario_service.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/worker_service.pb.cc $(GENDIR)/src/proto/grpc/testing/worker_service.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/stats.pb.cc $(GENDIR)/src/proto/grpc/testing/stats.grpc.pb.cc diff --git a/build.yaml b/build.yaml index 5480b83cc34..6719c3e9c02 100644 --- a/build.yaml +++ b/build.yaml @@ -1222,7 +1222,7 @@ filegroups: - include/grpcpp/impl/codegen/byte_buffer.h - include/grpcpp/impl/codegen/call.h - include/grpcpp/impl/codegen/call_hook.h - - include/grpcpp/impl/codegen/call_wrapper.h + - include/grpcpp/impl/codegen/call_op_set.h - include/grpcpp/impl/codegen/callback_common.h - include/grpcpp/impl/codegen/channel_interface.h - include/grpcpp/impl/codegen/client_callback.h @@ -1259,7 +1259,6 @@ filegroups: - name: grpc++_codegen_base_src language: c++ src: - - src/cpp/codegen/call_wrapper.cc - src/cpp/codegen/codegen_init.cc uses: - grpc++_codegen_base diff --git a/gRPC-C++.podspec b/gRPC-C++.podspec index 89503d65bf7..fb5d12ea171 100644 --- a/gRPC-C++.podspec +++ b/gRPC-C++.podspec @@ -128,7 +128,7 @@ Pod::Spec.new do |s| 'include/grpcpp/impl/codegen/byte_buffer.h', 'include/grpcpp/impl/codegen/call.h', 'include/grpcpp/impl/codegen/call_hook.h', - 'include/grpcpp/impl/codegen/call_wrapper.h', + 'include/grpcpp/impl/codegen/call_op_set.h', 'include/grpcpp/impl/codegen/callback_common.h', 'include/grpcpp/impl/codegen/channel_interface.h', 'include/grpcpp/impl/codegen/client_callback.h', @@ -218,7 +218,6 @@ Pod::Spec.new do |s| 'src/cpp/util/status.cc', 'src/cpp/util/string_ref.cc', 'src/cpp/util/time_cc.cc', - 'src/cpp/codegen/call_wrapper.cc', 'src/cpp/codegen/codegen_init.cc', 'src/core/lib/gpr/alloc.h', 'src/core/lib/gpr/arena.h', diff --git a/grpc.gyp b/grpc.gyp index 9129f234927..963b8b22b1e 100644 --- a/grpc.gyp +++ b/grpc.gyp @@ -1420,7 +1420,6 @@ 'third_party/nanopb/pb_common.c', 'third_party/nanopb/pb_decode.c', 'third_party/nanopb/pb_encode.c', - 'src/cpp/codegen/call_wrapper.cc', 'src/cpp/codegen/codegen_init.cc', ], }, @@ -1501,7 +1500,6 @@ 'test/cpp/util/string_ref_helper.cc', 'test/cpp/util/subprocess.cc', 'test/cpp/util/test_credentials_provider.cc', - 'src/cpp/codegen/call_wrapper.cc', 'src/cpp/codegen/codegen_init.cc', ], }, @@ -1522,7 +1520,6 @@ 'test/cpp/util/byte_buffer_proto_helper.cc', 'test/cpp/util/string_ref_helper.cc', 'test/cpp/util/subprocess.cc', - 'src/cpp/codegen/call_wrapper.cc', 'src/cpp/codegen/codegen_init.cc', ], }, @@ -1573,7 +1570,6 @@ 'third_party/nanopb/pb_common.c', 'third_party/nanopb/pb_decode.c', 'third_party/nanopb/pb_encode.c', - 'src/cpp/codegen/call_wrapper.cc', 'src/cpp/codegen/codegen_init.cc', ], }, diff --git a/include/grpcpp/channel.h b/include/grpcpp/channel.h index 624000b75fb..14209b85eeb 100644 --- a/include/grpcpp/channel.h +++ b/include/grpcpp/channel.h @@ -20,6 +20,7 @@ #define GRPCPP_CHANNEL_H #include +#include #include #include diff --git a/include/grpcpp/impl/codegen/call.h b/include/grpcpp/impl/codegen/call.h index ef6a5178027..c040c30dd9d 100644 --- a/include/grpcpp/impl/codegen/call.h +++ b/include/grpcpp/impl/codegen/call.h @@ -1,6 +1,6 @@ /* * - * Copyright 2015 gRPC authors. + * 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. @@ -15,1291 +15,79 @@ * limitations under the License. * */ - #ifndef GRPCPP_IMPL_CODEGEN_CALL_H #define GRPCPP_IMPL_CODEGEN_CALL_H -#include -#include -#include -#include -#include -#include -#include - -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include - -#include -#include #include +#include namespace grpc { - class CompletionQueue; -extern CoreCodegenInterface* g_core_codegen_interface; +namespace experimental { +class ClientRpcInfo; +class ServerRpcInfo; +} // namespace experimental namespace internal { -class Call; class CallHook; +class CallOpSetInterface; -// TODO(yangg) if the map is changed before we send, the pointers will be a -// mess. Make sure it does not happen. -inline grpc_metadata* FillMetadataArray( - const std::multimap& metadata, - size_t* metadata_count, const grpc::string& optional_error_details) { - *metadata_count = metadata.size() + (optional_error_details.empty() ? 0 : 1); - if (*metadata_count == 0) { - return nullptr; - } - grpc_metadata* metadata_array = - (grpc_metadata*)(g_core_codegen_interface->gpr_malloc( - (*metadata_count) * sizeof(grpc_metadata))); - size_t i = 0; - for (auto iter = metadata.cbegin(); iter != metadata.cend(); ++iter, ++i) { - metadata_array[i].key = SliceReferencingString(iter->first); - metadata_array[i].value = SliceReferencingString(iter->second); - } - if (!optional_error_details.empty()) { - metadata_array[i].key = - g_core_codegen_interface->grpc_slice_from_static_buffer( - kBinaryErrorDetailsKey, sizeof(kBinaryErrorDetailsKey) - 1); - metadata_array[i].value = SliceReferencingString(optional_error_details); - } - return metadata_array; -} -} // namespace internal - -/// Per-message write options. -class WriteOptions { - public: - WriteOptions() : flags_(0), last_message_(false) {} - WriteOptions(const WriteOptions& other) - : flags_(other.flags_), last_message_(other.last_message_) {} - - /// Clear all flags. - inline void Clear() { flags_ = 0; } - - /// Returns raw flags bitset. - inline uint32_t flags() const { return flags_; } - - /// Sets flag for the disabling of compression for the next message write. - /// - /// \sa GRPC_WRITE_NO_COMPRESS - inline WriteOptions& set_no_compression() { - SetBit(GRPC_WRITE_NO_COMPRESS); - return *this; - } - - /// Clears flag for the disabling of compression for the next message write. - /// - /// \sa GRPC_WRITE_NO_COMPRESS - inline WriteOptions& clear_no_compression() { - ClearBit(GRPC_WRITE_NO_COMPRESS); - return *this; - } - - /// Get value for the flag indicating whether compression for the next - /// message write is forcefully disabled. - /// - /// \sa GRPC_WRITE_NO_COMPRESS - inline bool get_no_compression() const { - return GetBit(GRPC_WRITE_NO_COMPRESS); - } - - /// Sets flag indicating that the write may be buffered and need not go out on - /// the wire immediately. - /// - /// \sa GRPC_WRITE_BUFFER_HINT - inline WriteOptions& set_buffer_hint() { - SetBit(GRPC_WRITE_BUFFER_HINT); - return *this; - } - - /// Clears flag indicating that the write may be buffered and need not go out - /// on the wire immediately. - /// - /// \sa GRPC_WRITE_BUFFER_HINT - inline WriteOptions& clear_buffer_hint() { - ClearBit(GRPC_WRITE_BUFFER_HINT); - return *this; - } - - /// Get value for the flag indicating that the write may be buffered and need - /// not go out on the wire immediately. - /// - /// \sa GRPC_WRITE_BUFFER_HINT - inline bool get_buffer_hint() const { return GetBit(GRPC_WRITE_BUFFER_HINT); } - - /// corked bit: aliases set_buffer_hint currently, with the intent that - /// set_buffer_hint will be removed in the future - inline WriteOptions& set_corked() { - SetBit(GRPC_WRITE_BUFFER_HINT); - return *this; - } - - inline WriteOptions& clear_corked() { - ClearBit(GRPC_WRITE_BUFFER_HINT); - return *this; - } - - inline bool is_corked() const { return GetBit(GRPC_WRITE_BUFFER_HINT); } - - /// last-message bit: indicates this is the last message in a stream - /// client-side: makes Write the equivalent of performing Write, WritesDone - /// in a single step - /// server-side: hold the Write until the service handler returns (sync api) - /// or until Finish is called (async api) - inline WriteOptions& set_last_message() { - last_message_ = true; - return *this; - } - - /// Clears flag indicating that this is the last message in a stream, - /// disabling coalescing. - inline WriteOptions& clear_last_message() { - last_message_ = false; - return *this; - } - - /// Guarantee that all bytes have been written to the socket before completing - /// this write (usually writes are completed when they pass flow control). - inline WriteOptions& set_write_through() { - SetBit(GRPC_WRITE_THROUGH); - return *this; - } - - inline bool is_write_through() const { return GetBit(GRPC_WRITE_THROUGH); } - - /// Get value for the flag indicating that this is the last message, and - /// should be coalesced with trailing metadata. - /// - /// \sa GRPC_WRITE_LAST_MESSAGE - bool is_last_message() const { return last_message_; } - - WriteOptions& operator=(const WriteOptions& rhs) { - flags_ = rhs.flags_; - return *this; - } - - private: - void SetBit(const uint32_t mask) { flags_ |= mask; } - - void ClearBit(const uint32_t mask) { flags_ &= ~mask; } - - bool GetBit(const uint32_t mask) const { return (flags_ & mask) != 0; } - - uint32_t flags_; - bool last_message_; -}; - -namespace internal { - -/// Internal methods for setting the state -class InternalInterceptorBatchMethods - : public experimental::InterceptorBatchMethods { - public: - virtual ~InternalInterceptorBatchMethods() {} - - virtual void AddInterceptionHookPoint( - experimental::InterceptionHookPoints type) = 0; - - virtual void SetSendMessage(ByteBuffer* buf) = 0; - - virtual void SetSendInitialMetadata( - std::multimap* metadata) = 0; - - virtual void SetSendStatus(grpc_status_code* code, - grpc::string* error_details, - grpc::string* error_message) = 0; - - virtual void SetSendTrailingMetadata( - std::multimap* metadata) = 0; - - virtual void SetRecvMessage(void* message) = 0; - - virtual void SetRecvInitialMetadata(internal::MetadataMap* map) = 0; - - virtual void SetRecvStatus(Status* status) = 0; - - virtual void SetRecvTrailingMetadata(internal::MetadataMap* map) = 0; - - virtual std::unique_ptr GetInterceptedChannel() = 0; -}; - -/// Default argument for CallOpSet. I is unused by the class, but can be -/// used for generating multiple names for the same thing. -template -class CallNoOp { - protected: - void AddOp(grpc_op* ops, size_t* nops) {} - void FinishOp(bool* status) {} - void SetInterceptionHookPoint( - InternalInterceptorBatchMethods* interceptor_methods) {} - void SetFinishInterceptionHookPoint( - InternalInterceptorBatchMethods* interceptor_methods) {} - void SetHijackingState(InternalInterceptorBatchMethods* interceptor_methods) { - } -}; - -class CallOpSendInitialMetadata { - public: - CallOpSendInitialMetadata() : send_(false) { - maybe_compression_level_.is_set = false; - } - - void SendInitialMetadata(std::multimap* metadata, - uint32_t flags) { - maybe_compression_level_.is_set = false; - send_ = true; - flags_ = flags; - metadata_map_ = metadata; - } - - void set_compression_level(grpc_compression_level level) { - maybe_compression_level_.is_set = true; - maybe_compression_level_.level = level; - } - - protected: - void AddOp(grpc_op* ops, size_t* nops) { - if (!send_ || hijacked_) return; - grpc_op* op = &ops[(*nops)++]; - op->op = GRPC_OP_SEND_INITIAL_METADATA; - op->flags = flags_; - op->reserved = NULL; - initial_metadata_ = - FillMetadataArray(*metadata_map_, &initial_metadata_count_, ""); - op->data.send_initial_metadata.count = initial_metadata_count_; - op->data.send_initial_metadata.metadata = initial_metadata_; - op->data.send_initial_metadata.maybe_compression_level.is_set = - maybe_compression_level_.is_set; - if (maybe_compression_level_.is_set) { - op->data.send_initial_metadata.maybe_compression_level.level = - maybe_compression_level_.level; - } - } - void FinishOp(bool* status) { - if (!send_ || hijacked_) return; - g_core_codegen_interface->gpr_free(initial_metadata_); - send_ = false; - } - - void SetInterceptionHookPoint( - InternalInterceptorBatchMethods* interceptor_methods) { - if (!send_) return; - interceptor_methods->AddInterceptionHookPoint( - experimental::InterceptionHookPoints::PRE_SEND_INITIAL_METADATA); - interceptor_methods->SetSendInitialMetadata(metadata_map_); - } - - void SetFinishInterceptionHookPoint( - InternalInterceptorBatchMethods* interceptor_methods) {} - - void SetHijackingState(InternalInterceptorBatchMethods* interceptor_methods) { - hijacked_ = true; - } - - bool hijacked_ = false; - bool send_; - uint32_t flags_; - size_t initial_metadata_count_; - std::multimap* metadata_map_; - grpc_metadata* initial_metadata_; - struct { - bool is_set; - grpc_compression_level level; - } maybe_compression_level_; -}; - -class CallOpSendMessage { - public: - CallOpSendMessage() : send_buf_() {} - - /// Send \a message using \a options for the write. The \a options are cleared - /// after use. - template - Status SendMessage(const M& message, - WriteOptions options) GRPC_MUST_USE_RESULT; - - template - Status SendMessage(const M& message) GRPC_MUST_USE_RESULT; - - protected: - void AddOp(grpc_op* ops, size_t* nops) { - if (!send_buf_.Valid() || hijacked_) return; - grpc_op* op = &ops[(*nops)++]; - op->op = GRPC_OP_SEND_MESSAGE; - op->flags = write_options_.flags(); - op->reserved = NULL; - op->data.send_message.send_message = send_buf_.c_buffer(); - // Flags are per-message: clear them after use. - write_options_.Clear(); - } - void FinishOp(bool* status) { send_buf_.Clear(); } - - void SetInterceptionHookPoint( - InternalInterceptorBatchMethods* interceptor_methods) { - if (!send_buf_.Valid()) return; - interceptor_methods->AddInterceptionHookPoint( - experimental::InterceptionHookPoints::PRE_SEND_MESSAGE); - interceptor_methods->SetSendMessage(&send_buf_); - } - - void SetFinishInterceptionHookPoint( - InternalInterceptorBatchMethods* interceptor_methods) {} - - void SetHijackingState(InternalInterceptorBatchMethods* interceptor_methods) { - hijacked_ = true; - } - - private: - bool hijacked_ = false; - ByteBuffer send_buf_; - WriteOptions write_options_; -}; - -template -Status CallOpSendMessage::SendMessage(const M& message, WriteOptions options) { - write_options_ = options; - bool own_buf; - // TODO(vjpai): Remove the void below when possible - // The void in the template parameter below should not be needed - // (since it should be implicit) but is needed due to an observed - // difference in behavior between clang and gcc for certain internal users - Status result = SerializationTraits::Serialize( - message, send_buf_.bbuf_ptr(), &own_buf); - if (!own_buf) { - send_buf_.Duplicate(); - } - return result; -} - -template -Status CallOpSendMessage::SendMessage(const M& message) { - return SendMessage(message, WriteOptions()); -} - -template -class CallOpRecvMessage { - public: - CallOpRecvMessage() - : got_message(false), - message_(nullptr), - allow_not_getting_message_(false) {} - - void RecvMessage(R* message) { message_ = message; } - - // Do not change status if no message is received. - void AllowNoMessage() { allow_not_getting_message_ = true; } - - bool got_message; - - protected: - void AddOp(grpc_op* ops, size_t* nops) { - if (message_ == nullptr || hijacked_) return; - grpc_op* op = &ops[(*nops)++]; - op->op = GRPC_OP_RECV_MESSAGE; - op->flags = 0; - op->reserved = NULL; - op->data.recv_message.recv_message = recv_buf_.c_buffer_ptr(); - } - - void FinishOp(bool* status) { - if (message_ == nullptr || hijacked_) return; - if (recv_buf_.Valid()) { - if (*status) { - got_message = *status = - SerializationTraits::Deserialize(recv_buf_.bbuf_ptr(), message_) - .ok(); - recv_buf_.Release(); - } else { - got_message = false; - recv_buf_.Clear(); - } - } else { - got_message = false; - if (!allow_not_getting_message_) { - *status = false; - } - } - message_ = nullptr; - } - - void SetInterceptionHookPoint( - InternalInterceptorBatchMethods* interceptor_methods) { - interceptor_methods->SetRecvMessage(message_); - } - - void SetFinishInterceptionHookPoint( - InternalInterceptorBatchMethods* interceptor_methods) { - if (!got_message) return; - interceptor_methods->AddInterceptionHookPoint( - experimental::InterceptionHookPoints::POST_RECV_MESSAGE); - } - void SetHijackingState(InternalInterceptorBatchMethods* interceptor_methods) { - hijacked_ = true; - if (message_ == nullptr) return; - interceptor_methods->AddInterceptionHookPoint( - experimental::InterceptionHookPoints::PRE_RECV_MESSAGE); - got_message = true; - } - - private: - R* message_; - ByteBuffer recv_buf_; - bool allow_not_getting_message_; - bool hijacked_ = false; -}; - -class DeserializeFunc { - public: - virtual Status Deserialize(ByteBuffer* buf) = 0; - virtual ~DeserializeFunc() {} -}; - -template -class DeserializeFuncType final : public DeserializeFunc { - public: - DeserializeFuncType(R* message) : message_(message) {} - Status Deserialize(ByteBuffer* buf) override { - return SerializationTraits::Deserialize(buf->bbuf_ptr(), message_); - } - - ~DeserializeFuncType() override {} - - private: - R* message_; // Not a managed pointer because management is external to this -}; - -class CallOpGenericRecvMessage { - public: - CallOpGenericRecvMessage() - : got_message(false), allow_not_getting_message_(false) {} - - template - void RecvMessage(R* message) { - // Use an explicit base class pointer to avoid resolution error in the - // following unique_ptr::reset for some old implementations. - DeserializeFunc* func = new DeserializeFuncType(message); - deserialize_.reset(func); - message_ = message; - } - - // Do not change status if no message is received. - void AllowNoMessage() { allow_not_getting_message_ = true; } - - bool got_message; - - protected: - void AddOp(grpc_op* ops, size_t* nops) { - if (!deserialize_ || hijacked_) return; - grpc_op* op = &ops[(*nops)++]; - op->op = GRPC_OP_RECV_MESSAGE; - op->flags = 0; - op->reserved = NULL; - op->data.recv_message.recv_message = recv_buf_.c_buffer_ptr(); - } - - void FinishOp(bool* status) { - if (!deserialize_ || hijacked_) return; - if (recv_buf_.Valid()) { - if (*status) { - got_message = true; - *status = deserialize_->Deserialize(&recv_buf_).ok(); - recv_buf_.Release(); - } else { - got_message = false; - recv_buf_.Clear(); - } - } else { - got_message = false; - if (!allow_not_getting_message_) { - *status = false; - } - } - deserialize_.reset(); - } - - void SetInterceptionHookPoint( - InternalInterceptorBatchMethods* interceptor_methods) { - interceptor_methods->SetRecvMessage(message_); - } - - void SetFinishInterceptionHookPoint( - InternalInterceptorBatchMethods* interceptor_methods) { - if (!got_message) return; - interceptor_methods->AddInterceptionHookPoint( - experimental::InterceptionHookPoints::POST_RECV_MESSAGE); - } - void SetHijackingState(InternalInterceptorBatchMethods* interceptor_methods) { - hijacked_ = true; - if (!deserialize_) return; - interceptor_methods->AddInterceptionHookPoint( - experimental::InterceptionHookPoints::PRE_RECV_MESSAGE); - } - - private: - void* message_; - bool hijacked_ = false; - std::unique_ptr deserialize_; - ByteBuffer recv_buf_; - bool allow_not_getting_message_; -}; - -class CallOpClientSendClose { - public: - CallOpClientSendClose() : send_(false) {} - - void ClientSendClose() { send_ = true; } - - protected: - void AddOp(grpc_op* ops, size_t* nops) { - if (!send_ || hijacked_) return; - grpc_op* op = &ops[(*nops)++]; - op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT; - op->flags = 0; - op->reserved = NULL; - } - void FinishOp(bool* status) { send_ = false; } - - void SetInterceptionHookPoint( - InternalInterceptorBatchMethods* interceptor_methods) { - if (!send_) return; - interceptor_methods->AddInterceptionHookPoint( - experimental::InterceptionHookPoints::PRE_SEND_CLOSE); - } - - void SetFinishInterceptionHookPoint( - InternalInterceptorBatchMethods* interceptor_methods) {} - - void SetHijackingState(InternalInterceptorBatchMethods* interceptor_methods) { - hijacked_ = true; - } - - private: - bool hijacked_ = false; - bool send_; -}; - -class CallOpServerSendStatus { - public: - CallOpServerSendStatus() : send_status_available_(false) {} - - void ServerSendStatus( - std::multimap* trailing_metadata, - const Status& status) { - send_error_details_ = status.error_details(); - metadata_map_ = trailing_metadata; - send_status_available_ = true; - send_status_code_ = static_cast(status.error_code()); - send_error_message_ = status.error_message(); - } - - protected: - void AddOp(grpc_op* ops, size_t* nops) { - if (!send_status_available_ || hijacked_) return; - trailing_metadata_ = FillMetadataArray( - *metadata_map_, &trailing_metadata_count_, send_error_details_); - grpc_op* op = &ops[(*nops)++]; - op->op = GRPC_OP_SEND_STATUS_FROM_SERVER; - op->data.send_status_from_server.trailing_metadata_count = - trailing_metadata_count_; - op->data.send_status_from_server.trailing_metadata = trailing_metadata_; - op->data.send_status_from_server.status = send_status_code_; - error_message_slice_ = SliceReferencingString(send_error_message_); - op->data.send_status_from_server.status_details = - send_error_message_.empty() ? nullptr : &error_message_slice_; - op->flags = 0; - op->reserved = NULL; - } - - void FinishOp(bool* status) { - if (!send_status_available_ || hijacked_) return; - g_core_codegen_interface->gpr_free(trailing_metadata_); - send_status_available_ = false; - } - - void SetInterceptionHookPoint( - InternalInterceptorBatchMethods* interceptor_methods) { - if (!send_status_available_) return; - interceptor_methods->AddInterceptionHookPoint( - experimental::InterceptionHookPoints::PRE_SEND_STATUS); - interceptor_methods->SetSendTrailingMetadata(metadata_map_); - interceptor_methods->SetSendStatus(&send_status_code_, &send_error_details_, - &send_error_message_); - } - - void SetFinishInterceptionHookPoint( - InternalInterceptorBatchMethods* interceptor_methods) {} - - void SetHijackingState(InternalInterceptorBatchMethods* interceptor_methods) { - hijacked_ = true; - } - - private: - bool hijacked_ = false; - bool send_status_available_; - grpc_status_code send_status_code_; - grpc::string send_error_details_; - grpc::string send_error_message_; - size_t trailing_metadata_count_; - std::multimap* metadata_map_; - grpc_metadata* trailing_metadata_; - grpc_slice error_message_slice_; -}; - -class CallOpRecvInitialMetadata { +/// Straightforward wrapping of the C call object +class Call final { public: - CallOpRecvInitialMetadata() : metadata_map_(nullptr) {} - - void RecvInitialMetadata(ClientContext* context) { - context->initial_metadata_received_ = true; - metadata_map_ = &context->recv_initial_metadata_; - } + Call() + : call_hook_(nullptr), + cq_(nullptr), + call_(nullptr), + max_receive_message_size_(-1) {} + /** call is owned by the caller */ + Call(grpc_call* call, CallHook* call_hook, CompletionQueue* cq) + : call_hook_(call_hook), + cq_(cq), + call_(call), + max_receive_message_size_(-1) {} - protected: - void AddOp(grpc_op* ops, size_t* nops) { - if (metadata_map_ == nullptr || hijacked_) return; - grpc_op* op = &ops[(*nops)++]; - op->op = GRPC_OP_RECV_INITIAL_METADATA; - op->data.recv_initial_metadata.recv_initial_metadata = metadata_map_->arr(); - op->flags = 0; - op->reserved = NULL; - } - - void FinishOp(bool* status) { - if (metadata_map_ == nullptr || hijacked_) return; - } - - void SetInterceptionHookPoint( - InternalInterceptorBatchMethods* interceptor_methods) { - interceptor_methods->SetRecvInitialMetadata(metadata_map_); - } - - void SetFinishInterceptionHookPoint( - InternalInterceptorBatchMethods* interceptor_methods) { - if (metadata_map_ == nullptr) return; - interceptor_methods->AddInterceptionHookPoint( - experimental::InterceptionHookPoints::POST_RECV_INITIAL_METADATA); - metadata_map_ = nullptr; - } - - void SetHijackingState(InternalInterceptorBatchMethods* interceptor_methods) { - hijacked_ = true; - if (metadata_map_ == nullptr) return; - interceptor_methods->AddInterceptionHookPoint( - experimental::InterceptionHookPoints::PRE_RECV_INITIAL_METADATA); - } - - private: - bool hijacked_ = false; - MetadataMap* metadata_map_; -}; - -class CallOpClientRecvStatus { - public: - CallOpClientRecvStatus() - : recv_status_(nullptr), debug_error_string_(nullptr) {} + Call(grpc_call* call, CallHook* call_hook, CompletionQueue* cq, + experimental::ClientRpcInfo* rpc_info) + : call_hook_(call_hook), + cq_(cq), + call_(call), + max_receive_message_size_(-1), + client_rpc_info_(rpc_info) {} - void ClientRecvStatus(ClientContext* context, Status* status) { - client_context_ = context; - metadata_map_ = &client_context_->trailing_metadata_; - recv_status_ = status; - error_message_ = g_core_codegen_interface->grpc_empty_slice(); - } + Call(grpc_call* call, CallHook* call_hook, CompletionQueue* cq, + int max_receive_message_size, experimental::ServerRpcInfo* rpc_info) + : call_hook_(call_hook), + cq_(cq), + call_(call), + max_receive_message_size_(max_receive_message_size), + server_rpc_info_(rpc_info) {} - protected: - void AddOp(grpc_op* ops, size_t* nops) { - if (recv_status_ == nullptr || hijacked_) return; - grpc_op* op = &ops[(*nops)++]; - op->op = GRPC_OP_RECV_STATUS_ON_CLIENT; - op->data.recv_status_on_client.trailing_metadata = metadata_map_->arr(); - op->data.recv_status_on_client.status = &status_code_; - op->data.recv_status_on_client.status_details = &error_message_; - op->data.recv_status_on_client.error_string = &debug_error_string_; - op->flags = 0; - op->reserved = NULL; + void PerformOps(CallOpSetInterface* ops) { + call_hook_->PerformOpsOnCall(ops, this); } - void FinishOp(bool* status) { - if (recv_status_ == nullptr || hijacked_) return; - grpc::string binary_error_details = metadata_map_->GetBinaryErrorDetails(); - *recv_status_ = - Status(static_cast(status_code_), - GRPC_SLICE_IS_EMPTY(error_message_) - ? grpc::string() - : grpc::string(GRPC_SLICE_START_PTR(error_message_), - GRPC_SLICE_END_PTR(error_message_)), - binary_error_details); - client_context_->set_debug_error_string( - debug_error_string_ != nullptr ? debug_error_string_ : ""); - g_core_codegen_interface->grpc_slice_unref(error_message_); - if (debug_error_string_ != nullptr) { - g_core_codegen_interface->gpr_free((void*)debug_error_string_); - } - } + grpc_call* call() const { return call_; } + CompletionQueue* cq() const { return cq_; } - void SetInterceptionHookPoint( - InternalInterceptorBatchMethods* interceptor_methods) { - interceptor_methods->SetRecvStatus(recv_status_); - interceptor_methods->SetRecvTrailingMetadata(metadata_map_); - } + int max_receive_message_size() const { return max_receive_message_size_; } - void SetFinishInterceptionHookPoint( - InternalInterceptorBatchMethods* interceptor_methods) { - if (recv_status_ == nullptr) return; - interceptor_methods->AddInterceptionHookPoint( - experimental::InterceptionHookPoints::POST_RECV_STATUS); - recv_status_ = nullptr; + experimental::ClientRpcInfo* client_rpc_info() const { + return client_rpc_info_; } - void SetHijackingState(InternalInterceptorBatchMethods* interceptor_methods) { - hijacked_ = true; - if (recv_status_ == nullptr) return; - interceptor_methods->AddInterceptionHookPoint( - experimental::InterceptionHookPoints::PRE_RECV_STATUS); + experimental::ServerRpcInfo* server_rpc_info() const { + return server_rpc_info_; } private: - bool hijacked_ = false; - ClientContext* client_context_; - MetadataMap* metadata_map_; - Status* recv_status_; - const char* debug_error_string_; - grpc_status_code status_code_; - grpc_slice error_message_; -}; - -/// An abstract collection of call ops, used to generate the -/// grpc_call_op structure to pass down to the lower layers, -/// and as it is-a CompletionQueueTag, also massages the final -/// completion into the correct form for consumption in the C++ -/// API. -class CallOpSetInterface : public CompletionQueueTag { - public: - /// Fills in grpc_op, starting from ops[*nops] and moving - /// upwards. - 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 - /// want core to process the tag differently (e.g., as a core callback) - virtual void* 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. - virtual void SetHijackingState() = 0; - - // Should be called after interceptors are done running - virtual void ContinueFillOpsAfterInterception() = 0; - - // Should be called after interceptors are done running on the finalize result - // path - virtual void ContinueFinalizeResultAfterInterception() = 0; + CallHook* call_hook_; + CompletionQueue* cq_; + grpc_call* call_; + int max_receive_message_size_; + experimental::ClientRpcInfo* client_rpc_info_ = nullptr; + experimental::ServerRpcInfo* server_rpc_info_ = nullptr; }; - -template , class Op2 = CallNoOp<2>, - class Op3 = CallNoOp<3>, class Op4 = CallNoOp<4>, - class Op5 = CallNoOp<5>, class Op6 = CallNoOp<6>> -class CallOpSet; - -class InterceptorBatchMethodsImpl : public InternalInterceptorBatchMethods { - public: - InterceptorBatchMethodsImpl() { - for (auto i = 0; - i < static_cast( - experimental::InterceptionHookPoints::NUM_INTERCEPTION_HOOKS); - i++) { - hooks_[i] = false; - } - } - - virtual ~InterceptorBatchMethodsImpl() {} - - virtual bool QueryInterceptionHookPoint( - experimental::InterceptionHookPoints type) override { - return hooks_[static_cast(type)]; - } - - virtual void Proceed() override { /* fill this */ - if (call_->client_rpc_info() != nullptr) { - return ProceedClient(); - } - GPR_CODEGEN_ASSERT(call_->server_rpc_info() != nullptr); - ProceedServer(); - } - - virtual void Hijack() override { - // Only the client can hijack when sending down initial metadata - GPR_CODEGEN_ASSERT(!reverse_ && ops_ != nullptr && - call_->client_rpc_info() != nullptr); - // It is illegal to call Hijack twice - GPR_CODEGEN_ASSERT(!ran_hijacking_interceptor_); - auto* rpc_info = call_->client_rpc_info(); - rpc_info->hijacked_ = true; - rpc_info->hijacked_interceptor_ = curr_iteration_; - ClearHookPoints(); - ops_->SetHijackingState(); - ran_hijacking_interceptor_ = true; - rpc_info->RunInterceptor(this, curr_iteration_); - } - - virtual void AddInterceptionHookPoint( - experimental::InterceptionHookPoints type) override { - hooks_[static_cast(type)] = true; - } - - virtual ByteBuffer* GetSendMessage() override { return send_message_; } - - virtual std::multimap* GetSendInitialMetadata() - override { - return send_initial_metadata_; - } - - virtual Status GetSendStatus() override { - return Status(static_cast(*code_), *error_message_, - *error_details_); - } - - virtual void ModifySendStatus(const Status& status) override { - *code_ = static_cast(status.error_code()); - *error_details_ = status.error_details(); - *error_message_ = status.error_message(); - } - - virtual std::multimap* GetSendTrailingMetadata() - override { - return send_trailing_metadata_; - } - - virtual void* GetRecvMessage() override { return recv_message_; } - - virtual std::multimap* - GetRecvInitialMetadata() override { - return recv_initial_metadata_->map(); - } - - virtual Status* GetRecvStatus() override { return recv_status_; } - - virtual std::multimap* - GetRecvTrailingMetadata() override { - return recv_trailing_metadata_->map(); - } - - virtual void SetSendMessage(ByteBuffer* buf) override { send_message_ = buf; } - - virtual void SetSendInitialMetadata( - std::multimap* metadata) override { - send_initial_metadata_ = metadata; - } - - virtual void SetSendStatus(grpc_status_code* code, - grpc::string* error_details, - grpc::string* error_message) override { - code_ = code; - error_details_ = error_details; - error_message_ = error_message; - } - - virtual void SetSendTrailingMetadata( - std::multimap* metadata) override { - send_trailing_metadata_ = metadata; - } - - virtual void SetRecvMessage(void* message) override { - recv_message_ = message; - } - - virtual void SetRecvInitialMetadata(internal::MetadataMap* map) override { - recv_initial_metadata_ = map; - } - - virtual void SetRecvStatus(Status* status) override { recv_status_ = status; } - - virtual void SetRecvTrailingMetadata(internal::MetadataMap* map) override { - recv_trailing_metadata_ = map; - } - - virtual std::unique_ptr GetInterceptedChannel() override { - auto* info = call_->client_rpc_info(); - if (info == nullptr) { - return std::unique_ptr(nullptr); - } - // The intercepted channel starts from the interceptor just after the - // current interceptor - return std::unique_ptr(new internal::InterceptedChannel( - reinterpret_cast(info->channel()), - curr_iteration_ + 1)); - } - - // Clears all state - void ClearState() { - reverse_ = false; - ran_hijacking_interceptor_ = false; - ClearHookPoints(); - } - - // Prepares for Post_recv operations - void SetReverse() { - reverse_ = true; - ran_hijacking_interceptor_ = false; - ClearHookPoints(); - } - - // This needs to be set before interceptors are run - void SetCall(Call* call) { call_ = call; } - - // This needs to be set before interceptors are run using RunInterceptors(). - // Alternatively, RunInterceptors(std::function f) can be used. - void SetCallOpSetInterface(CallOpSetInterface* ops) { ops_ = ops; } - - // Returns true if no interceptors are run. This should be used only by - // subclasses of CallOpSetInterface. SetCall and SetCallOpSetInterface should - // have been called before this. After all the interceptors are done running, - // either ContinueFillOpsAfterInterception or - // ContinueFinalizeOpsAfterInterception will be called. Note that neither of - // them is invoked if there were no interceptors registered. - bool RunInterceptors() { - GPR_CODEGEN_ASSERT(ops_); - auto* client_rpc_info = call_->client_rpc_info(); - if (client_rpc_info != nullptr) { - if (client_rpc_info->interceptors_.size() == 0) { - return true; - } else { - RunClientInterceptors(); - return false; - } - } - - auto* server_rpc_info = call_->server_rpc_info(); - if (server_rpc_info == nullptr || - server_rpc_info->interceptors_.size() == 0) { - return true; - } - RunServerInterceptors(); - return false; - } - - // Returns true if no interceptors are run. Returns false otherwise if there - // are interceptors registered. After the interceptors are done running \a f - // will be invoked. This is to be used only by BaseAsyncRequest and - // SyncRequest. - bool RunInterceptors(std::function f) { - // This is used only by the server for initial call request - GPR_CODEGEN_ASSERT(reverse_ == true); - GPR_CODEGEN_ASSERT(call_->client_rpc_info() == nullptr); - auto* server_rpc_info = call_->server_rpc_info(); - if (server_rpc_info == nullptr || - server_rpc_info->interceptors_.size() == 0) { - return true; - } - callback_ = std::move(f); - RunServerInterceptors(); - return false; - } - - private: - void RunClientInterceptors() { - auto* rpc_info = call_->client_rpc_info(); - if (!reverse_) { - curr_iteration_ = 0; - } else { - if (rpc_info->hijacked_) { - curr_iteration_ = rpc_info->hijacked_interceptor_; - } else { - curr_iteration_ = rpc_info->interceptors_.size() - 1; - } - } - rpc_info->RunInterceptor(this, curr_iteration_); - } - - void RunServerInterceptors() { - auto* rpc_info = call_->server_rpc_info(); - if (!reverse_) { - curr_iteration_ = 0; - } else { - curr_iteration_ = rpc_info->interceptors_.size() - 1; - } - rpc_info->RunInterceptor(this, curr_iteration_); - } - - void ProceedClient() { - auto* rpc_info = call_->client_rpc_info(); - if (rpc_info->hijacked_ && !reverse_ && - curr_iteration_ == rpc_info->hijacked_interceptor_ && - !ran_hijacking_interceptor_) { - // We now need to provide hijacked recv ops to this interceptor - ClearHookPoints(); - ops_->SetHijackingState(); - ran_hijacking_interceptor_ = true; - rpc_info->RunInterceptor(this, curr_iteration_); - return; - } - if (!reverse_) { - curr_iteration_++; - // We are going down the stack of interceptors - if (curr_iteration_ < static_cast(rpc_info->interceptors_.size())) { - if (rpc_info->hijacked_ && - curr_iteration_ > rpc_info->hijacked_interceptor_) { - // This is a hijacked RPC and we are done with hijacking - ops_->ContinueFillOpsAfterInterception(); - } else { - rpc_info->RunInterceptor(this, curr_iteration_); - } - } else { - // we are done running all the interceptors without any hijacking - ops_->ContinueFillOpsAfterInterception(); - } - } else { - curr_iteration_--; - // We are going up the stack of interceptors - if (curr_iteration_ >= 0) { - // Continue running interceptors - rpc_info->RunInterceptor(this, curr_iteration_); - } else { - // we are done running all the interceptors without any hijacking - ops_->ContinueFinalizeResultAfterInterception(); - } - } - } - - void ProceedServer() { - auto* rpc_info = call_->server_rpc_info(); - if (!reverse_) { - curr_iteration_++; - if (curr_iteration_ < static_cast(rpc_info->interceptors_.size())) { - return rpc_info->RunInterceptor(this, curr_iteration_); - } else if (ops_) { - return ops_->ContinueFillOpsAfterInterception(); - } - } else { - curr_iteration_--; - // We are going up the stack of interceptors - if (curr_iteration_ >= 0) { - // Continue running interceptors - return rpc_info->RunInterceptor(this, curr_iteration_); - } else if (ops_) { - return ops_->ContinueFinalizeResultAfterInterception(); - } - } - GPR_CODEGEN_ASSERT(callback_); - callback_(); - } - - void ClearHookPoints() { - for (auto i = 0; - i < static_cast( - experimental::InterceptionHookPoints::NUM_INTERCEPTION_HOOKS); - i++) { - hooks_[i] = false; - } - } - - std::array( - experimental::InterceptionHookPoints::NUM_INTERCEPTION_HOOKS)> - hooks_; - - int curr_iteration_ = 0; // Current iterator - bool reverse_ = false; - bool ran_hijacking_interceptor_ = false; - Call* call_ = - nullptr; // The Call object is present along with CallOpSet object - CallOpSetInterface* ops_ = nullptr; - std::function callback_; - - ByteBuffer* send_message_ = nullptr; - - std::multimap* send_initial_metadata_; - - grpc_status_code* code_ = nullptr; - grpc::string* error_details_ = nullptr; - grpc::string* error_message_ = nullptr; - Status send_status_; - - std::multimap* send_trailing_metadata_ = nullptr; - - void* recv_message_ = nullptr; - - internal::MetadataMap* recv_initial_metadata_ = nullptr; - - Status* recv_status_ = nullptr; - - internal::MetadataMap* recv_trailing_metadata_ = nullptr; -}; - -/// Primary implementation of CallOpSetInterface. -/// Since we cannot use variadic templates, we declare slots up to -/// the maximum count of ops we'll need in a set. We leverage the -/// empty base class optimization to slim this class (especially -/// when there are many unused slots used). To avoid duplicate base classes, -/// the template parmeter for CallNoOp is varied by argument position. -template -class CallOpSet : public CallOpSetInterface, - public Op1, - public Op2, - public Op3, - public Op4, - public Op5, - public Op6 { - public: - CallOpSet() : cq_tag_(this), return_tag_(this) {} - // The copy constructor and assignment operator reset the value of - // cq_tag_ and return_tag_ since those are only meaningful on a specific - // object, not across objects. - CallOpSet(const CallOpSet& other) - : cq_tag_(this), - return_tag_(this), - call_(other.call_), - done_intercepting_(false), - interceptor_methods_(InterceptorBatchMethodsImpl()) {} - - CallOpSet& operator=(const CallOpSet& other) { - cq_tag_ = this; - return_tag_ = this; - call_ = other.call_; - done_intercepting_ = false; - interceptor_methods_ = InterceptorBatchMethodsImpl(); - return *this; - } - - void FillOps(Call* call) override { - done_intercepting_ = false; - g_core_codegen_interface->grpc_call_ref(call->call()); - call_ = - *call; // It's fine to create a copy of call since it's just pointers - - if (RunInterceptors()) { - ContinueFillOpsAfterInterception(); - } else { - // After the interceptors are run, ContinueFillOpsAfterInterception will - // be run - } - } - - bool FinalizeResult(void** tag, bool* status) override { - if (done_intercepting_) { - // We have already finished intercepting and filling in the results. This - // round trip from the core needed to be made because interceptors were - // run - *tag = return_tag_; - *status = saved_status_; - g_core_codegen_interface->grpc_call_unref(call_.call()); - return true; - } - - this->Op1::FinishOp(status); - this->Op2::FinishOp(status); - this->Op3::FinishOp(status); - this->Op4::FinishOp(status); - this->Op5::FinishOp(status); - this->Op6::FinishOp(status); - saved_status_ = *status; - if (RunInterceptorsPostRecv()) { - *tag = return_tag_; - g_core_codegen_interface->grpc_call_unref(call_.call()); - return true; - } - // Interceptors are going to be run, so we can't return the tag just yet. - // After the interceptors are run, ContinueFinalizeResultAfterInterception - return false; - } - - void set_output_tag(void* return_tag) { return_tag_ = return_tag; } - - void* cq_tag() override { return cq_tag_; } - - /// set_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; } - - // 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. - void SetHijackingState() override { - this->Op1::SetHijackingState(&interceptor_methods_); - this->Op2::SetHijackingState(&interceptor_methods_); - this->Op3::SetHijackingState(&interceptor_methods_); - this->Op4::SetHijackingState(&interceptor_methods_); - this->Op5::SetHijackingState(&interceptor_methods_); - this->Op6::SetHijackingState(&interceptor_methods_); - } - - // Should be called after interceptors are done running - void ContinueFillOpsAfterInterception() override { - static const size_t MAX_OPS = 6; - grpc_op ops[MAX_OPS]; - size_t nops = 0; - this->Op1::AddOp(ops, &nops); - this->Op2::AddOp(ops, &nops); - this->Op3::AddOp(ops, &nops); - this->Op4::AddOp(ops, &nops); - this->Op5::AddOp(ops, &nops); - 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)); - } - - // Should be called after interceptors are done running on the finalize result - // path - void ContinueFinalizeResultAfterInterception() override { - done_intercepting_ = true; - GPR_CODEGEN_ASSERT(GRPC_CALL_OK == - g_core_codegen_interface->grpc_call_start_batch( - call_.call(), nullptr, 0, cq_tag(), nullptr)); - } - - private: - // Returns true if no interceptors need to be run - bool RunInterceptors() { - interceptor_methods_.ClearState(); - interceptor_methods_.SetCallOpSetInterface(this); - interceptor_methods_.SetCall(&call_); - this->Op1::SetInterceptionHookPoint(&interceptor_methods_); - this->Op2::SetInterceptionHookPoint(&interceptor_methods_); - this->Op3::SetInterceptionHookPoint(&interceptor_methods_); - this->Op4::SetInterceptionHookPoint(&interceptor_methods_); - this->Op5::SetInterceptionHookPoint(&interceptor_methods_); - this->Op6::SetInterceptionHookPoint(&interceptor_methods_); - return interceptor_methods_.RunInterceptors(); - } - // Returns true if no interceptors need to be run - bool RunInterceptorsPostRecv() { - // Call and OpSet had already been set on the set state. - // SetReverse also clears previously set hook points - interceptor_methods_.SetReverse(); - this->Op1::SetFinishInterceptionHookPoint(&interceptor_methods_); - this->Op2::SetFinishInterceptionHookPoint(&interceptor_methods_); - this->Op3::SetFinishInterceptionHookPoint(&interceptor_methods_); - this->Op4::SetFinishInterceptionHookPoint(&interceptor_methods_); - this->Op5::SetFinishInterceptionHookPoint(&interceptor_methods_); - this->Op6::SetFinishInterceptionHookPoint(&interceptor_methods_); - return interceptor_methods_.RunInterceptors(); - } - - void* cq_tag_; - void* return_tag_; - Call call_; - bool done_intercepting_ = false; - InterceptorBatchMethodsImpl interceptor_methods_; - bool saved_status_; -}; - } // namespace internal } // namespace grpc diff --git a/include/grpcpp/impl/codegen/call_op_set.h b/include/grpcpp/impl/codegen/call_op_set.h new file mode 100644 index 00000000000..c7149081114 --- /dev/null +++ b/include/grpcpp/impl/codegen/call_op_set.h @@ -0,0 +1,1306 @@ +/* + * + * Copyright 2015 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +#ifndef GRPCPP_IMPL_CODEGEN_CALL_OP_SET_H +#define GRPCPP_IMPL_CODEGEN_CALL_OP_SET_H + +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include + +namespace grpc { + +class CompletionQueue; +extern CoreCodegenInterface* g_core_codegen_interface; + +namespace internal { +class Call; +class CallHook; + +// TODO(yangg) if the map is changed before we send, the pointers will be a +// mess. Make sure it does not happen. +inline grpc_metadata* FillMetadataArray( + const std::multimap& metadata, + size_t* metadata_count, const grpc::string& optional_error_details) { + *metadata_count = metadata.size() + (optional_error_details.empty() ? 0 : 1); + if (*metadata_count == 0) { + return nullptr; + } + grpc_metadata* metadata_array = + (grpc_metadata*)(g_core_codegen_interface->gpr_malloc( + (*metadata_count) * sizeof(grpc_metadata))); + size_t i = 0; + for (auto iter = metadata.cbegin(); iter != metadata.cend(); ++iter, ++i) { + metadata_array[i].key = SliceReferencingString(iter->first); + metadata_array[i].value = SliceReferencingString(iter->second); + } + if (!optional_error_details.empty()) { + metadata_array[i].key = + g_core_codegen_interface->grpc_slice_from_static_buffer( + kBinaryErrorDetailsKey, sizeof(kBinaryErrorDetailsKey) - 1); + metadata_array[i].value = SliceReferencingString(optional_error_details); + } + return metadata_array; +} +} // namespace internal + +/// Per-message write options. +class WriteOptions { + public: + WriteOptions() : flags_(0), last_message_(false) {} + WriteOptions(const WriteOptions& other) + : flags_(other.flags_), last_message_(other.last_message_) {} + + /// Clear all flags. + inline void Clear() { flags_ = 0; } + + /// Returns raw flags bitset. + inline uint32_t flags() const { return flags_; } + + /// Sets flag for the disabling of compression for the next message write. + /// + /// \sa GRPC_WRITE_NO_COMPRESS + inline WriteOptions& set_no_compression() { + SetBit(GRPC_WRITE_NO_COMPRESS); + return *this; + } + + /// Clears flag for the disabling of compression for the next message write. + /// + /// \sa GRPC_WRITE_NO_COMPRESS + inline WriteOptions& clear_no_compression() { + ClearBit(GRPC_WRITE_NO_COMPRESS); + return *this; + } + + /// Get value for the flag indicating whether compression for the next + /// message write is forcefully disabled. + /// + /// \sa GRPC_WRITE_NO_COMPRESS + inline bool get_no_compression() const { + return GetBit(GRPC_WRITE_NO_COMPRESS); + } + + /// Sets flag indicating that the write may be buffered and need not go out on + /// the wire immediately. + /// + /// \sa GRPC_WRITE_BUFFER_HINT + inline WriteOptions& set_buffer_hint() { + SetBit(GRPC_WRITE_BUFFER_HINT); + return *this; + } + + /// Clears flag indicating that the write may be buffered and need not go out + /// on the wire immediately. + /// + /// \sa GRPC_WRITE_BUFFER_HINT + inline WriteOptions& clear_buffer_hint() { + ClearBit(GRPC_WRITE_BUFFER_HINT); + return *this; + } + + /// Get value for the flag indicating that the write may be buffered and need + /// not go out on the wire immediately. + /// + /// \sa GRPC_WRITE_BUFFER_HINT + inline bool get_buffer_hint() const { return GetBit(GRPC_WRITE_BUFFER_HINT); } + + /// corked bit: aliases set_buffer_hint currently, with the intent that + /// set_buffer_hint will be removed in the future + inline WriteOptions& set_corked() { + SetBit(GRPC_WRITE_BUFFER_HINT); + return *this; + } + + inline WriteOptions& clear_corked() { + ClearBit(GRPC_WRITE_BUFFER_HINT); + return *this; + } + + inline bool is_corked() const { return GetBit(GRPC_WRITE_BUFFER_HINT); } + + /// last-message bit: indicates this is the last message in a stream + /// client-side: makes Write the equivalent of performing Write, WritesDone + /// in a single step + /// server-side: hold the Write until the service handler returns (sync api) + /// or until Finish is called (async api) + inline WriteOptions& set_last_message() { + last_message_ = true; + return *this; + } + + /// Clears flag indicating that this is the last message in a stream, + /// disabling coalescing. + inline WriteOptions& clear_last_message() { + last_message_ = false; + return *this; + } + + /// Guarantee that all bytes have been written to the socket before completing + /// this write (usually writes are completed when they pass flow control). + inline WriteOptions& set_write_through() { + SetBit(GRPC_WRITE_THROUGH); + return *this; + } + + inline bool is_write_through() const { return GetBit(GRPC_WRITE_THROUGH); } + + /// Get value for the flag indicating that this is the last message, and + /// should be coalesced with trailing metadata. + /// + /// \sa GRPC_WRITE_LAST_MESSAGE + bool is_last_message() const { return last_message_; } + + WriteOptions& operator=(const WriteOptions& rhs) { + flags_ = rhs.flags_; + return *this; + } + + private: + void SetBit(const uint32_t mask) { flags_ |= mask; } + + void ClearBit(const uint32_t mask) { flags_ &= ~mask; } + + bool GetBit(const uint32_t mask) const { return (flags_ & mask) != 0; } + + uint32_t flags_; + bool last_message_; +}; + +namespace internal { + +/// Internal methods for setting the state +class InternalInterceptorBatchMethods + : public experimental::InterceptorBatchMethods { + public: + virtual ~InternalInterceptorBatchMethods() {} + + virtual void AddInterceptionHookPoint( + experimental::InterceptionHookPoints type) = 0; + + virtual void SetSendMessage(ByteBuffer* buf) = 0; + + virtual void SetSendInitialMetadata( + std::multimap* metadata) = 0; + + virtual void SetSendStatus(grpc_status_code* code, + grpc::string* error_details, + grpc::string* error_message) = 0; + + virtual void SetSendTrailingMetadata( + std::multimap* metadata) = 0; + + virtual void SetRecvMessage(void* message) = 0; + + virtual void SetRecvInitialMetadata(internal::MetadataMap* map) = 0; + + virtual void SetRecvStatus(Status* status) = 0; + + virtual void SetRecvTrailingMetadata(internal::MetadataMap* map) = 0; + + virtual std::unique_ptr GetInterceptedChannel() = 0; +}; + +/// Default argument for CallOpSet. I is unused by the class, but can be +/// used for generating multiple names for the same thing. +template +class CallNoOp { + protected: + void AddOp(grpc_op* ops, size_t* nops) {} + void FinishOp(bool* status) {} + void SetInterceptionHookPoint( + InternalInterceptorBatchMethods* interceptor_methods) {} + void SetFinishInterceptionHookPoint( + InternalInterceptorBatchMethods* interceptor_methods) {} + void SetHijackingState(InternalInterceptorBatchMethods* interceptor_methods) { + } +}; + +class CallOpSendInitialMetadata { + public: + CallOpSendInitialMetadata() : send_(false) { + maybe_compression_level_.is_set = false; + } + + void SendInitialMetadata(std::multimap* metadata, + uint32_t flags) { + maybe_compression_level_.is_set = false; + send_ = true; + flags_ = flags; + metadata_map_ = metadata; + } + + void set_compression_level(grpc_compression_level level) { + maybe_compression_level_.is_set = true; + maybe_compression_level_.level = level; + } + + protected: + void AddOp(grpc_op* ops, size_t* nops) { + if (!send_ || hijacked_) return; + grpc_op* op = &ops[(*nops)++]; + op->op = GRPC_OP_SEND_INITIAL_METADATA; + op->flags = flags_; + op->reserved = NULL; + initial_metadata_ = + FillMetadataArray(*metadata_map_, &initial_metadata_count_, ""); + op->data.send_initial_metadata.count = initial_metadata_count_; + op->data.send_initial_metadata.metadata = initial_metadata_; + op->data.send_initial_metadata.maybe_compression_level.is_set = + maybe_compression_level_.is_set; + if (maybe_compression_level_.is_set) { + op->data.send_initial_metadata.maybe_compression_level.level = + maybe_compression_level_.level; + } + } + void FinishOp(bool* status) { + if (!send_ || hijacked_) return; + g_core_codegen_interface->gpr_free(initial_metadata_); + send_ = false; + } + + void SetInterceptionHookPoint( + InternalInterceptorBatchMethods* interceptor_methods) { + if (!send_) return; + interceptor_methods->AddInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_SEND_INITIAL_METADATA); + interceptor_methods->SetSendInitialMetadata(metadata_map_); + } + + void SetFinishInterceptionHookPoint( + InternalInterceptorBatchMethods* interceptor_methods) {} + + void SetHijackingState(InternalInterceptorBatchMethods* interceptor_methods) { + hijacked_ = true; + } + + bool hijacked_ = false; + bool send_; + uint32_t flags_; + size_t initial_metadata_count_; + std::multimap* metadata_map_; + grpc_metadata* initial_metadata_; + struct { + bool is_set; + grpc_compression_level level; + } maybe_compression_level_; +}; + +class CallOpSendMessage { + public: + CallOpSendMessage() : send_buf_() {} + + /// Send \a message using \a options for the write. The \a options are cleared + /// after use. + template + Status SendMessage(const M& message, + WriteOptions options) GRPC_MUST_USE_RESULT; + + template + Status SendMessage(const M& message) GRPC_MUST_USE_RESULT; + + protected: + void AddOp(grpc_op* ops, size_t* nops) { + if (!send_buf_.Valid() || hijacked_) return; + grpc_op* op = &ops[(*nops)++]; + op->op = GRPC_OP_SEND_MESSAGE; + op->flags = write_options_.flags(); + op->reserved = NULL; + op->data.send_message.send_message = send_buf_.c_buffer(); + // Flags are per-message: clear them after use. + write_options_.Clear(); + } + void FinishOp(bool* status) { send_buf_.Clear(); } + + void SetInterceptionHookPoint( + InternalInterceptorBatchMethods* interceptor_methods) { + if (!send_buf_.Valid()) return; + interceptor_methods->AddInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_SEND_MESSAGE); + interceptor_methods->SetSendMessage(&send_buf_); + } + + void SetFinishInterceptionHookPoint( + InternalInterceptorBatchMethods* interceptor_methods) {} + + void SetHijackingState(InternalInterceptorBatchMethods* interceptor_methods) { + hijacked_ = true; + } + + private: + bool hijacked_ = false; + ByteBuffer send_buf_; + WriteOptions write_options_; +}; + +template +Status CallOpSendMessage::SendMessage(const M& message, WriteOptions options) { + write_options_ = options; + bool own_buf; + // TODO(vjpai): Remove the void below when possible + // The void in the template parameter below should not be needed + // (since it should be implicit) but is needed due to an observed + // difference in behavior between clang and gcc for certain internal users + Status result = SerializationTraits::Serialize( + message, send_buf_.bbuf_ptr(), &own_buf); + if (!own_buf) { + send_buf_.Duplicate(); + } + return result; +} + +template +Status CallOpSendMessage::SendMessage(const M& message) { + return SendMessage(message, WriteOptions()); +} + +template +class CallOpRecvMessage { + public: + CallOpRecvMessage() + : got_message(false), + message_(nullptr), + allow_not_getting_message_(false) {} + + void RecvMessage(R* message) { message_ = message; } + + // Do not change status if no message is received. + void AllowNoMessage() { allow_not_getting_message_ = true; } + + bool got_message; + + protected: + void AddOp(grpc_op* ops, size_t* nops) { + if (message_ == nullptr || hijacked_) return; + grpc_op* op = &ops[(*nops)++]; + op->op = GRPC_OP_RECV_MESSAGE; + op->flags = 0; + op->reserved = NULL; + op->data.recv_message.recv_message = recv_buf_.c_buffer_ptr(); + } + + void FinishOp(bool* status) { + if (message_ == nullptr || hijacked_) return; + if (recv_buf_.Valid()) { + if (*status) { + got_message = *status = + SerializationTraits::Deserialize(recv_buf_.bbuf_ptr(), message_) + .ok(); + recv_buf_.Release(); + } else { + got_message = false; + recv_buf_.Clear(); + } + } else { + got_message = false; + if (!allow_not_getting_message_) { + *status = false; + } + } + message_ = nullptr; + } + + void SetInterceptionHookPoint( + InternalInterceptorBatchMethods* interceptor_methods) { + interceptor_methods->SetRecvMessage(message_); + } + + void SetFinishInterceptionHookPoint( + InternalInterceptorBatchMethods* interceptor_methods) { + if (!got_message) return; + interceptor_methods->AddInterceptionHookPoint( + experimental::InterceptionHookPoints::POST_RECV_MESSAGE); + } + void SetHijackingState(InternalInterceptorBatchMethods* interceptor_methods) { + hijacked_ = true; + if (message_ == nullptr) return; + interceptor_methods->AddInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_RECV_MESSAGE); + got_message = true; + } + + private: + R* message_; + ByteBuffer recv_buf_; + bool allow_not_getting_message_; + bool hijacked_ = false; +}; + +class DeserializeFunc { + public: + virtual Status Deserialize(ByteBuffer* buf) = 0; + virtual ~DeserializeFunc() {} +}; + +template +class DeserializeFuncType final : public DeserializeFunc { + public: + DeserializeFuncType(R* message) : message_(message) {} + Status Deserialize(ByteBuffer* buf) override { + return SerializationTraits::Deserialize(buf->bbuf_ptr(), message_); + } + + ~DeserializeFuncType() override {} + + private: + R* message_; // Not a managed pointer because management is external to this +}; + +class CallOpGenericRecvMessage { + public: + CallOpGenericRecvMessage() + : got_message(false), allow_not_getting_message_(false) {} + + template + void RecvMessage(R* message) { + // Use an explicit base class pointer to avoid resolution error in the + // following unique_ptr::reset for some old implementations. + DeserializeFunc* func = new DeserializeFuncType(message); + deserialize_.reset(func); + message_ = message; + } + + // Do not change status if no message is received. + void AllowNoMessage() { allow_not_getting_message_ = true; } + + bool got_message; + + protected: + void AddOp(grpc_op* ops, size_t* nops) { + if (!deserialize_ || hijacked_) return; + grpc_op* op = &ops[(*nops)++]; + op->op = GRPC_OP_RECV_MESSAGE; + op->flags = 0; + op->reserved = NULL; + op->data.recv_message.recv_message = recv_buf_.c_buffer_ptr(); + } + + void FinishOp(bool* status) { + if (!deserialize_ || hijacked_) return; + if (recv_buf_.Valid()) { + if (*status) { + got_message = true; + *status = deserialize_->Deserialize(&recv_buf_).ok(); + recv_buf_.Release(); + } else { + got_message = false; + recv_buf_.Clear(); + } + } else { + got_message = false; + if (!allow_not_getting_message_) { + *status = false; + } + } + deserialize_.reset(); + } + + void SetInterceptionHookPoint( + InternalInterceptorBatchMethods* interceptor_methods) { + interceptor_methods->SetRecvMessage(message_); + } + + void SetFinishInterceptionHookPoint( + InternalInterceptorBatchMethods* interceptor_methods) { + if (!got_message) return; + interceptor_methods->AddInterceptionHookPoint( + experimental::InterceptionHookPoints::POST_RECV_MESSAGE); + } + void SetHijackingState(InternalInterceptorBatchMethods* interceptor_methods) { + hijacked_ = true; + if (!deserialize_) return; + interceptor_methods->AddInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_RECV_MESSAGE); + } + + private: + void* message_; + bool hijacked_ = false; + std::unique_ptr deserialize_; + ByteBuffer recv_buf_; + bool allow_not_getting_message_; +}; + +class CallOpClientSendClose { + public: + CallOpClientSendClose() : send_(false) {} + + void ClientSendClose() { send_ = true; } + + protected: + void AddOp(grpc_op* ops, size_t* nops) { + if (!send_ || hijacked_) return; + grpc_op* op = &ops[(*nops)++]; + op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT; + op->flags = 0; + op->reserved = NULL; + } + void FinishOp(bool* status) { send_ = false; } + + void SetInterceptionHookPoint( + InternalInterceptorBatchMethods* interceptor_methods) { + if (!send_) return; + interceptor_methods->AddInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_SEND_CLOSE); + } + + void SetFinishInterceptionHookPoint( + InternalInterceptorBatchMethods* interceptor_methods) {} + + void SetHijackingState(InternalInterceptorBatchMethods* interceptor_methods) { + hijacked_ = true; + } + + private: + bool hijacked_ = false; + bool send_; +}; + +class CallOpServerSendStatus { + public: + CallOpServerSendStatus() : send_status_available_(false) {} + + void ServerSendStatus( + std::multimap* trailing_metadata, + const Status& status) { + send_error_details_ = status.error_details(); + metadata_map_ = trailing_metadata; + send_status_available_ = true; + send_status_code_ = static_cast(status.error_code()); + send_error_message_ = status.error_message(); + } + + protected: + void AddOp(grpc_op* ops, size_t* nops) { + if (!send_status_available_ || hijacked_) return; + trailing_metadata_ = FillMetadataArray( + *metadata_map_, &trailing_metadata_count_, send_error_details_); + grpc_op* op = &ops[(*nops)++]; + op->op = GRPC_OP_SEND_STATUS_FROM_SERVER; + op->data.send_status_from_server.trailing_metadata_count = + trailing_metadata_count_; + op->data.send_status_from_server.trailing_metadata = trailing_metadata_; + op->data.send_status_from_server.status = send_status_code_; + error_message_slice_ = SliceReferencingString(send_error_message_); + op->data.send_status_from_server.status_details = + send_error_message_.empty() ? nullptr : &error_message_slice_; + op->flags = 0; + op->reserved = NULL; + } + + void FinishOp(bool* status) { + if (!send_status_available_ || hijacked_) return; + g_core_codegen_interface->gpr_free(trailing_metadata_); + send_status_available_ = false; + } + + void SetInterceptionHookPoint( + InternalInterceptorBatchMethods* interceptor_methods) { + if (!send_status_available_) return; + interceptor_methods->AddInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_SEND_STATUS); + interceptor_methods->SetSendTrailingMetadata(metadata_map_); + interceptor_methods->SetSendStatus(&send_status_code_, &send_error_details_, + &send_error_message_); + } + + void SetFinishInterceptionHookPoint( + InternalInterceptorBatchMethods* interceptor_methods) {} + + void SetHijackingState(InternalInterceptorBatchMethods* interceptor_methods) { + hijacked_ = true; + } + + private: + bool hijacked_ = false; + bool send_status_available_; + grpc_status_code send_status_code_; + grpc::string send_error_details_; + grpc::string send_error_message_; + size_t trailing_metadata_count_; + std::multimap* metadata_map_; + grpc_metadata* trailing_metadata_; + grpc_slice error_message_slice_; +}; + +class CallOpRecvInitialMetadata { + public: + CallOpRecvInitialMetadata() : metadata_map_(nullptr) {} + + void RecvInitialMetadata(ClientContext* context) { + context->initial_metadata_received_ = true; + metadata_map_ = &context->recv_initial_metadata_; + } + + protected: + void AddOp(grpc_op* ops, size_t* nops) { + if (metadata_map_ == nullptr || hijacked_) return; + grpc_op* op = &ops[(*nops)++]; + op->op = GRPC_OP_RECV_INITIAL_METADATA; + op->data.recv_initial_metadata.recv_initial_metadata = metadata_map_->arr(); + op->flags = 0; + op->reserved = NULL; + } + + void FinishOp(bool* status) { + if (metadata_map_ == nullptr || hijacked_) return; + } + + void SetInterceptionHookPoint( + InternalInterceptorBatchMethods* interceptor_methods) { + interceptor_methods->SetRecvInitialMetadata(metadata_map_); + } + + void SetFinishInterceptionHookPoint( + InternalInterceptorBatchMethods* interceptor_methods) { + if (metadata_map_ == nullptr) return; + interceptor_methods->AddInterceptionHookPoint( + experimental::InterceptionHookPoints::POST_RECV_INITIAL_METADATA); + metadata_map_ = nullptr; + } + + void SetHijackingState(InternalInterceptorBatchMethods* interceptor_methods) { + hijacked_ = true; + if (metadata_map_ == nullptr) return; + interceptor_methods->AddInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_RECV_INITIAL_METADATA); + } + + private: + bool hijacked_ = false; + MetadataMap* metadata_map_; +}; + +class CallOpClientRecvStatus { + public: + CallOpClientRecvStatus() + : recv_status_(nullptr), debug_error_string_(nullptr) {} + + void ClientRecvStatus(ClientContext* context, Status* status) { + client_context_ = context; + metadata_map_ = &client_context_->trailing_metadata_; + recv_status_ = status; + error_message_ = g_core_codegen_interface->grpc_empty_slice(); + } + + protected: + void AddOp(grpc_op* ops, size_t* nops) { + if (recv_status_ == nullptr || hijacked_) return; + grpc_op* op = &ops[(*nops)++]; + op->op = GRPC_OP_RECV_STATUS_ON_CLIENT; + op->data.recv_status_on_client.trailing_metadata = metadata_map_->arr(); + op->data.recv_status_on_client.status = &status_code_; + op->data.recv_status_on_client.status_details = &error_message_; + op->data.recv_status_on_client.error_string = &debug_error_string_; + op->flags = 0; + op->reserved = NULL; + } + + void FinishOp(bool* status) { + if (recv_status_ == nullptr || hijacked_) return; + grpc::string binary_error_details = metadata_map_->GetBinaryErrorDetails(); + *recv_status_ = + Status(static_cast(status_code_), + GRPC_SLICE_IS_EMPTY(error_message_) + ? grpc::string() + : grpc::string(GRPC_SLICE_START_PTR(error_message_), + GRPC_SLICE_END_PTR(error_message_)), + binary_error_details); + client_context_->set_debug_error_string( + debug_error_string_ != nullptr ? debug_error_string_ : ""); + g_core_codegen_interface->grpc_slice_unref(error_message_); + if (debug_error_string_ != nullptr) { + g_core_codegen_interface->gpr_free((void*)debug_error_string_); + } + } + + void SetInterceptionHookPoint( + InternalInterceptorBatchMethods* interceptor_methods) { + interceptor_methods->SetRecvStatus(recv_status_); + interceptor_methods->SetRecvTrailingMetadata(metadata_map_); + } + + void SetFinishInterceptionHookPoint( + InternalInterceptorBatchMethods* interceptor_methods) { + if (recv_status_ == nullptr) return; + interceptor_methods->AddInterceptionHookPoint( + experimental::InterceptionHookPoints::POST_RECV_STATUS); + recv_status_ = nullptr; + } + + void SetHijackingState(InternalInterceptorBatchMethods* interceptor_methods) { + hijacked_ = true; + if (recv_status_ == nullptr) return; + interceptor_methods->AddInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_RECV_STATUS); + } + + private: + bool hijacked_ = false; + ClientContext* client_context_; + MetadataMap* metadata_map_; + Status* recv_status_; + const char* debug_error_string_; + grpc_status_code status_code_; + grpc_slice error_message_; +}; + +/// An abstract collection of call ops, used to generate the +/// grpc_call_op structure to pass down to the lower layers, +/// and as it is-a CompletionQueueTag, also massages the final +/// completion into the correct form for consumption in the C++ +/// API. +class CallOpSetInterface : public CompletionQueueTag { + public: + /// Fills in grpc_op, starting from ops[*nops] and moving + /// upwards. + 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 + /// want core to process the tag differently (e.g., as a core callback) + virtual void* 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. + virtual void SetHijackingState() = 0; + + // Should be called after interceptors are done running + virtual void ContinueFillOpsAfterInterception() = 0; + + // Should be called after interceptors are done running on the finalize result + // path + virtual void ContinueFinalizeResultAfterInterception() = 0; +}; + +template , class Op2 = CallNoOp<2>, + class Op3 = CallNoOp<3>, class Op4 = CallNoOp<4>, + class Op5 = CallNoOp<5>, class Op6 = CallNoOp<6>> +class CallOpSet; + +class InterceptorBatchMethodsImpl : public InternalInterceptorBatchMethods { + public: + InterceptorBatchMethodsImpl() { + for (auto i = 0; + i < static_cast( + experimental::InterceptionHookPoints::NUM_INTERCEPTION_HOOKS); + i++) { + hooks_[i] = false; + } + } + + virtual ~InterceptorBatchMethodsImpl() {} + + virtual bool QueryInterceptionHookPoint( + experimental::InterceptionHookPoints type) override { + return hooks_[static_cast(type)]; + } + + virtual void Proceed() override { /* fill this */ + if (call_->client_rpc_info() != nullptr) { + return ProceedClient(); + } + GPR_CODEGEN_ASSERT(call_->server_rpc_info() != nullptr); + ProceedServer(); + } + + virtual void Hijack() override { + // Only the client can hijack when sending down initial metadata + GPR_CODEGEN_ASSERT(!reverse_ && ops_ != nullptr && + call_->client_rpc_info() != nullptr); + // It is illegal to call Hijack twice + GPR_CODEGEN_ASSERT(!ran_hijacking_interceptor_); + auto* rpc_info = call_->client_rpc_info(); + rpc_info->hijacked_ = true; + rpc_info->hijacked_interceptor_ = curr_iteration_; + ClearHookPoints(); + ops_->SetHijackingState(); + ran_hijacking_interceptor_ = true; + rpc_info->RunInterceptor(this, curr_iteration_); + } + + virtual void AddInterceptionHookPoint( + experimental::InterceptionHookPoints type) override { + hooks_[static_cast(type)] = true; + } + + virtual ByteBuffer* GetSendMessage() override { return send_message_; } + + virtual std::multimap* GetSendInitialMetadata() + override { + return send_initial_metadata_; + } + + virtual Status GetSendStatus() override { + return Status(static_cast(*code_), *error_message_, + *error_details_); + } + + virtual void ModifySendStatus(const Status& status) override { + *code_ = static_cast(status.error_code()); + *error_details_ = status.error_details(); + *error_message_ = status.error_message(); + } + + virtual std::multimap* GetSendTrailingMetadata() + override { + return send_trailing_metadata_; + } + + virtual void* GetRecvMessage() override { return recv_message_; } + + virtual std::multimap* + GetRecvInitialMetadata() override { + return recv_initial_metadata_->map(); + } + + virtual Status* GetRecvStatus() override { return recv_status_; } + + virtual std::multimap* + GetRecvTrailingMetadata() override { + return recv_trailing_metadata_->map(); + } + + virtual void SetSendMessage(ByteBuffer* buf) override { send_message_ = buf; } + + virtual void SetSendInitialMetadata( + std::multimap* metadata) override { + send_initial_metadata_ = metadata; + } + + virtual void SetSendStatus(grpc_status_code* code, + grpc::string* error_details, + grpc::string* error_message) override { + code_ = code; + error_details_ = error_details; + error_message_ = error_message; + } + + virtual void SetSendTrailingMetadata( + std::multimap* metadata) override { + send_trailing_metadata_ = metadata; + } + + virtual void SetRecvMessage(void* message) override { + recv_message_ = message; + } + + virtual void SetRecvInitialMetadata(internal::MetadataMap* map) override { + recv_initial_metadata_ = map; + } + + virtual void SetRecvStatus(Status* status) override { recv_status_ = status; } + + virtual void SetRecvTrailingMetadata(internal::MetadataMap* map) override { + recv_trailing_metadata_ = map; + } + + virtual std::unique_ptr GetInterceptedChannel() override { + auto* info = call_->client_rpc_info(); + if (info == nullptr) { + return std::unique_ptr(nullptr); + } + // The intercepted channel starts from the interceptor just after the + // current interceptor + return std::unique_ptr(new internal::InterceptedChannel( + reinterpret_cast(info->channel()), + curr_iteration_ + 1)); + } + + // Clears all state + void ClearState() { + reverse_ = false; + ran_hijacking_interceptor_ = false; + ClearHookPoints(); + } + + // Prepares for Post_recv operations + void SetReverse() { + reverse_ = true; + ran_hijacking_interceptor_ = false; + ClearHookPoints(); + } + + // This needs to be set before interceptors are run + void SetCall(Call* call) { call_ = call; } + + // This needs to be set before interceptors are run using RunInterceptors(). + // Alternatively, RunInterceptors(std::function f) can be used. + void SetCallOpSetInterface(CallOpSetInterface* ops) { ops_ = ops; } + + // Returns true if no interceptors are run. This should be used only by + // subclasses of CallOpSetInterface. SetCall and SetCallOpSetInterface should + // have been called before this. After all the interceptors are done running, + // either ContinueFillOpsAfterInterception or + // ContinueFinalizeOpsAfterInterception will be called. Note that neither of + // them is invoked if there were no interceptors registered. + bool RunInterceptors() { + GPR_CODEGEN_ASSERT(ops_); + auto* client_rpc_info = call_->client_rpc_info(); + if (client_rpc_info != nullptr) { + if (client_rpc_info->interceptors_.size() == 0) { + return true; + } else { + RunClientInterceptors(); + return false; + } + } + + auto* server_rpc_info = call_->server_rpc_info(); + if (server_rpc_info == nullptr || + server_rpc_info->interceptors_.size() == 0) { + return true; + } + RunServerInterceptors(); + return false; + } + + // Returns true if no interceptors are run. Returns false otherwise if there + // are interceptors registered. After the interceptors are done running \a f + // will be invoked. This is to be used only by BaseAsyncRequest and + // SyncRequest. + bool RunInterceptors(std::function f) { + // This is used only by the server for initial call request + GPR_CODEGEN_ASSERT(reverse_ == true); + GPR_CODEGEN_ASSERT(call_->client_rpc_info() == nullptr); + auto* server_rpc_info = call_->server_rpc_info(); + if (server_rpc_info == nullptr || + server_rpc_info->interceptors_.size() == 0) { + return true; + } + callback_ = std::move(f); + RunServerInterceptors(); + return false; + } + + private: + void RunClientInterceptors() { + auto* rpc_info = call_->client_rpc_info(); + if (!reverse_) { + curr_iteration_ = 0; + } else { + if (rpc_info->hijacked_) { + curr_iteration_ = rpc_info->hijacked_interceptor_; + } else { + curr_iteration_ = rpc_info->interceptors_.size() - 1; + } + } + rpc_info->RunInterceptor(this, curr_iteration_); + } + + void RunServerInterceptors() { + auto* rpc_info = call_->server_rpc_info(); + if (!reverse_) { + curr_iteration_ = 0; + } else { + curr_iteration_ = rpc_info->interceptors_.size() - 1; + } + rpc_info->RunInterceptor(this, curr_iteration_); + } + + void ProceedClient() { + auto* rpc_info = call_->client_rpc_info(); + if (rpc_info->hijacked_ && !reverse_ && + curr_iteration_ == rpc_info->hijacked_interceptor_ && + !ran_hijacking_interceptor_) { + // We now need to provide hijacked recv ops to this interceptor + ClearHookPoints(); + ops_->SetHijackingState(); + ran_hijacking_interceptor_ = true; + rpc_info->RunInterceptor(this, curr_iteration_); + return; + } + if (!reverse_) { + curr_iteration_++; + // We are going down the stack of interceptors + if (curr_iteration_ < static_cast(rpc_info->interceptors_.size())) { + if (rpc_info->hijacked_ && + curr_iteration_ > rpc_info->hijacked_interceptor_) { + // This is a hijacked RPC and we are done with hijacking + ops_->ContinueFillOpsAfterInterception(); + } else { + rpc_info->RunInterceptor(this, curr_iteration_); + } + } else { + // we are done running all the interceptors without any hijacking + ops_->ContinueFillOpsAfterInterception(); + } + } else { + curr_iteration_--; + // We are going up the stack of interceptors + if (curr_iteration_ >= 0) { + // Continue running interceptors + rpc_info->RunInterceptor(this, curr_iteration_); + } else { + // we are done running all the interceptors without any hijacking + ops_->ContinueFinalizeResultAfterInterception(); + } + } + } + + void ProceedServer() { + auto* rpc_info = call_->server_rpc_info(); + if (!reverse_) { + curr_iteration_++; + if (curr_iteration_ < static_cast(rpc_info->interceptors_.size())) { + return rpc_info->RunInterceptor(this, curr_iteration_); + } else if (ops_) { + return ops_->ContinueFillOpsAfterInterception(); + } + } else { + curr_iteration_--; + // We are going up the stack of interceptors + if (curr_iteration_ >= 0) { + // Continue running interceptors + return rpc_info->RunInterceptor(this, curr_iteration_); + } else if (ops_) { + return ops_->ContinueFinalizeResultAfterInterception(); + } + } + GPR_CODEGEN_ASSERT(callback_); + callback_(); + } + + void ClearHookPoints() { + for (auto i = 0; + i < static_cast( + experimental::InterceptionHookPoints::NUM_INTERCEPTION_HOOKS); + i++) { + hooks_[i] = false; + } + } + + std::array( + experimental::InterceptionHookPoints::NUM_INTERCEPTION_HOOKS)> + hooks_; + + int curr_iteration_ = 0; // Current iterator + bool reverse_ = false; + bool ran_hijacking_interceptor_ = false; + Call* call_ = + nullptr; // The Call object is present along with CallOpSet object + CallOpSetInterface* ops_ = nullptr; + std::function callback_; + + ByteBuffer* send_message_ = nullptr; + + std::multimap* send_initial_metadata_; + + grpc_status_code* code_ = nullptr; + grpc::string* error_details_ = nullptr; + grpc::string* error_message_ = nullptr; + Status send_status_; + + std::multimap* send_trailing_metadata_ = nullptr; + + void* recv_message_ = nullptr; + + internal::MetadataMap* recv_initial_metadata_ = nullptr; + + Status* recv_status_ = nullptr; + + internal::MetadataMap* recv_trailing_metadata_ = nullptr; +}; + +/// Primary implementation of CallOpSetInterface. +/// Since we cannot use variadic templates, we declare slots up to +/// the maximum count of ops we'll need in a set. We leverage the +/// empty base class optimization to slim this class (especially +/// when there are many unused slots used). To avoid duplicate base classes, +/// the template parmeter for CallNoOp is varied by argument position. +template +class CallOpSet : public CallOpSetInterface, + public Op1, + public Op2, + public Op3, + public Op4, + public Op5, + public Op6 { + public: + CallOpSet() : cq_tag_(this), return_tag_(this) {} + // The copy constructor and assignment operator reset the value of + // cq_tag_ and return_tag_ since those are only meaningful on a specific + // object, not across objects. + CallOpSet(const CallOpSet& other) + : cq_tag_(this), + return_tag_(this), + call_(other.call_), + done_intercepting_(false), + interceptor_methods_(InterceptorBatchMethodsImpl()) {} + + CallOpSet& operator=(const CallOpSet& other) { + cq_tag_ = this; + return_tag_ = this; + call_ = other.call_; + done_intercepting_ = false; + interceptor_methods_ = InterceptorBatchMethodsImpl(); + return *this; + } + + void FillOps(Call* call) override { + done_intercepting_ = false; + g_core_codegen_interface->grpc_call_ref(call->call()); + call_ = + *call; // It's fine to create a copy of call since it's just pointers + + if (RunInterceptors()) { + ContinueFillOpsAfterInterception(); + } else { + // After the interceptors are run, ContinueFillOpsAfterInterception will + // be run + } + } + + bool FinalizeResult(void** tag, bool* status) override { + if (done_intercepting_) { + // We have already finished intercepting and filling in the results. This + // round trip from the core needed to be made because interceptors were + // run + *tag = return_tag_; + *status = saved_status_; + g_core_codegen_interface->grpc_call_unref(call_.call()); + return true; + } + + this->Op1::FinishOp(status); + this->Op2::FinishOp(status); + this->Op3::FinishOp(status); + this->Op4::FinishOp(status); + this->Op5::FinishOp(status); + this->Op6::FinishOp(status); + saved_status_ = *status; + if (RunInterceptorsPostRecv()) { + *tag = return_tag_; + g_core_codegen_interface->grpc_call_unref(call_.call()); + return true; + } + // Interceptors are going to be run, so we can't return the tag just yet. + // After the interceptors are run, ContinueFinalizeResultAfterInterception + return false; + } + + void set_output_tag(void* return_tag) { return_tag_ = return_tag; } + + void* cq_tag() override { return cq_tag_; } + + /// set_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; } + + // 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. + void SetHijackingState() override { + this->Op1::SetHijackingState(&interceptor_methods_); + this->Op2::SetHijackingState(&interceptor_methods_); + this->Op3::SetHijackingState(&interceptor_methods_); + this->Op4::SetHijackingState(&interceptor_methods_); + this->Op5::SetHijackingState(&interceptor_methods_); + this->Op6::SetHijackingState(&interceptor_methods_); + } + + // Should be called after interceptors are done running + void ContinueFillOpsAfterInterception() override { + static const size_t MAX_OPS = 6; + grpc_op ops[MAX_OPS]; + size_t nops = 0; + this->Op1::AddOp(ops, &nops); + this->Op2::AddOp(ops, &nops); + this->Op3::AddOp(ops, &nops); + this->Op4::AddOp(ops, &nops); + this->Op5::AddOp(ops, &nops); + 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)); + } + + // Should be called after interceptors are done running on the finalize result + // path + void ContinueFinalizeResultAfterInterception() override { + done_intercepting_ = true; + GPR_CODEGEN_ASSERT(GRPC_CALL_OK == + g_core_codegen_interface->grpc_call_start_batch( + call_.call(), nullptr, 0, cq_tag(), nullptr)); + } + + private: + // Returns true if no interceptors need to be run + bool RunInterceptors() { + interceptor_methods_.ClearState(); + interceptor_methods_.SetCallOpSetInterface(this); + interceptor_methods_.SetCall(&call_); + this->Op1::SetInterceptionHookPoint(&interceptor_methods_); + this->Op2::SetInterceptionHookPoint(&interceptor_methods_); + this->Op3::SetInterceptionHookPoint(&interceptor_methods_); + this->Op4::SetInterceptionHookPoint(&interceptor_methods_); + this->Op5::SetInterceptionHookPoint(&interceptor_methods_); + this->Op6::SetInterceptionHookPoint(&interceptor_methods_); + return interceptor_methods_.RunInterceptors(); + } + // Returns true if no interceptors need to be run + bool RunInterceptorsPostRecv() { + // Call and OpSet had already been set on the set state. + // SetReverse also clears previously set hook points + interceptor_methods_.SetReverse(); + this->Op1::SetFinishInterceptionHookPoint(&interceptor_methods_); + this->Op2::SetFinishInterceptionHookPoint(&interceptor_methods_); + this->Op3::SetFinishInterceptionHookPoint(&interceptor_methods_); + this->Op4::SetFinishInterceptionHookPoint(&interceptor_methods_); + this->Op5::SetFinishInterceptionHookPoint(&interceptor_methods_); + this->Op6::SetFinishInterceptionHookPoint(&interceptor_methods_); + return interceptor_methods_.RunInterceptors(); + } + + void* cq_tag_; + void* return_tag_; + Call call_; + bool done_intercepting_ = false; + InterceptorBatchMethodsImpl interceptor_methods_; + bool saved_status_; +}; + +} // namespace internal +} // namespace grpc + +#endif // GRPCPP_IMPL_CODEGEN_CALL_OP_SET_H diff --git a/include/grpcpp/impl/codegen/call_wrapper.h b/include/grpcpp/impl/codegen/call_wrapper.h deleted file mode 100644 index 06ab1fbc1ca..00000000000 --- a/include/grpcpp/impl/codegen/call_wrapper.h +++ /dev/null @@ -1,91 +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. - * - */ -#ifndef GRPCPP_IMPL_CODEGEN_CALL_WRAPPER_H -#define GRPCPP_IMPL_CODEGEN_CALL_WRAPPER_H - -#include - -namespace grpc { -class CompletionQueue; - -namespace experimental { -class ClientRpcInfo; -class ServerRpcInfo; -} // namespace experimental -namespace internal { -class CallHook; -class CallOpSetInterface; - -/// Straightforward wrapping of the C call object -class Call final { - public: - Call() - : call_hook_(nullptr), - cq_(nullptr), - call_(nullptr), - max_receive_message_size_(-1) {} - /** call is owned by the caller */ - Call(grpc_call* call, CallHook* call_hook, CompletionQueue* cq) - : call_hook_(call_hook), - cq_(cq), - call_(call), - max_receive_message_size_(-1) {} - - Call(grpc_call* call, CallHook* call_hook, CompletionQueue* cq, - experimental::ClientRpcInfo* rpc_info) - : call_hook_(call_hook), - cq_(cq), - call_(call), - max_receive_message_size_(-1), - client_rpc_info_(rpc_info) {} - - Call(grpc_call* call, CallHook* call_hook, CompletionQueue* cq, - int max_receive_message_size, experimental::ServerRpcInfo* rpc_info) - : call_hook_(call_hook), - cq_(cq), - call_(call), - max_receive_message_size_(max_receive_message_size), - server_rpc_info_(rpc_info) {} - - void PerformOps(CallOpSetInterface* ops); - - grpc_call* call() const { return call_; } - CompletionQueue* cq() const { return cq_; } - - int max_receive_message_size() const { return max_receive_message_size_; } - - experimental::ClientRpcInfo* client_rpc_info() const { - return client_rpc_info_; - } - - experimental::ServerRpcInfo* server_rpc_info() const { - return server_rpc_info_; - } - - private: - CallHook* call_hook_; - CompletionQueue* cq_; - grpc_call* call_; - int max_receive_message_size_; - experimental::ClientRpcInfo* client_rpc_info_ = nullptr; - experimental::ServerRpcInfo* server_rpc_info_ = nullptr; -}; -} // namespace internal -} // namespace grpc - -#endif // GRPCPP_IMPL_CODEGEN_CALL_WRAPPER_H diff --git a/include/grpcpp/impl/codegen/channel_interface.h b/include/grpcpp/impl/codegen/channel_interface.h index 03fa502df67..57026a235e6 100644 --- a/include/grpcpp/impl/codegen/channel_interface.h +++ b/include/grpcpp/impl/codegen/channel_interface.h @@ -20,7 +20,7 @@ #define GRPCPP_IMPL_CODEGEN_CHANNEL_INTERFACE_H #include -#include +#include #include #include diff --git a/include/grpcpp/impl/codegen/metadata_map.h b/include/grpcpp/impl/codegen/metadata_map.h index 5e062a50f83..0bba3ed4e36 100644 --- a/include/grpcpp/impl/codegen/metadata_map.h +++ b/include/grpcpp/impl/codegen/metadata_map.h @@ -19,6 +19,8 @@ #ifndef GRPCPP_IMPL_CODEGEN_METADATA_MAP_H #define GRPCPP_IMPL_CODEGEN_METADATA_MAP_H +#include + #include #include diff --git a/include/grpcpp/impl/codegen/server_context.h b/include/grpcpp/impl/codegen/server_context.h index 5a3249b9044..7559fb3b346 100644 --- a/include/grpcpp/impl/codegen/server_context.h +++ b/include/grpcpp/impl/codegen/server_context.h @@ -26,11 +26,13 @@ #include #include +#include #include #include #include #include #include +#include #include #include diff --git a/src/cpp/client/channel_cc.cc b/src/cpp/client/channel_cc.cc index f2a2a2fdc92..15e3ccb3c9f 100644 --- a/src/cpp/client/channel_cc.cc +++ b/src/cpp/client/channel_cc.cc @@ -33,6 +33,7 @@ #include #include #include +#include #include #include #include diff --git a/src/cpp/codegen/call_wrapper.cc b/src/cpp/codegen/call_wrapper.cc deleted file mode 100644 index 668fa886ef7..00000000000 --- a/src/cpp/codegen/call_wrapper.cc +++ /dev/null @@ -1,31 +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 - -#include - -namespace grpc { -namespace internal { - -void Call::PerformOps(CallOpSetInterface* ops) { - call_hook_->PerformOpsOnCall(ops, this); -} - -} // namespace internal -} // namespace grpc diff --git a/tools/doxygen/Doxyfile.c++ b/tools/doxygen/Doxyfile.c++ index 5b73fd897a3..a59021756a5 100644 --- a/tools/doxygen/Doxyfile.c++ +++ b/tools/doxygen/Doxyfile.c++ @@ -943,7 +943,7 @@ include/grpcpp/impl/codegen/async_unary_call.h \ include/grpcpp/impl/codegen/byte_buffer.h \ include/grpcpp/impl/codegen/call.h \ include/grpcpp/impl/codegen/call_hook.h \ -include/grpcpp/impl/codegen/call_wrapper.h \ +include/grpcpp/impl/codegen/call_op_set.h \ include/grpcpp/impl/codegen/callback_common.h \ include/grpcpp/impl/codegen/channel_interface.h \ include/grpcpp/impl/codegen/client_callback.h \ diff --git a/tools/doxygen/Doxyfile.c++.internal b/tools/doxygen/Doxyfile.c++.internal index 69135d24fcb..b162f557557 100644 --- a/tools/doxygen/Doxyfile.c++.internal +++ b/tools/doxygen/Doxyfile.c++.internal @@ -944,7 +944,7 @@ include/grpcpp/impl/codegen/async_unary_call.h \ include/grpcpp/impl/codegen/byte_buffer.h \ include/grpcpp/impl/codegen/call.h \ include/grpcpp/impl/codegen/call_hook.h \ -include/grpcpp/impl/codegen/call_wrapper.h \ +include/grpcpp/impl/codegen/call_op_set.h \ include/grpcpp/impl/codegen/callback_common.h \ include/grpcpp/impl/codegen/channel_interface.h \ include/grpcpp/impl/codegen/client_callback.h \ @@ -1191,7 +1191,6 @@ src/cpp/client/generic_stub.cc \ src/cpp/client/insecure_credentials.cc \ src/cpp/client/secure_credentials.cc \ src/cpp/client/secure_credentials.h \ -src/cpp/codegen/call_wrapper.cc \ src/cpp/codegen/codegen_init.cc \ src/cpp/common/alarm.cc \ src/cpp/common/auth_property_iterator.cc \ diff --git a/tools/run_tests/generated/sources_and_headers.json b/tools/run_tests/generated/sources_and_headers.json index ab5a9db8ef2..85c76bad3a4 100644 --- a/tools/run_tests/generated/sources_and_headers.json +++ b/tools/run_tests/generated/sources_and_headers.json @@ -11176,7 +11176,7 @@ "include/grpcpp/impl/codegen/byte_buffer.h", "include/grpcpp/impl/codegen/call.h", "include/grpcpp/impl/codegen/call_hook.h", - "include/grpcpp/impl/codegen/call_wrapper.h", + "include/grpcpp/impl/codegen/call_op_set.h", "include/grpcpp/impl/codegen/callback_common.h", "include/grpcpp/impl/codegen/channel_interface.h", "include/grpcpp/impl/codegen/client_callback.h", @@ -11249,7 +11249,7 @@ "include/grpcpp/impl/codegen/byte_buffer.h", "include/grpcpp/impl/codegen/call.h", "include/grpcpp/impl/codegen/call_hook.h", - "include/grpcpp/impl/codegen/call_wrapper.h", + "include/grpcpp/impl/codegen/call_op_set.h", "include/grpcpp/impl/codegen/callback_common.h", "include/grpcpp/impl/codegen/channel_interface.h", "include/grpcpp/impl/codegen/client_callback.h", @@ -11294,7 +11294,6 @@ "language": "c++", "name": "grpc++_codegen_base_src", "src": [ - "src/cpp/codegen/call_wrapper.cc", "src/cpp/codegen/codegen_init.cc" ], "third_party": false,