Merge github.com:google/grpc into c++api
diff --git a/Makefile b/Makefile
index 74e855f..24a1141 100644
--- a/Makefile
+++ b/Makefile
@@ -391,7 +391,6 @@
qps_server: bins/$(CONFIG)/qps_server
ruby_plugin: bins/$(CONFIG)/ruby_plugin
status_test: bins/$(CONFIG)/status_test
-sync_client_async_server_test: bins/$(CONFIG)/sync_client_async_server_test
thread_pool_test: bins/$(CONFIG)/thread_pool_test
tips_client: bins/$(CONFIG)/tips_client
tips_publisher_test: bins/$(CONFIG)/tips_publisher_test
@@ -725,7 +724,7 @@
buildtests_c: privatelibs_c bins/$(CONFIG)/alarm_heap_test bins/$(CONFIG)/alarm_list_test bins/$(CONFIG)/alarm_test bins/$(CONFIG)/alpn_test bins/$(CONFIG)/bin_encoder_test bins/$(CONFIG)/census_hash_table_test bins/$(CONFIG)/census_statistics_multiple_writers_circular_buffer_test bins/$(CONFIG)/census_statistics_multiple_writers_test bins/$(CONFIG)/census_statistics_performance_test bins/$(CONFIG)/census_statistics_quick_test bins/$(CONFIG)/census_statistics_small_log_test bins/$(CONFIG)/census_stub_test bins/$(CONFIG)/census_window_stats_test bins/$(CONFIG)/chttp2_status_conversion_test bins/$(CONFIG)/chttp2_stream_encoder_test bins/$(CONFIG)/chttp2_stream_map_test bins/$(CONFIG)/chttp2_transport_end2end_test bins/$(CONFIG)/dualstack_socket_test bins/$(CONFIG)/echo_client bins/$(CONFIG)/echo_server bins/$(CONFIG)/echo_test bins/$(CONFIG)/fd_posix_test bins/$(CONFIG)/fling_client bins/$(CONFIG)/fling_server bins/$(CONFIG)/fling_stream_test bins/$(CONFIG)/fling_test bins/$(CONFIG)/gpr_cancellable_test bins/$(CONFIG)/gpr_cmdline_test bins/$(CONFIG)/gpr_env_test bins/$(CONFIG)/gpr_file_test bins/$(CONFIG)/gpr_histogram_test bins/$(CONFIG)/gpr_host_port_test bins/$(CONFIG)/gpr_log_test bins/$(CONFIG)/gpr_slice_buffer_test bins/$(CONFIG)/gpr_slice_test bins/$(CONFIG)/gpr_string_test bins/$(CONFIG)/gpr_sync_test bins/$(CONFIG)/gpr_thd_test bins/$(CONFIG)/gpr_time_test bins/$(CONFIG)/gpr_useful_test bins/$(CONFIG)/grpc_base64_test bins/$(CONFIG)/grpc_byte_buffer_reader_test bins/$(CONFIG)/grpc_channel_stack_test bins/$(CONFIG)/grpc_completion_queue_test bins/$(CONFIG)/grpc_credentials_test bins/$(CONFIG)/grpc_json_token_test bins/$(CONFIG)/grpc_stream_op_test bins/$(CONFIG)/hpack_parser_test bins/$(CONFIG)/hpack_table_test bins/$(CONFIG)/httpcli_format_request_test bins/$(CONFIG)/httpcli_parser_test bins/$(CONFIG)/httpcli_test bins/$(CONFIG)/json_rewrite bins/$(CONFIG)/json_rewrite_test bins/$(CONFIG)/json_test bins/$(CONFIG)/lame_client_test bins/$(CONFIG)/message_compress_test bins/$(CONFIG)/metadata_buffer_test bins/$(CONFIG)/murmur_hash_test bins/$(CONFIG)/no_server_test bins/$(CONFIG)/poll_kick_posix_test bins/$(CONFIG)/resolve_address_test bins/$(CONFIG)/secure_endpoint_test bins/$(CONFIG)/sockaddr_utils_test bins/$(CONFIG)/tcp_client_posix_test bins/$(CONFIG)/tcp_posix_test bins/$(CONFIG)/tcp_server_posix_test bins/$(CONFIG)/time_averaged_stats_test bins/$(CONFIG)/time_test bins/$(CONFIG)/timeout_encoding_test bins/$(CONFIG)/transport_metadata_test bins/$(CONFIG)/chttp2_fake_security_cancel_after_accept_test bins/$(CONFIG)/chttp2_fake_security_cancel_after_accept_and_writes_closed_test bins/$(CONFIG)/chttp2_fake_security_cancel_after_invoke_test bins/$(CONFIG)/chttp2_fake_security_cancel_before_invoke_test bins/$(CONFIG)/chttp2_fake_security_cancel_in_a_vacuum_test bins/$(CONFIG)/chttp2_fake_security_census_simple_request_test bins/$(CONFIG)/chttp2_fake_security_disappearing_server_test bins/$(CONFIG)/chttp2_fake_security_early_server_shutdown_finishes_inflight_calls_test bins/$(CONFIG)/chttp2_fake_security_early_server_shutdown_finishes_tags_test bins/$(CONFIG)/chttp2_fake_security_graceful_server_shutdown_test bins/$(CONFIG)/chttp2_fake_security_invoke_large_request_test bins/$(CONFIG)/chttp2_fake_security_max_concurrent_streams_test bins/$(CONFIG)/chttp2_fake_security_no_op_test bins/$(CONFIG)/chttp2_fake_security_ping_pong_streaming_test bins/$(CONFIG)/chttp2_fake_security_request_response_with_binary_metadata_and_payload_test bins/$(CONFIG)/chttp2_fake_security_request_response_with_metadata_and_payload_test bins/$(CONFIG)/chttp2_fake_security_request_response_with_payload_test bins/$(CONFIG)/chttp2_fake_security_request_with_large_metadata_test bins/$(CONFIG)/chttp2_fake_security_request_with_payload_test bins/$(CONFIG)/chttp2_fake_security_simple_delayed_request_test bins/$(CONFIG)/chttp2_fake_security_simple_request_test bins/$(CONFIG)/chttp2_fake_security_thread_stress_test bins/$(CONFIG)/chttp2_fake_security_writes_done_hangs_with_pending_read_test bins/$(CONFIG)/chttp2_fake_security_cancel_after_accept_legacy_test bins/$(CONFIG)/chttp2_fake_security_cancel_after_accept_and_writes_closed_legacy_test bins/$(CONFIG)/chttp2_fake_security_cancel_after_invoke_legacy_test bins/$(CONFIG)/chttp2_fake_security_cancel_before_invoke_legacy_test bins/$(CONFIG)/chttp2_fake_security_cancel_in_a_vacuum_legacy_test bins/$(CONFIG)/chttp2_fake_security_census_simple_request_legacy_test bins/$(CONFIG)/chttp2_fake_security_disappearing_server_legacy_test bins/$(CONFIG)/chttp2_fake_security_early_server_shutdown_finishes_inflight_calls_legacy_test bins/$(CONFIG)/chttp2_fake_security_early_server_shutdown_finishes_tags_legacy_test bins/$(CONFIG)/chttp2_fake_security_graceful_server_shutdown_legacy_test bins/$(CONFIG)/chttp2_fake_security_invoke_large_request_legacy_test bins/$(CONFIG)/chttp2_fake_security_max_concurrent_streams_legacy_test bins/$(CONFIG)/chttp2_fake_security_no_op_legacy_test bins/$(CONFIG)/chttp2_fake_security_ping_pong_streaming_legacy_test bins/$(CONFIG)/chttp2_fake_security_request_response_with_binary_metadata_and_payload_legacy_test bins/$(CONFIG)/chttp2_fake_security_request_response_with_metadata_and_payload_legacy_test bins/$(CONFIG)/chttp2_fake_security_request_response_with_payload_legacy_test bins/$(CONFIG)/chttp2_fake_security_request_response_with_trailing_metadata_and_payload_legacy_test bins/$(CONFIG)/chttp2_fake_security_request_with_large_metadata_legacy_test bins/$(CONFIG)/chttp2_fake_security_request_with_payload_legacy_test bins/$(CONFIG)/chttp2_fake_security_simple_delayed_request_legacy_test bins/$(CONFIG)/chttp2_fake_security_simple_request_legacy_test bins/$(CONFIG)/chttp2_fake_security_thread_stress_legacy_test bins/$(CONFIG)/chttp2_fake_security_writes_done_hangs_with_pending_read_legacy_test bins/$(CONFIG)/chttp2_fullstack_cancel_after_accept_test bins/$(CONFIG)/chttp2_fullstack_cancel_after_accept_and_writes_closed_test bins/$(CONFIG)/chttp2_fullstack_cancel_after_invoke_test bins/$(CONFIG)/chttp2_fullstack_cancel_before_invoke_test bins/$(CONFIG)/chttp2_fullstack_cancel_in_a_vacuum_test bins/$(CONFIG)/chttp2_fullstack_census_simple_request_test bins/$(CONFIG)/chttp2_fullstack_disappearing_server_test bins/$(CONFIG)/chttp2_fullstack_early_server_shutdown_finishes_inflight_calls_test bins/$(CONFIG)/chttp2_fullstack_early_server_shutdown_finishes_tags_test bins/$(CONFIG)/chttp2_fullstack_graceful_server_shutdown_test bins/$(CONFIG)/chttp2_fullstack_invoke_large_request_test bins/$(CONFIG)/chttp2_fullstack_max_concurrent_streams_test bins/$(CONFIG)/chttp2_fullstack_no_op_test bins/$(CONFIG)/chttp2_fullstack_ping_pong_streaming_test bins/$(CONFIG)/chttp2_fullstack_request_response_with_binary_metadata_and_payload_test bins/$(CONFIG)/chttp2_fullstack_request_response_with_metadata_and_payload_test bins/$(CONFIG)/chttp2_fullstack_request_response_with_payload_test bins/$(CONFIG)/chttp2_fullstack_request_with_large_metadata_test bins/$(CONFIG)/chttp2_fullstack_request_with_payload_test bins/$(CONFIG)/chttp2_fullstack_simple_delayed_request_test bins/$(CONFIG)/chttp2_fullstack_simple_request_test bins/$(CONFIG)/chttp2_fullstack_thread_stress_test bins/$(CONFIG)/chttp2_fullstack_writes_done_hangs_with_pending_read_test bins/$(CONFIG)/chttp2_fullstack_cancel_after_accept_legacy_test bins/$(CONFIG)/chttp2_fullstack_cancel_after_accept_and_writes_closed_legacy_test bins/$(CONFIG)/chttp2_fullstack_cancel_after_invoke_legacy_test bins/$(CONFIG)/chttp2_fullstack_cancel_before_invoke_legacy_test bins/$(CONFIG)/chttp2_fullstack_cancel_in_a_vacuum_legacy_test bins/$(CONFIG)/chttp2_fullstack_census_simple_request_legacy_test bins/$(CONFIG)/chttp2_fullstack_disappearing_server_legacy_test bins/$(CONFIG)/chttp2_fullstack_early_server_shutdown_finishes_inflight_calls_legacy_test bins/$(CONFIG)/chttp2_fullstack_early_server_shutdown_finishes_tags_legacy_test bins/$(CONFIG)/chttp2_fullstack_graceful_server_shutdown_legacy_test bins/$(CONFIG)/chttp2_fullstack_invoke_large_request_legacy_test bins/$(CONFIG)/chttp2_fullstack_max_concurrent_streams_legacy_test bins/$(CONFIG)/chttp2_fullstack_no_op_legacy_test bins/$(CONFIG)/chttp2_fullstack_ping_pong_streaming_legacy_test bins/$(CONFIG)/chttp2_fullstack_request_response_with_binary_metadata_and_payload_legacy_test bins/$(CONFIG)/chttp2_fullstack_request_response_with_metadata_and_payload_legacy_test bins/$(CONFIG)/chttp2_fullstack_request_response_with_payload_legacy_test bins/$(CONFIG)/chttp2_fullstack_request_response_with_trailing_metadata_and_payload_legacy_test bins/$(CONFIG)/chttp2_fullstack_request_with_large_metadata_legacy_test bins/$(CONFIG)/chttp2_fullstack_request_with_payload_legacy_test bins/$(CONFIG)/chttp2_fullstack_simple_delayed_request_legacy_test bins/$(CONFIG)/chttp2_fullstack_simple_request_legacy_test bins/$(CONFIG)/chttp2_fullstack_thread_stress_legacy_test bins/$(CONFIG)/chttp2_fullstack_writes_done_hangs_with_pending_read_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_cancel_after_accept_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_cancel_after_accept_and_writes_closed_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_cancel_after_invoke_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_cancel_before_invoke_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_cancel_in_a_vacuum_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_census_simple_request_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_disappearing_server_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_early_server_shutdown_finishes_inflight_calls_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_early_server_shutdown_finishes_tags_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_graceful_server_shutdown_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_invoke_large_request_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_max_concurrent_streams_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_no_op_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_ping_pong_streaming_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_request_response_with_binary_metadata_and_payload_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_request_response_with_metadata_and_payload_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_request_response_with_payload_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_request_with_large_metadata_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_request_with_payload_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_simple_delayed_request_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_simple_request_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_thread_stress_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_writes_done_hangs_with_pending_read_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_cancel_after_accept_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_cancel_after_accept_and_writes_closed_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_cancel_after_invoke_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_cancel_before_invoke_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_cancel_in_a_vacuum_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_census_simple_request_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_disappearing_server_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_early_server_shutdown_finishes_inflight_calls_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_early_server_shutdown_finishes_tags_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_graceful_server_shutdown_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_invoke_large_request_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_max_concurrent_streams_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_no_op_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_ping_pong_streaming_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_request_response_with_binary_metadata_and_payload_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_request_response_with_metadata_and_payload_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_request_response_with_payload_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_request_response_with_trailing_metadata_and_payload_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_request_with_large_metadata_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_request_with_payload_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_simple_delayed_request_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_simple_request_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_thread_stress_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_fullstack_writes_done_hangs_with_pending_read_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_cancel_after_accept_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_cancel_after_accept_and_writes_closed_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_cancel_after_invoke_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_cancel_before_invoke_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_cancel_in_a_vacuum_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_census_simple_request_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_disappearing_server_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_early_server_shutdown_finishes_inflight_calls_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_early_server_shutdown_finishes_tags_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_graceful_server_shutdown_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_invoke_large_request_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_max_concurrent_streams_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_no_op_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_ping_pong_streaming_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_request_response_with_binary_metadata_and_payload_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_request_response_with_metadata_and_payload_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_request_response_with_payload_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_request_with_large_metadata_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_request_with_payload_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_simple_delayed_request_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_simple_request_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_thread_stress_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_writes_done_hangs_with_pending_read_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_cancel_after_accept_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_cancel_after_accept_and_writes_closed_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_cancel_after_invoke_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_cancel_before_invoke_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_cancel_in_a_vacuum_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_census_simple_request_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_disappearing_server_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_early_server_shutdown_finishes_inflight_calls_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_early_server_shutdown_finishes_tags_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_graceful_server_shutdown_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_invoke_large_request_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_max_concurrent_streams_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_no_op_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_ping_pong_streaming_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_request_response_with_binary_metadata_and_payload_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_request_response_with_metadata_and_payload_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_request_response_with_payload_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_request_response_with_trailing_metadata_and_payload_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_request_with_large_metadata_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_request_with_payload_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_simple_delayed_request_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_simple_request_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_thread_stress_legacy_test bins/$(CONFIG)/chttp2_simple_ssl_with_oauth2_fullstack_writes_done_hangs_with_pending_read_legacy_test bins/$(CONFIG)/chttp2_socket_pair_cancel_after_accept_test bins/$(CONFIG)/chttp2_socket_pair_cancel_after_accept_and_writes_closed_test bins/$(CONFIG)/chttp2_socket_pair_cancel_after_invoke_test bins/$(CONFIG)/chttp2_socket_pair_cancel_before_invoke_test bins/$(CONFIG)/chttp2_socket_pair_cancel_in_a_vacuum_test bins/$(CONFIG)/chttp2_socket_pair_census_simple_request_test bins/$(CONFIG)/chttp2_socket_pair_disappearing_server_test bins/$(CONFIG)/chttp2_socket_pair_early_server_shutdown_finishes_inflight_calls_test bins/$(CONFIG)/chttp2_socket_pair_early_server_shutdown_finishes_tags_test bins/$(CONFIG)/chttp2_socket_pair_graceful_server_shutdown_test bins/$(CONFIG)/chttp2_socket_pair_invoke_large_request_test bins/$(CONFIG)/chttp2_socket_pair_max_concurrent_streams_test bins/$(CONFIG)/chttp2_socket_pair_no_op_test bins/$(CONFIG)/chttp2_socket_pair_ping_pong_streaming_test bins/$(CONFIG)/chttp2_socket_pair_request_response_with_binary_metadata_and_payload_test bins/$(CONFIG)/chttp2_socket_pair_request_response_with_metadata_and_payload_test bins/$(CONFIG)/chttp2_socket_pair_request_response_with_payload_test bins/$(CONFIG)/chttp2_socket_pair_request_with_large_metadata_test bins/$(CONFIG)/chttp2_socket_pair_request_with_payload_test bins/$(CONFIG)/chttp2_socket_pair_simple_delayed_request_test bins/$(CONFIG)/chttp2_socket_pair_simple_request_test bins/$(CONFIG)/chttp2_socket_pair_thread_stress_test bins/$(CONFIG)/chttp2_socket_pair_writes_done_hangs_with_pending_read_test bins/$(CONFIG)/chttp2_socket_pair_cancel_after_accept_legacy_test bins/$(CONFIG)/chttp2_socket_pair_cancel_after_accept_and_writes_closed_legacy_test bins/$(CONFIG)/chttp2_socket_pair_cancel_after_invoke_legacy_test bins/$(CONFIG)/chttp2_socket_pair_cancel_before_invoke_legacy_test bins/$(CONFIG)/chttp2_socket_pair_cancel_in_a_vacuum_legacy_test bins/$(CONFIG)/chttp2_socket_pair_census_simple_request_legacy_test bins/$(CONFIG)/chttp2_socket_pair_disappearing_server_legacy_test bins/$(CONFIG)/chttp2_socket_pair_early_server_shutdown_finishes_inflight_calls_legacy_test bins/$(CONFIG)/chttp2_socket_pair_early_server_shutdown_finishes_tags_legacy_test bins/$(CONFIG)/chttp2_socket_pair_graceful_server_shutdown_legacy_test bins/$(CONFIG)/chttp2_socket_pair_invoke_large_request_legacy_test bins/$(CONFIG)/chttp2_socket_pair_max_concurrent_streams_legacy_test bins/$(CONFIG)/chttp2_socket_pair_no_op_legacy_test bins/$(CONFIG)/chttp2_socket_pair_ping_pong_streaming_legacy_test bins/$(CONFIG)/chttp2_socket_pair_request_response_with_binary_metadata_and_payload_legacy_test bins/$(CONFIG)/chttp2_socket_pair_request_response_with_metadata_and_payload_legacy_test bins/$(CONFIG)/chttp2_socket_pair_request_response_with_payload_legacy_test bins/$(CONFIG)/chttp2_socket_pair_request_response_with_trailing_metadata_and_payload_legacy_test bins/$(CONFIG)/chttp2_socket_pair_request_with_large_metadata_legacy_test bins/$(CONFIG)/chttp2_socket_pair_request_with_payload_legacy_test bins/$(CONFIG)/chttp2_socket_pair_simple_delayed_request_legacy_test bins/$(CONFIG)/chttp2_socket_pair_simple_request_legacy_test bins/$(CONFIG)/chttp2_socket_pair_thread_stress_legacy_test bins/$(CONFIG)/chttp2_socket_pair_writes_done_hangs_with_pending_read_legacy_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_cancel_after_accept_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_cancel_after_accept_and_writes_closed_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_cancel_after_invoke_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_cancel_before_invoke_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_cancel_in_a_vacuum_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_census_simple_request_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_disappearing_server_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_early_server_shutdown_finishes_inflight_calls_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_early_server_shutdown_finishes_tags_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_graceful_server_shutdown_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_invoke_large_request_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_max_concurrent_streams_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_no_op_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_ping_pong_streaming_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_request_response_with_binary_metadata_and_payload_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_request_response_with_metadata_and_payload_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_request_response_with_payload_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_request_with_large_metadata_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_request_with_payload_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_simple_delayed_request_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_simple_request_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_thread_stress_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_writes_done_hangs_with_pending_read_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_cancel_after_accept_legacy_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_cancel_after_accept_and_writes_closed_legacy_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_cancel_after_invoke_legacy_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_cancel_before_invoke_legacy_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_cancel_in_a_vacuum_legacy_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_census_simple_request_legacy_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_disappearing_server_legacy_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_early_server_shutdown_finishes_inflight_calls_legacy_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_early_server_shutdown_finishes_tags_legacy_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_graceful_server_shutdown_legacy_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_invoke_large_request_legacy_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_max_concurrent_streams_legacy_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_no_op_legacy_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_ping_pong_streaming_legacy_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_request_response_with_binary_metadata_and_payload_legacy_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_request_response_with_metadata_and_payload_legacy_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_request_response_with_payload_legacy_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_request_response_with_trailing_metadata_and_payload_legacy_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_request_with_large_metadata_legacy_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_request_with_payload_legacy_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_simple_delayed_request_legacy_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_simple_request_legacy_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_thread_stress_legacy_test bins/$(CONFIG)/chttp2_socket_pair_one_byte_at_a_time_writes_done_hangs_with_pending_read_legacy_test
-buildtests_cxx: privatelibs_cxx bins/$(CONFIG)/channel_arguments_test bins/$(CONFIG)/credentials_test bins/$(CONFIG)/end2end_test bins/$(CONFIG)/interop_client bins/$(CONFIG)/interop_server bins/$(CONFIG)/qps_client bins/$(CONFIG)/qps_server bins/$(CONFIG)/status_test bins/$(CONFIG)/sync_client_async_server_test bins/$(CONFIG)/thread_pool_test bins/$(CONFIG)/tips_client bins/$(CONFIG)/tips_publisher_test bins/$(CONFIG)/tips_subscriber_test
+buildtests_cxx: privatelibs_cxx bins/$(CONFIG)/channel_arguments_test bins/$(CONFIG)/credentials_test bins/$(CONFIG)/end2end_test bins/$(CONFIG)/interop_client bins/$(CONFIG)/interop_server bins/$(CONFIG)/qps_client bins/$(CONFIG)/qps_server bins/$(CONFIG)/status_test bins/$(CONFIG)/thread_pool_test bins/$(CONFIG)/tips_client bins/$(CONFIG)/tips_publisher_test bins/$(CONFIG)/tips_subscriber_test
test: test_c test_cxx
@@ -1435,8 +1434,6 @@
$(Q) ./bins/$(CONFIG)/end2end_test || ( echo test end2end_test failed ; exit 1 )
$(E) "[RUN] Testing status_test"
$(Q) ./bins/$(CONFIG)/status_test || ( echo test status_test failed ; exit 1 )
- $(E) "[RUN] Testing sync_client_async_server_test"
- $(Q) ./bins/$(CONFIG)/sync_client_async_server_test || ( echo test sync_client_async_server_test failed ; exit 1 )
$(E) "[RUN] Testing thread_pool_test"
$(Q) ./bins/$(CONFIG)/thread_pool_test || ( echo test thread_pool_test failed ; exit 1 )
$(E) "[RUN] Testing tips_publisher_test"
@@ -2645,27 +2642,23 @@
src/cpp/client/channel.cc \
src/cpp/client/channel_arguments.cc \
src/cpp/client/client_context.cc \
+ src/cpp/client/client_unary_call.cc \
src/cpp/client/create_channel.cc \
src/cpp/client/credentials.cc \
src/cpp/client/internal_stub.cc \
+ src/cpp/common/call.cc \
src/cpp/common/completion_queue.cc \
src/cpp/common/rpc_method.cc \
src/cpp/proto/proto_utils.cc \
- src/cpp/server/async_server.cc \
- src/cpp/server/async_server_context.cc \
src/cpp/server/server.cc \
src/cpp/server/server_builder.cc \
- src/cpp/server/server_context_impl.cc \
+ src/cpp/server/server_context.cc \
src/cpp/server/server_credentials.cc \
- src/cpp/server/server_rpc_handler.cc \
src/cpp/server/thread_pool.cc \
- src/cpp/stream/stream_context.cc \
src/cpp/util/status.cc \
src/cpp/util/time.cc \
PUBLIC_HEADERS_CXX += \
- include/grpc++/async_server.h \
- include/grpc++/async_server_context.h \
include/grpc++/channel_arguments.h \
include/grpc++/channel_interface.h \
include/grpc++/client_context.h \
@@ -2673,6 +2666,8 @@
include/grpc++/config.h \
include/grpc++/create_channel.h \
include/grpc++/credentials.h \
+ include/grpc++/impl/call.h \
+ include/grpc++/impl/client_unary_call.h \
include/grpc++/impl/internal_stub.h \
include/grpc++/impl/rpc_method.h \
include/grpc++/impl/rpc_service_method.h \
@@ -2704,21 +2699,19 @@
src/cpp/client/channel.cc: $(OPENSSL_DEP)
src/cpp/client/channel_arguments.cc: $(OPENSSL_DEP)
src/cpp/client/client_context.cc: $(OPENSSL_DEP)
+src/cpp/client/client_unary_call.cc: $(OPENSSL_DEP)
src/cpp/client/create_channel.cc: $(OPENSSL_DEP)
src/cpp/client/credentials.cc: $(OPENSSL_DEP)
src/cpp/client/internal_stub.cc: $(OPENSSL_DEP)
+src/cpp/common/call.cc: $(OPENSSL_DEP)
src/cpp/common/completion_queue.cc: $(OPENSSL_DEP)
src/cpp/common/rpc_method.cc: $(OPENSSL_DEP)
src/cpp/proto/proto_utils.cc: $(OPENSSL_DEP)
-src/cpp/server/async_server.cc: $(OPENSSL_DEP)
-src/cpp/server/async_server_context.cc: $(OPENSSL_DEP)
src/cpp/server/server.cc: $(OPENSSL_DEP)
src/cpp/server/server_builder.cc: $(OPENSSL_DEP)
-src/cpp/server/server_context_impl.cc: $(OPENSSL_DEP)
+src/cpp/server/server_context.cc: $(OPENSSL_DEP)
src/cpp/server/server_credentials.cc: $(OPENSSL_DEP)
-src/cpp/server/server_rpc_handler.cc: $(OPENSSL_DEP)
src/cpp/server/thread_pool.cc: $(OPENSSL_DEP)
-src/cpp/stream/stream_context.cc: $(OPENSSL_DEP)
src/cpp/util/status.cc: $(OPENSSL_DEP)
src/cpp/util/time.cc: $(OPENSSL_DEP)
endif
@@ -2764,21 +2757,19 @@
objs/$(CONFIG)/src/cpp/client/channel.o:
objs/$(CONFIG)/src/cpp/client/channel_arguments.o:
objs/$(CONFIG)/src/cpp/client/client_context.o:
+objs/$(CONFIG)/src/cpp/client/client_unary_call.o:
objs/$(CONFIG)/src/cpp/client/create_channel.o:
objs/$(CONFIG)/src/cpp/client/credentials.o:
objs/$(CONFIG)/src/cpp/client/internal_stub.o:
+objs/$(CONFIG)/src/cpp/common/call.o:
objs/$(CONFIG)/src/cpp/common/completion_queue.o:
objs/$(CONFIG)/src/cpp/common/rpc_method.o:
objs/$(CONFIG)/src/cpp/proto/proto_utils.o:
-objs/$(CONFIG)/src/cpp/server/async_server.o:
-objs/$(CONFIG)/src/cpp/server/async_server_context.o:
objs/$(CONFIG)/src/cpp/server/server.o:
objs/$(CONFIG)/src/cpp/server/server_builder.o:
-objs/$(CONFIG)/src/cpp/server/server_context_impl.o:
+objs/$(CONFIG)/src/cpp/server/server_context.o:
objs/$(CONFIG)/src/cpp/server/server_credentials.o:
-objs/$(CONFIG)/src/cpp/server/server_rpc_handler.o:
objs/$(CONFIG)/src/cpp/server/thread_pool.o:
-objs/$(CONFIG)/src/cpp/stream/stream_context.o:
objs/$(CONFIG)/src/cpp/util/status.o:
objs/$(CONFIG)/src/cpp/util/time.o:
@@ -2787,7 +2778,6 @@
gens/test/cpp/util/messages.pb.cc \
gens/test/cpp/util/echo.pb.cc \
gens/test/cpp/util/echo_duplicate.pb.cc \
- test/cpp/end2end/async_test_server.cc \
test/cpp/util/create_test_channel.cc \
@@ -2806,7 +2796,6 @@
test/cpp/util/messages.proto: $(OPENSSL_DEP)
test/cpp/util/echo.proto: $(OPENSSL_DEP)
test/cpp/util/echo_duplicate.proto: $(OPENSSL_DEP)
-test/cpp/end2end/async_test_server.cc: $(OPENSSL_DEP)
test/cpp/util/create_test_channel.cc: $(OPENSSL_DEP)
endif
@@ -2834,7 +2823,6 @@
-objs/$(CONFIG)/test/cpp/end2end/async_test_server.o: gens/test/cpp/util/messages.pb.cc gens/test/cpp/util/echo.pb.cc gens/test/cpp/util/echo_duplicate.pb.cc
objs/$(CONFIG)/test/cpp/util/create_test_channel.o: gens/test/cpp/util/messages.pb.cc gens/test/cpp/util/echo.pb.cc gens/test/cpp/util/echo_duplicate.pb.cc
@@ -7122,37 +7110,6 @@
endif
-SYNC_CLIENT_ASYNC_SERVER_TEST_SRC = \
- test/cpp/end2end/sync_client_async_server_test.cc \
-
-SYNC_CLIENT_ASYNC_SERVER_TEST_OBJS = $(addprefix objs/$(CONFIG)/, $(addsuffix .o, $(basename $(SYNC_CLIENT_ASYNC_SERVER_TEST_SRC))))
-
-ifeq ($(NO_SECURE),true)
-
-# You can't build secure targets if you don't have OpenSSL with ALPN.
-
-bins/$(CONFIG)/sync_client_async_server_test: openssl_dep_error
-
-else
-
-bins/$(CONFIG)/sync_client_async_server_test: $(SYNC_CLIENT_ASYNC_SERVER_TEST_OBJS) libs/$(CONFIG)/libgrpc++_test_util.a libs/$(CONFIG)/libgrpc_test_util.a libs/$(CONFIG)/libgrpc++.a libs/$(CONFIG)/libgrpc.a libs/$(CONFIG)/libgpr_test_util.a libs/$(CONFIG)/libgpr.a
- $(E) "[LD] Linking $@"
- $(Q) mkdir -p `dirname $@`
- $(Q) $(LDXX) $(LDFLAGS) $(SYNC_CLIENT_ASYNC_SERVER_TEST_OBJS) $(GTEST_LIB) libs/$(CONFIG)/libgrpc++_test_util.a libs/$(CONFIG)/libgrpc_test_util.a libs/$(CONFIG)/libgrpc++.a libs/$(CONFIG)/libgrpc.a libs/$(CONFIG)/libgpr_test_util.a libs/$(CONFIG)/libgpr.a $(LDLIBSXX) $(LDLIBS) $(LDLIBS_SECURE) -o bins/$(CONFIG)/sync_client_async_server_test
-
-endif
-
-objs/$(CONFIG)/test/cpp/end2end/sync_client_async_server_test.o: libs/$(CONFIG)/libgrpc++_test_util.a libs/$(CONFIG)/libgrpc_test_util.a libs/$(CONFIG)/libgrpc++.a libs/$(CONFIG)/libgrpc.a libs/$(CONFIG)/libgpr_test_util.a libs/$(CONFIG)/libgpr.a
-
-deps_sync_client_async_server_test: $(SYNC_CLIENT_ASYNC_SERVER_TEST_OBJS:.o=.dep)
-
-ifneq ($(NO_SECURE),true)
-ifneq ($(NO_DEPS),true)
--include $(SYNC_CLIENT_ASYNC_SERVER_TEST_OBJS:.o=.dep)
-endif
-endif
-
-
THREAD_POOL_TEST_SRC = \
test/cpp/server/thread_pool_test.cc \
diff --git a/build.json b/build.json
index 6ac197f..4895c79 100644
--- a/build.json
+++ b/build.json
@@ -391,8 +391,6 @@
"build": "all",
"language": "c++",
"public_headers": [
- "include/grpc++/async_server.h",
- "include/grpc++/async_server_context.h",
"include/grpc++/channel_arguments.h",
"include/grpc++/channel_interface.h",
"include/grpc++/client_context.h",
@@ -400,6 +398,8 @@
"include/grpc++/config.h",
"include/grpc++/create_channel.h",
"include/grpc++/credentials.h",
+ "include/grpc++/impl/call.h",
+ "include/grpc++/impl/client_unary_call.h",
"include/grpc++/impl/internal_stub.h",
"include/grpc++/impl/rpc_method.h",
"include/grpc++/impl/rpc_service_method.h",
@@ -414,30 +414,26 @@
"headers": [
"src/cpp/client/channel.h",
"src/cpp/proto/proto_utils.h",
- "src/cpp/server/server_rpc_handler.h",
"src/cpp/server/thread_pool.h",
- "src/cpp/stream/stream_context.h",
"src/cpp/util/time.h"
],
"src": [
"src/cpp/client/channel.cc",
"src/cpp/client/channel_arguments.cc",
"src/cpp/client/client_context.cc",
+ "src/cpp/client/client_unary_call.cc",
"src/cpp/client/create_channel.cc",
"src/cpp/client/credentials.cc",
"src/cpp/client/internal_stub.cc",
+ "src/cpp/common/call.cc",
"src/cpp/common/completion_queue.cc",
"src/cpp/common/rpc_method.cc",
"src/cpp/proto/proto_utils.cc",
- "src/cpp/server/async_server.cc",
- "src/cpp/server/async_server_context.cc",
"src/cpp/server/server.cc",
"src/cpp/server/server_builder.cc",
- "src/cpp/server/server_context_impl.cc",
+ "src/cpp/server/server_context.cc",
"src/cpp/server/server_credentials.cc",
- "src/cpp/server/server_rpc_handler.cc",
"src/cpp/server/thread_pool.cc",
- "src/cpp/stream/stream_context.cc",
"src/cpp/util/status.cc",
"src/cpp/util/time.cc"
],
@@ -455,7 +451,6 @@
"test/cpp/util/messages.proto",
"test/cpp/util/echo.proto",
"test/cpp/util/echo_duplicate.proto",
- "test/cpp/end2end/async_test_server.cc",
"test/cpp/util/create_test_channel.cc"
]
},
@@ -1629,7 +1624,6 @@
{
"name": "qps_client",
"build": "test",
- "run": false,
"language": "c++",
"src": [
"test/cpp/qps/qpstest.proto",
@@ -1642,12 +1636,12 @@
"grpc",
"gpr_test_util",
"gpr"
- ]
+ ],
+ "run": false
},
{
"name": "qps_server",
"build": "test",
- "run": false,
"language": "c++",
"src": [
"test/cpp/qps/qpstest.proto",
@@ -1660,7 +1654,8 @@
"grpc",
"gpr_test_util",
"gpr"
- ]
+ ],
+ "run": false
},
{
"name": "ruby_plugin",
@@ -1695,22 +1690,6 @@
]
},
{
- "name": "sync_client_async_server_test",
- "build": "test",
- "language": "c++",
- "src": [
- "test/cpp/end2end/sync_client_async_server_test.cc"
- ],
- "deps": [
- "grpc++_test_util",
- "grpc_test_util",
- "grpc++",
- "grpc",
- "gpr_test_util",
- "gpr"
- ]
- },
- {
"name": "thread_pool_test",
"build": "test",
"language": "c++",
diff --git a/examples/tips/publisher_test.cc b/examples/tips/publisher_test.cc
index 34737ae..db3e378 100644
--- a/examples/tips/publisher_test.cc
+++ b/examples/tips/publisher_test.cc
@@ -107,7 +107,7 @@
server_address_ << "localhost:" << port;
ServerBuilder builder;
builder.AddPort(server_address_.str());
- builder.RegisterService(service_.service());
+ builder.RegisterService(&service_);
server_ = builder.BuildAndStart();
channel_ = CreateChannel(server_address_.str(), ChannelArguments());
diff --git a/examples/tips/subscriber_test.cc b/examples/tips/subscriber_test.cc
index fda8909..736e6da 100644
--- a/examples/tips/subscriber_test.cc
+++ b/examples/tips/subscriber_test.cc
@@ -106,7 +106,7 @@
server_address_ << "localhost:" << port;
ServerBuilder builder;
builder.AddPort(server_address_.str());
- builder.RegisterService(service_.service());
+ builder.RegisterService(&service_);
server_ = builder.BuildAndStart();
channel_ = CreateChannel(server_address_.str(), ChannelArguments());
diff --git a/include/grpc++/async_server.h b/include/grpc++/async_server.h
deleted file mode 100644
index fe2c5d9..0000000
--- a/include/grpc++/async_server.h
+++ /dev/null
@@ -1,70 +0,0 @@
-/*
- *
- * Copyright 2014, Google Inc.
- * All rights reserved.
- *
- * Redistribution and use in source and binary forms, with or without
- * modification, are permitted provided that the following conditions are
- * met:
- *
- * * Redistributions of source code must retain the above copyright
- * notice, this list of conditions and the following disclaimer.
- * * Redistributions in binary form must reproduce the above
- * copyright notice, this list of conditions and the following disclaimer
- * in the documentation and/or other materials provided with the
- * distribution.
- * * Neither the name of Google Inc. nor the names of its
- * contributors may be used to endorse or promote products derived from
- * this software without specific prior written permission.
- *
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
- * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
- * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
- * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
- * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
- * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
- * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
- * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
- * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
- * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
- * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
- *
- */
-
-#ifndef __GRPCPP_ASYNC_SERVER_H__
-#define __GRPCPP_ASYNC_SERVER_H__
-
-#include <mutex>
-
-#include <grpc++/config.h>
-
-struct grpc_server;
-
-namespace grpc {
-class CompletionQueue;
-
-class AsyncServer {
- public:
- explicit AsyncServer(CompletionQueue* cc);
- ~AsyncServer();
-
- void AddPort(const grpc::string& addr);
-
- void Start();
-
- // The user has to call this to get one new rpc on the completion
- // queue.
- void RequestOneRpc();
-
- void Shutdown();
-
- private:
- bool started_;
- std::mutex shutdown_mu_;
- bool shutdown_;
- grpc_server* server_;
-};
-
-} // namespace grpc
-
-#endif // __GRPCPP_ASYNC_SERVER_H__
diff --git a/include/grpc++/async_server_context.h b/include/grpc++/async_server_context.h
deleted file mode 100644
index c038286..0000000
--- a/include/grpc++/async_server_context.h
+++ /dev/null
@@ -1,95 +0,0 @@
-/*
- *
- * Copyright 2014, Google Inc.
- * All rights reserved.
- *
- * Redistribution and use in source and binary forms, with or without
- * modification, are permitted provided that the following conditions are
- * met:
- *
- * * Redistributions of source code must retain the above copyright
- * notice, this list of conditions and the following disclaimer.
- * * Redistributions in binary form must reproduce the above
- * copyright notice, this list of conditions and the following disclaimer
- * in the documentation and/or other materials provided with the
- * distribution.
- * * Neither the name of Google Inc. nor the names of its
- * contributors may be used to endorse or promote products derived from
- * this software without specific prior written permission.
- *
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
- * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
- * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
- * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
- * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
- * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
- * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
- * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
- * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
- * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
- * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
- *
- */
-
-#ifndef __GRPCPP_ASYNC_SERVER_CONTEXT_H__
-#define __GRPCPP_ASYNC_SERVER_CONTEXT_H__
-
-#include <chrono>
-
-#include <grpc++/config.h>
-
-struct grpc_byte_buffer;
-struct grpc_call;
-struct grpc_completion_queue;
-
-namespace google {
-namespace protobuf {
-class Message;
-}
-}
-
-using std::chrono::system_clock;
-
-namespace grpc {
-class Status;
-
-// TODO(rocking): wrap grpc c structures.
-class AsyncServerContext {
- public:
- AsyncServerContext(grpc_call* call, const grpc::string& method,
- const grpc::string& host,
- system_clock::time_point absolute_deadline);
- ~AsyncServerContext();
-
- // Accept this rpc, bind it to a completion queue.
- void Accept(grpc_completion_queue* cq);
-
- // Read and write calls, all async. Return true for success.
- bool StartRead(google::protobuf::Message* request);
- bool StartWrite(const google::protobuf::Message& response, int flags);
- bool StartWriteStatus(const Status& status);
-
- bool ParseRead(grpc_byte_buffer* read_buffer);
-
- grpc::string method() const { return method_; }
- grpc::string host() const { return host_; }
- system_clock::time_point absolute_deadline() { return absolute_deadline_; }
-
- grpc_call* call() { return call_; }
-
- private:
- AsyncServerContext(const AsyncServerContext&);
- AsyncServerContext& operator=(const AsyncServerContext&);
-
- // These properties may be moved to a ServerContext class.
- const grpc::string method_;
- const grpc::string host_;
- system_clock::time_point absolute_deadline_;
-
- google::protobuf::Message* request_; // not owned
- grpc_call* call_; // owned
-};
-
-} // namespace grpc
-
-#endif // __GRPCPP_ASYNC_SERVER_CONTEXT_H__
diff --git a/include/grpc++/channel_interface.h b/include/grpc++/channel_interface.h
index 9ed3542..b0366fa 100644
--- a/include/grpc++/channel_interface.h
+++ b/include/grpc++/channel_interface.h
@@ -35,32 +35,30 @@
#define __GRPCPP_CHANNEL_INTERFACE_H__
#include <grpc++/status.h>
+#include <grpc++/impl/call.h>
namespace google {
namespace protobuf {
class Message;
-}
-}
+} // namespace protobuf
+} // namespace google
+
+struct grpc_call;
namespace grpc {
-
+class Call;
+class CallOpBuffer;
class ClientContext;
+class CompletionQueue;
class RpcMethod;
-class StreamContextInterface;
+class CallInterface;
-class ChannelInterface {
+class ChannelInterface : public CallHook {
public:
virtual ~ChannelInterface() {}
- virtual Status StartBlockingRpc(const RpcMethod& method,
- ClientContext* context,
- const google::protobuf::Message& request,
- google::protobuf::Message* result) = 0;
-
- virtual StreamContextInterface* CreateStream(
- const RpcMethod& method, ClientContext* context,
- const google::protobuf::Message* request,
- google::protobuf::Message* result) = 0;
+ virtual Call CreateCall(const RpcMethod &method, ClientContext *context,
+ CompletionQueue *cq) = 0;
};
} // namespace grpc
diff --git a/include/grpc++/client_context.h b/include/grpc++/client_context.h
index 0cf6bdc..0a81f6a 100644
--- a/include/grpc++/client_context.h
+++ b/include/grpc++/client_context.h
@@ -35,8 +35,8 @@
#define __GRPCPP_CLIENT_CONTEXT_H__
#include <chrono>
+#include <map>
#include <string>
-#include <vector>
#include <grpc/support/log.h>
#include <grpc/support/time.h>
@@ -49,6 +49,14 @@
namespace grpc {
+class CallOpBuffer;
+template <class R> class ClientReader;
+template <class W> class ClientWriter;
+template <class R, class W> class ClientReaderWriter;
+template <class R> class ClientAsyncReader;
+template <class W> class ClientAsyncWriter;
+template <class R, class W> class ClientAsyncReaderWriter;
+
class ClientContext {
public:
ClientContext();
@@ -67,8 +75,14 @@
ClientContext(const ClientContext &);
ClientContext &operator=(const ClientContext &);
+ friend class CallOpBuffer;
friend class Channel;
- friend class StreamContext;
+ template <class R> friend class ::grpc::ClientReader;
+ template <class W> friend class ::grpc::ClientWriter;
+ template <class R, class W> friend class ::grpc::ClientReaderWriter;
+ template <class R> friend class ::grpc::ClientAsyncReader;
+ template <class W> friend class ::grpc::ClientAsyncWriter;
+ template <class R, class W> friend class ::grpc::ClientAsyncReaderWriter;
grpc_call *call() { return call_; }
void set_call(grpc_call *call) {
@@ -84,7 +98,9 @@
grpc_call *call_;
grpc_completion_queue *cq_;
gpr_timespec absolute_deadline_;
- std::vector<std::pair<grpc::string, grpc::string> > metadata_;
+ std::multimap<grpc::string, grpc::string> send_initial_metadata_;
+ std::multimap<grpc::string, grpc::string> recv_initial_metadata_;
+ std::multimap<grpc::string, grpc::string> trailing_metadata_;
};
} // namespace grpc
diff --git a/include/grpc++/completion_queue.h b/include/grpc++/completion_queue.h
index 72f6253..3e68cf3 100644
--- a/include/grpc++/completion_queue.h
+++ b/include/grpc++/completion_queue.h
@@ -34,51 +34,69 @@
#ifndef __GRPCPP_COMPLETION_QUEUE_H__
#define __GRPCPP_COMPLETION_QUEUE_H__
+#include <grpc++/impl/client_unary_call.h>
+
struct grpc_completion_queue;
namespace grpc {
+template <class R>
+class ClientReader;
+template <class W>
+class ClientWriter;
+template <class R, class W>
+class ClientReaderWriter;
+template <class R>
+class ServerReader;
+template <class W>
+class ServerWriter;
+template <class R, class W>
+class ServerReaderWriter;
+
+class CompletionQueue;
+class Server;
+
+class CompletionQueueTag {
+ public:
+ // Called prior to returning from Next(), return value
+ // is the status of the operation (return status is the default thing
+ // to do)
+ virtual void FinalizeResult(void **tag, bool *status) = 0;
+};
+
// grpc_completion_queue wrapper class
class CompletionQueue {
public:
CompletionQueue();
+ explicit CompletionQueue(grpc_completion_queue *take);
~CompletionQueue();
- enum CompletionType {
- QUEUE_CLOSED = 0, // Shutting down.
- RPC_END = 1, // An RPC finished. Either at client or server.
- CLIENT_READ_OK = 2, // A client-side read has finished successfully.
- CLIENT_READ_ERROR = 3, // A client-side read has finished with error.
- CLIENT_WRITE_OK = 4,
- CLIENT_WRITE_ERROR = 5,
- SERVER_RPC_NEW = 6, // A new RPC just arrived at the server.
- SERVER_READ_OK = 7, // A server-side read has finished successfully.
- SERVER_READ_ERROR = 8, // A server-side read has finished with error.
- SERVER_WRITE_OK = 9,
- SERVER_WRITE_ERROR = 10,
- // Client or server has sent half close successfully.
- HALFCLOSE_OK = 11,
- // New CompletionTypes may be added in the future, so user code should
- // always
- // handle the default case of a CompletionType that appears after such code
- // was
- // written.
- DO_NOT_USE = 20,
- };
-
// Blocking read from queue.
- // For QUEUE_CLOSED, *tag is not changed.
- // For SERVER_RPC_NEW, *tag will be a newly allocated AsyncServerContext.
- // For others, *tag will be the AsyncServerContext of this rpc.
- CompletionType Next(void** tag);
+ // Returns true if an event was received, false if the queue is ready
+ // for destruction.
+ bool Next(void **tag, bool *ok);
// Shutdown has to be called, and the CompletionQueue can only be
- // destructed when the QUEUE_CLOSED message has been read with Next().
+ // destructed when false is returned from Next().
void Shutdown();
grpc_completion_queue* cq() { return cq_; }
private:
+ template <class R> friend class ::grpc::ClientReader;
+ template <class W> friend class ::grpc::ClientWriter;
+ template <class R, class W> friend class ::grpc::ClientReaderWriter;
+ template <class R> friend class ::grpc::ServerReader;
+ template <class W> friend class ::grpc::ServerWriter;
+ template <class R, class W> friend class ::grpc::ServerReaderWriter;
+ friend class ::grpc::Server;
+ friend Status BlockingUnaryCall(ChannelInterface *channel, const RpcMethod &method,
+ ClientContext *context,
+ const google::protobuf::Message &request,
+ google::protobuf::Message *result);
+
+ bool Pluck(CompletionQueueTag *tag);
+
grpc_completion_queue* cq_; // owned
};
diff --git a/include/grpc++/config.h b/include/grpc++/config.h
index 52913fb..1b4b463 100644
--- a/include/grpc++/config.h
+++ b/include/grpc++/config.h
@@ -39,6 +39,7 @@
namespace grpc {
typedef std::string string;
-}
+
+} // namespace grpc
#endif // __GRPCPP_CONFIG_H__
diff --git a/include/grpc++/impl/call.h b/include/grpc++/impl/call.h
new file mode 100644
index 0000000..a1ef926
--- /dev/null
+++ b/include/grpc++/impl/call.h
@@ -0,0 +1,145 @@
+/*
+ *
+ * Copyright 2014, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ * * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#ifndef __GRPCPP_CALL_H__
+#define __GRPCPP_CALL_H__
+
+#include <grpc/grpc.h>
+#include <grpc++/status.h>
+#include <grpc++/completion_queue.h>
+
+#include <memory>
+#include <map>
+
+namespace google {
+namespace protobuf {
+class Message;
+} // namespace protobuf
+} // namespace google
+
+struct grpc_call;
+struct grpc_op;
+
+namespace grpc {
+
+class Call;
+
+class CallOpBuffer final : public CompletionQueueTag {
+ public:
+ CallOpBuffer() : return_tag_(this) {}
+
+ void Reset(void *next_return_tag);
+
+ // Does not take ownership.
+ void AddSendInitialMetadata(
+ std::multimap<grpc::string, grpc::string> *metadata);
+ void AddSendInitialMetadata(ClientContext *ctx);
+ void AddRecvInitialMetadata(
+ std::multimap<grpc::string, grpc::string> *metadata);
+ void AddSendMessage(const google::protobuf::Message &message);
+ void AddRecvMessage(google::protobuf::Message *message, bool* got_message);
+ void AddClientSendClose();
+ void AddClientRecvStatus(std::multimap<grpc::string, grpc::string> *metadata,
+ Status *status);
+ void AddServerSendStatus(std::multimap<grpc::string, grpc::string> *metadata,
+ const Status& status);
+ void AddServerRecvClose(bool* cancelled);
+
+ // INTERNAL API:
+
+ // Convert to an array of grpc_op elements
+ void FillOps(grpc_op *ops, size_t *nops);
+
+ // Called by completion queue just prior to returning from Next() or Pluck()
+ void FinalizeResult(void **tag, bool *status) override;
+
+ private:
+ void *return_tag_ = nullptr;
+ // Send initial metadata
+ bool send_initial_metadata_ = false;
+ size_t initial_metadata_count_ = 0;
+ grpc_metadata* initial_metadata_ = nullptr;
+ // Recv initial metadta
+ std::multimap<grpc::string, grpc::string>* recv_initial_metadata_ = nullptr;
+ grpc_metadata_array recv_initial_metadata_arr_ = {0, 0, nullptr};
+ // Send message
+ const google::protobuf::Message* send_message_ = nullptr;
+ grpc_byte_buffer* send_message_buf_ = nullptr;
+ // Recv message
+ google::protobuf::Message* recv_message_ = nullptr;
+ bool* got_message_ = nullptr;
+ grpc_byte_buffer* recv_message_buf_ = nullptr;
+ // Client send close
+ bool client_send_close_ = false;
+ // Client recv status
+ std::multimap<grpc::string, grpc::string>* recv_trailing_metadata_ = nullptr;
+ Status* recv_status_ = nullptr;
+ grpc_metadata_array recv_trailing_metadata_arr_ = {0, 0, nullptr};
+ grpc_status_code status_code_ = GRPC_STATUS_OK;
+ char *status_details_ = nullptr;
+ size_t status_details_capacity_ = 0;
+ // Server send status
+ const Status* send_status_ = nullptr;
+ size_t trailing_metadata_count_ = 0;
+ grpc_metadata *trailing_metadata_ = nullptr;
+ int cancelled_buf_;
+ bool *recv_closed_ = nullptr;
+};
+
+// Channel and Server implement this to allow them to hook performing ops
+class CallHook {
+ public:
+ virtual ~CallHook() {}
+ virtual void PerformOpsOnCall(CallOpBuffer *ops, Call *call) = 0;
+};
+
+// Straightforward wrapping of the C call object
+class Call final {
+ public:
+ /* call is owned by the caller */
+ Call(grpc_call *call, CallHook *call_hook_, CompletionQueue *cq);
+
+ void PerformOps(CallOpBuffer *buffer);
+
+ grpc_call *call() { return call_; }
+ CompletionQueue *cq() { return cq_; }
+
+ private:
+ CallHook *call_hook_;
+ CompletionQueue *cq_;
+ grpc_call* call_;
+};
+
+} // namespace grpc
+
+#endif // __GRPCPP_CALL_INTERFACE_H__
diff --git a/include/grpc++/impl/client_unary_call.h b/include/grpc++/impl/client_unary_call.h
new file mode 100644
index 0000000..091430b
--- /dev/null
+++ b/include/grpc++/impl/client_unary_call.h
@@ -0,0 +1,67 @@
+/*
+*
+* Copyright 2014, Google Inc.
+* All rights reserved.
+*
+* Redistribution and use in source and binary forms, with or without
+* modification, are permitted provided that the following conditions are
+* met:
+*
+* * Redistributions of source code must retain the above copyright
+* notice, this list of conditions and the following disclaimer.
+* * Redistributions in binary form must reproduce the above
+* copyright notice, this list of conditions and the following disclaimer
+* in the documentation and/or other materials provided with the
+* distribution.
+* * Neither the name of Google Inc. nor the names of its
+* contributors may be used to endorse or promote products derived from
+* this software without specific prior written permission.
+*
+* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+*
+*/
+
+#ifndef __GRPCPP_CLIENT_UNARY_CALL_H__
+#define __GRPCPP_CLIENT_UNARY_CALL_H__
+
+namespace google {
+namespace protobuf {
+class Message;
+} // namespace protobuf
+} // namespace google
+
+namespace grpc {
+
+class ChannelInterface;
+class ClientContext;
+class CompletionQueue;
+class RpcMethod;
+class Status;
+
+// Wrapper that begins an asynchronous unary call
+void AsyncUnaryCall(ChannelInterface *channel, const RpcMethod &method,
+ ClientContext *context,
+ const google::protobuf::Message &request,
+ google::protobuf::Message *result, Status *status,
+ CompletionQueue *cq, void *tag);
+
+// Wrapper that performs a blocking unary call
+Status BlockingUnaryCall(ChannelInterface *channel, const RpcMethod &method,
+ ClientContext *context,
+ const google::protobuf::Message &request,
+ google::protobuf::Message *result);
+
+} // namespace grpc
+
+#endif
+
diff --git a/include/grpc++/impl/rpc_method.h b/include/grpc++/impl/rpc_method.h
index 75fec35..bb16e64 100644
--- a/include/grpc++/impl/rpc_method.h
+++ b/include/grpc++/impl/rpc_method.h
@@ -37,8 +37,8 @@
namespace google {
namespace protobuf {
class Message;
-}
-}
+} // namespace protobuf
+} // namespace google
namespace grpc {
diff --git a/include/grpc++/impl/rpc_service_method.h b/include/grpc++/impl/rpc_service_method.h
index 620de5e..c201676 100644
--- a/include/grpc++/impl/rpc_service_method.h
+++ b/include/grpc++/impl/rpc_service_method.h
@@ -55,25 +55,18 @@
public:
virtual ~MethodHandler() {}
struct HandlerParameter {
- HandlerParameter(ServerContext* context,
+ HandlerParameter(Call *c,
+ ServerContext* context,
const google::protobuf::Message* req,
google::protobuf::Message* resp)
- : server_context(context),
+ : call(c),
+ server_context(context),
request(req),
- response(resp),
- stream_context(nullptr) {}
- HandlerParameter(ServerContext* context,
- const google::protobuf::Message* req,
- google::protobuf::Message* resp,
- StreamContextInterface* stream)
- : server_context(context),
- request(req),
- response(resp),
- stream_context(stream) {}
+ response(resp) {}
+ Call* call;
ServerContext* server_context;
const google::protobuf::Message* request;
google::protobuf::Message* response;
- StreamContextInterface* stream_context;
};
virtual Status RunHandler(const HandlerParameter& param) = 0;
};
@@ -114,7 +107,7 @@
: func_(func), service_(service) {}
Status RunHandler(const HandlerParameter& param) final {
- ServerReader<RequestType> reader(param.stream_context);
+ ServerReader<RequestType> reader(param.call, param.server_context);
return func_(service_, param.server_context, &reader,
dynamic_cast<ResponseType*>(param.response));
}
@@ -136,7 +129,7 @@
: func_(func), service_(service) {}
Status RunHandler(const HandlerParameter& param) final {
- ServerWriter<ResponseType> writer(param.stream_context);
+ ServerWriter<ResponseType> writer(param.call, param.server_context);
return func_(service_, param.server_context,
dynamic_cast<const RequestType*>(param.request), &writer);
}
@@ -159,7 +152,7 @@
: func_(func), service_(service) {}
Status RunHandler(const HandlerParameter& param) final {
- ServerReaderWriter<ResponseType, RequestType> stream(param.stream_context);
+ ServerReaderWriter<ResponseType, RequestType> stream(param.call, param.server_context);
return func_(service_, param.server_context, &stream);
}
@@ -203,7 +196,7 @@
public:
// Takes ownership.
void AddMethod(RpcServiceMethod* method) {
- methods_.push_back(std::unique_ptr<RpcServiceMethod>(method));
+ methods_.emplace_back(method);
}
RpcServiceMethod* GetMethod(int i) { return methods_[i].get(); }
diff --git a/src/cpp/server/server_context_impl.cc b/include/grpc++/impl/service_type.h
similarity index 78%
rename from src/cpp/server/server_context_impl.cc
rename to include/grpc++/impl/service_type.h
index 467cc80..0684f32 100644
--- a/src/cpp/server/server_context_impl.cc
+++ b/include/grpc++/impl/service_type.h
@@ -31,6 +31,25 @@
*
*/
-#include "src/cpp/server/server_context_impl.h"
+#ifndef __GRPCPP_IMPL_SERVICE_TYPE_H__
+#define __GRPCPP_IMPL_SERVICE_TYPE_H__
-namespace grpc {} // namespace grpc
+namespace grpc {
+
+class RpcService;
+
+class SynchronousService {
+ public:
+ virtual ~SynchronousService() {}
+ virtual RpcService *service() = 0;
+};
+
+class AsynchronousService {
+ public:
+ virtual ~AsynchronousService() {}
+ virtual RpcService *service() = 0;
+};
+
+} // namespace grpc
+
+#endif // __GRPCPP_IMPL_SERVICE_TYPE_H__
\ No newline at end of file
diff --git a/include/grpc++/server.h b/include/grpc++/server.h
index 5fa371b..98f3f17 100644
--- a/include/grpc++/server.h
+++ b/include/grpc++/server.h
@@ -35,12 +35,13 @@
#define __GRPCPP_SERVER_H__
#include <condition_variable>
-#include <map>
+#include <list>
#include <memory>
#include <mutex>
#include <grpc++/completion_queue.h>
#include <grpc++/config.h>
+#include <grpc++/impl/call.h>
#include <grpc++/status.h>
struct grpc_server;
@@ -48,8 +49,8 @@
namespace google {
namespace protobuf {
class Message;
-}
-}
+} // namespace protobuf
+} // namespace google
namespace grpc {
class AsyncServerContext;
@@ -59,7 +60,7 @@
class ThreadPoolInterface;
// Currently it only supports handling rpcs in a single thread.
-class Server {
+class Server final : private CallHook {
public:
~Server();
@@ -69,22 +70,26 @@
private:
friend class ServerBuilder;
+ class MethodRequestData;
+
// ServerBuilder use only
- Server(ThreadPoolInterface* thread_pool, ServerCredentials* creds);
+ Server(ThreadPoolInterface* thread_pool, bool thread_pool_owned,
+ ServerCredentials* creds);
Server();
// Register a service. This call does not take ownership of the service.
// The service must exist for the lifetime of the Server instance.
- void RegisterService(RpcService* service);
+ bool RegisterService(RpcService* service);
// Add a listening port. Can be called multiple times.
- void AddPort(const grpc::string& addr);
+ int AddPort(const grpc::string& addr);
// Start the server.
- void Start();
+ bool Start();
- void AllowOneRpc();
void HandleQueueClosed();
void RunRpc();
void ScheduleCallback();
+ void PerformOpsOnCall(CallOpBuffer* ops, Call* call) override;
+
// Completion queue.
CompletionQueue cq_;
@@ -96,12 +101,11 @@
int num_running_cb_;
std::condition_variable callback_cv_;
+ std::list<MethodRequestData> methods_;
+
// Pointer to the c grpc server.
grpc_server* server_;
- // A map for all method information.
- std::map<grpc::string, RpcServiceMethod*> method_map_;
-
ThreadPoolInterface* thread_pool_;
// Whether the thread pool is created and owned by the server.
bool thread_pool_owned_;
diff --git a/include/grpc++/server_builder.h b/include/grpc++/server_builder.h
index cf27452..8b4c81b 100644
--- a/include/grpc++/server_builder.h
+++ b/include/grpc++/server_builder.h
@@ -41,9 +41,11 @@
namespace grpc {
+class AsynchronousService;
class RpcService;
class Server;
class ServerCredentials;
+class SynchronousService;
class ThreadPoolInterface;
class ServerBuilder {
@@ -53,7 +55,9 @@
// Register a service. This call does not take ownership of the service.
// The service must exist for the lifetime of the Server instance returned by
// BuildAndStart().
- void RegisterService(RpcService* service);
+ void RegisterService(SynchronousService* service);
+
+ void RegisterAsyncService(AsynchronousService *service);
// Add a listening port. Can be called multiple times.
void AddPort(const grpc::string& addr);
@@ -71,9 +75,10 @@
private:
std::vector<RpcService*> services_;
+ std::vector<AsynchronousService*> async_services_;
std::vector<grpc::string> ports_;
std::shared_ptr<ServerCredentials> creds_;
- ThreadPoolInterface* thread_pool_;
+ ThreadPoolInterface* thread_pool_ = nullptr;
};
} // namespace grpc
diff --git a/include/grpc++/server_context.h b/include/grpc++/server_context.h
index 47fd6cf..a30f6cb 100644
--- a/include/grpc++/server_context.h
+++ b/include/grpc++/server_context.h
@@ -35,15 +35,53 @@
#define __GRPCPP_SERVER_CONTEXT_H_
#include <chrono>
+#include <map>
+
+#include "config.h"
+
+struct grpc_metadata;
+struct gpr_timespec;
namespace grpc {
+template <class R> class ServerAsyncReader;
+template <class W> class ServerAsyncWriter;
+template <class R, class W> class ServerAsyncReaderWriter;
+template <class R> class ServerReader;
+template <class W> class ServerWriter;
+template <class R, class W> class ServerReaderWriter;
+
+class CallOpBuffer;
+class Server;
+
// Interface of server side rpc context.
class ServerContext {
public:
virtual ~ServerContext() {}
- virtual std::chrono::system_clock::time_point absolute_deadline() const = 0;
+ std::chrono::system_clock::time_point absolute_deadline() { return deadline_; }
+
+ void AddInitialMetadata(const grpc::string& key, const grpc::string& value);
+ void AddTrailingMetadata(const grpc::string& key, const grpc::string& value);
+
+ private:
+ friend class ::grpc::Server;
+ template <class R> friend class ::grpc::ServerAsyncReader;
+ template <class W> friend class ::grpc::ServerAsyncWriter;
+ template <class R, class W> friend class ::grpc::ServerAsyncReaderWriter;
+ template <class R> friend class ::grpc::ServerReader;
+ template <class W> friend class ::grpc::ServerWriter;
+ template <class R, class W> friend class ::grpc::ServerReaderWriter;
+
+ ServerContext(gpr_timespec deadline, grpc_metadata *metadata, size_t metadata_count);
+
+ void SendInitialMetadataIfNeeded(CallOpBuffer *buf);
+
+ const std::chrono::system_clock::time_point deadline_;
+ bool sent_initial_metadata_ = false;
+ std::multimap<grpc::string, grpc::string> client_metadata_;
+ std::multimap<grpc::string, grpc::string> initial_metadata_;
+ std::multimap<grpc::string, grpc::string> trailing_metadata_;
};
} // namespace grpc
diff --git a/include/grpc++/stream.h b/include/grpc++/stream.h
index b8982f4..a59ccd8 100644
--- a/include/grpc++/stream.h
+++ b/include/grpc++/stream.h
@@ -34,7 +34,11 @@
#ifndef __GRPCPP_STREAM_H__
#define __GRPCPP_STREAM_H__
-#include <grpc++/stream_context_interface.h>
+#include <grpc++/channel_interface.h>
+#include <grpc++/client_context.h>
+#include <grpc++/completion_queue.h>
+#include <grpc++/server_context.h>
+#include <grpc++/impl/call.h>
#include <grpc++/status.h>
#include <grpc/support/log.h>
@@ -45,16 +49,12 @@
public:
virtual ~ClientStreamingInterface() {}
- // Try to cancel the stream. Wait() still needs to be called to get the final
- // status. Cancelling after the stream has finished has no effects.
- virtual void Cancel() = 0;
-
// Wait until the stream finishes, and return the final status. When the
// client side declares it has no more message to send, either implicitly or
// by calling WritesDone, it needs to make sure there is no more message to
// be received from the server, either implicitly or by getting a false from
// a Read(). Otherwise, this implicitly cancels the stream.
- virtual const Status& Wait() = 0;
+ virtual Status Finish() = 0;
};
// An interface that yields a sequence of R messages.
@@ -82,147 +82,415 @@
};
template <class R>
-class ClientReader : public ClientStreamingInterface,
- public ReaderInterface<R> {
+class ClientReader final : public ClientStreamingInterface,
+ public ReaderInterface<R> {
public:
// Blocking create a stream and write the first request out.
- explicit ClientReader(StreamContextInterface* context) : context_(context) {
- GPR_ASSERT(context_);
- context_->Start(true);
- context_->Write(context_->request(), true);
+ ClientReader(ChannelInterface *channel, const RpcMethod &method,
+ ClientContext *context,
+ const google::protobuf::Message &request)
+ : context_(context), call_(channel->CreateCall(method, context, &cq_)) {
+ CallOpBuffer buf;
+ buf.AddSendInitialMetadata(&context->send_initial_metadata_);
+ buf.AddSendMessage(request);
+ buf.AddClientSendClose();
+ call_.PerformOps(&buf);
+ cq_.Pluck(&buf);
}
- ~ClientReader() { delete context_; }
+ virtual bool Read(R *msg) override {
+ CallOpBuffer buf;
+ bool got_message;
+ buf.AddRecvMessage(msg, &got_message);
+ call_.PerformOps(&buf);
+ return cq_.Pluck(&buf) && got_message;
+ }
- virtual bool Read(R* msg) { return context_->Read(msg); }
-
- virtual void Cancel() { context_->Cancel(); }
-
- virtual const Status& Wait() { return context_->Wait(); }
+ virtual Status Finish() override {
+ CallOpBuffer buf;
+ Status status;
+ buf.AddClientRecvStatus(&context_->trailing_metadata_, &status);
+ call_.PerformOps(&buf);
+ GPR_ASSERT(cq_.Pluck(&buf));
+ return status;
+ }
private:
- StreamContextInterface* const context_;
+ ClientContext* context_;
+ CompletionQueue cq_;
+ Call call_;
};
template <class W>
-class ClientWriter : public ClientStreamingInterface,
- public WriterInterface<W> {
+class ClientWriter final : public ClientStreamingInterface,
+ public WriterInterface<W> {
public:
// Blocking create a stream.
- explicit ClientWriter(StreamContextInterface* context) : context_(context) {
- GPR_ASSERT(context_);
- context_->Start(false);
+ ClientWriter(ChannelInterface *channel, const RpcMethod &method,
+ ClientContext *context,
+ google::protobuf::Message *response)
+ : context_(context), response_(response),
+ call_(channel->CreateCall(method, context, &cq_)) {
+ CallOpBuffer buf;
+ buf.AddSendInitialMetadata(&context->send_initial_metadata_);
+ call_.PerformOps(&buf);
+ cq_.Pluck(&buf);
}
- ~ClientWriter() { delete context_; }
-
- virtual bool Write(const W& msg) {
- return context_->Write(const_cast<W*>(&msg), false);
+ virtual bool Write(const W& msg) override {
+ CallOpBuffer buf;
+ buf.AddSendMessage(msg);
+ call_.PerformOps(&buf);
+ return cq_.Pluck(&buf);
}
- virtual void WritesDone() { context_->Write(nullptr, true); }
-
- virtual void Cancel() { context_->Cancel(); }
+ virtual bool WritesDone() {
+ CallOpBuffer buf;
+ buf.AddClientSendClose();
+ call_.PerformOps(&buf);
+ return cq_.Pluck(&buf);
+ }
// Read the final response and wait for the final status.
- virtual const Status& Wait() {
- bool success = context_->Read(context_->response());
- if (!success) {
- Cancel();
- } else {
- success = context_->Read(nullptr);
- if (success) {
- Cancel();
- }
- }
- return context_->Wait();
+ virtual Status Finish() override {
+ CallOpBuffer buf;
+ Status status;
+ bool got_message;
+ buf.AddRecvMessage(response_, &got_message);
+ buf.AddClientRecvStatus(&context_->trailing_metadata_, &status);
+ call_.PerformOps(&buf);
+ GPR_ASSERT(cq_.Pluck(&buf) && got_message);
+ return status;
}
private:
- StreamContextInterface* const context_;
+ ClientContext* context_;
+ google::protobuf::Message *const response_;
+ CompletionQueue cq_;
+ Call call_;
};
// Client-side interface for bi-directional streaming.
template <class W, class R>
-class ClientReaderWriter : public ClientStreamingInterface,
- public WriterInterface<W>,
- public ReaderInterface<R> {
+class ClientReaderWriter final : public ClientStreamingInterface,
+ public WriterInterface<W>,
+ public ReaderInterface<R> {
public:
// Blocking create a stream.
- explicit ClientReaderWriter(StreamContextInterface* context)
- : context_(context) {
- GPR_ASSERT(context_);
- context_->Start(false);
+ ClientReaderWriter(ChannelInterface *channel,
+ const RpcMethod &method, ClientContext *context)
+ : context_(context), call_(channel->CreateCall(method, context, &cq_)) {
+ CallOpBuffer buf;
+ buf.AddSendInitialMetadata(&context->send_initial_metadata_);
+ call_.PerformOps(&buf);
+ GPR_ASSERT(cq_.Pluck(&buf));
}
- ~ClientReaderWriter() { delete context_; }
-
- virtual bool Read(R* msg) { return context_->Read(msg); }
-
- virtual bool Write(const W& msg) {
- return context_->Write(const_cast<W*>(&msg), false);
+ virtual bool Read(R *msg) override {
+ CallOpBuffer buf;
+ bool got_message;
+ buf.AddRecvMessage(msg, &got_message);
+ call_.PerformOps(&buf);
+ return cq_.Pluck(&buf) && got_message;
}
- virtual void WritesDone() { context_->Write(nullptr, true); }
+ virtual bool Write(const W& msg) override {
+ CallOpBuffer buf;
+ buf.AddSendMessage(msg);
+ call_.PerformOps(&buf);
+ return cq_.Pluck(&buf);
+ }
- virtual void Cancel() { context_->Cancel(); }
+ virtual bool WritesDone() {
+ CallOpBuffer buf;
+ buf.AddClientSendClose();
+ call_.PerformOps(&buf);
+ return cq_.Pluck(&buf);
+ }
- virtual const Status& Wait() { return context_->Wait(); }
+ virtual Status Finish() override {
+ CallOpBuffer buf;
+ Status status;
+ buf.AddClientRecvStatus(&context_->trailing_metadata_, &status);
+ call_.PerformOps(&buf);
+ GPR_ASSERT(cq_.Pluck(&buf));
+ return status;
+ }
private:
- StreamContextInterface* const context_;
+ ClientContext* context_;
+ CompletionQueue cq_;
+ Call call_;
};
template <class R>
-class ServerReader : public ReaderInterface<R> {
+class ServerReader final : public ReaderInterface<R> {
public:
- explicit ServerReader(StreamContextInterface* context) : context_(context) {
- GPR_ASSERT(context_);
- context_->Start(true);
+ explicit ServerReader(Call* call, ServerContext* ctx) : call_(call), ctx_(ctx) {}
+
+ virtual bool Read(R* msg) override {
+ CallOpBuffer buf;
+ bool got_message;
+ buf.AddRecvMessage(msg, &got_message);
+ call_->PerformOps(&buf);
+ return call_->cq()->Pluck(&buf) && got_message;
}
- virtual bool Read(R* msg) { return context_->Read(msg); }
-
private:
- StreamContextInterface* const context_; // not owned
+ Call* const call_;
+ ServerContext* const ctx_;
};
template <class W>
-class ServerWriter : public WriterInterface<W> {
+class ServerWriter final : public WriterInterface<W> {
public:
- explicit ServerWriter(StreamContextInterface* context) : context_(context) {
- GPR_ASSERT(context_);
- context_->Start(true);
- context_->Read(context_->request());
- }
+ explicit ServerWriter(Call* call, ServerContext* ctx) : call_(call), ctx_(ctx) {}
- virtual bool Write(const W& msg) {
- return context_->Write(const_cast<W*>(&msg), false);
+ virtual bool Write(const W& msg) override {
+ CallOpBuffer buf;
+ ctx_->SendInitialMetadataIfNeeded(&buf);
+ buf.AddSendMessage(msg);
+ call_->PerformOps(&buf);
+ return call_->cq()->Pluck(&buf);
}
private:
- StreamContextInterface* const context_; // not owned
+ Call* const call_;
+ ServerContext* const ctx_;
};
// Server-side interface for bi-directional streaming.
template <class W, class R>
-class ServerReaderWriter : public WriterInterface<W>,
+class ServerReaderWriter final : public WriterInterface<W>,
public ReaderInterface<R> {
public:
- explicit ServerReaderWriter(StreamContextInterface* context)
- : context_(context) {
- GPR_ASSERT(context_);
- context_->Start(true);
+ explicit ServerReaderWriter(Call* call, ServerContext* ctx) : call_(call), ctx_(ctx) {}
+
+ virtual bool Read(R* msg) override {
+ CallOpBuffer buf;
+ bool got_message;
+ buf.AddRecvMessage(msg, &got_message);
+ call_->PerformOps(&buf);
+ return call_->cq()->Pluck(&buf) && got_message;
}
- virtual bool Read(R* msg) { return context_->Read(msg); }
-
- virtual bool Write(const W& msg) {
- return context_->Write(const_cast<W*>(&msg), false);
+ virtual bool Write(const W& msg) override {
+ CallOpBuffer buf;
+ ctx_->SendInitialMetadataIfNeeded(&buf);
+ buf.AddSendMessage(msg);
+ call_->PerformOps(&buf);
+ return call_->cq()->Pluck(&buf);
}
private:
- StreamContextInterface* const context_; // not owned
+ Call* const call_;
+ ServerContext* const ctx_;
+};
+
+// Async interfaces
+// Common interface for all client side streaming.
+class ClientAsyncStreamingInterface {
+ public:
+ virtual ~ClientAsyncStreamingInterface() {}
+
+ virtual void Finish(Status* status, void* tag) = 0;
+};
+
+// An interface that yields a sequence of R messages.
+template <class R>
+class AsyncReaderInterface {
+ public:
+ virtual ~AsyncReaderInterface() {}
+
+ virtual void Read(R* msg, void* tag) = 0;
+};
+
+// An interface that can be fed a sequence of W messages.
+template <class W>
+class AsyncWriterInterface {
+ public:
+ virtual ~AsyncWriterInterface() {}
+
+ virtual void Write(const W& msg, void* tag) = 0;
+};
+
+template <class R>
+class ClientAsyncReader final : public ClientAsyncStreamingInterface,
+ public AsyncReaderInterface<R> {
+ public:
+ // Blocking create a stream and write the first request out.
+ ClientAsyncReader(ChannelInterface *channel, const RpcMethod &method,
+ ClientContext *context,
+ const google::protobuf::Message &request, void* tag)
+ : call_(channel->CreateCall(method, context, &cq_)) {
+ init_buf_.Reset(tag);
+ init_buf_.AddSendMessage(request);
+ init_buf_.AddClientSendClose();
+ call_.PerformOps(&init_buf_);
+ }
+
+ virtual void Read(R *msg, void* tag) override {
+ read_buf_.Reset(tag);
+ read_buf_.AddRecvMessage(msg);
+ call_.PerformOps(&read_buf_);
+ }
+
+ virtual void Finish(Status* status, void* tag) override {
+ finish_buf_.Reset(tag);
+ finish_buf_.AddClientRecvStatus(nullptr, status); // TODO metadata
+ call_.PerformOps(&finish_buf_);
+ }
+
+ private:
+ CompletionQueue cq_;
+ Call call_;
+ CallOpBuffer init_buf_;
+ CallOpBuffer read_buf_;
+ CallOpBuffer finish_buf_;
+};
+
+template <class W>
+class ClientAsyncWriter final : public ClientAsyncStreamingInterface,
+ public WriterInterface<W> {
+ public:
+ // Blocking create a stream.
+ ClientAsyncWriter(ChannelInterface *channel, const RpcMethod &method,
+ ClientContext *context,
+ google::protobuf::Message *response)
+ : response_(response),
+ call_(channel->CreateCall(method, context, &cq_)) {}
+
+ virtual void Write(const W& msg, void* tag) override {
+ write_buf_.Reset(tag);
+ write_buf_.AddSendMessage(msg);
+ call_.PerformOps(&write_buf_);
+ }
+
+ virtual void WritesDone(void* tag) override {
+ writes_done_buf_.Reset(tag);
+ writes_done_buf_.AddClientSendClose();
+ call_.PerformOps(&writes_done_buf_);
+ }
+
+ virtual void Finish(Status* status, void* tag) override {
+ finish_buf_.Reset(tag);
+ finish_buf_.AddRecvMessage(response_, &got_message_);
+ finish_buf_.AddClientRecvStatus(nullptr, status); // TODO metadata
+ call_.PerformOps(&finish_buf_);
+ }
+
+ private:
+ google::protobuf::Message *const response_;
+ bool got_message_;
+ CompletionQueue cq_;
+ Call call_;
+ CallOpBuffer write_buf_;
+ CallOpBuffer writes_done_buf_;
+ CallOpBuffer finish_buf_;
+};
+
+// Client-side interface for bi-directional streaming.
+template <class W, class R>
+class ClientAsyncReaderWriter final : public ClientAsyncStreamingInterface,
+ public AsyncWriterInterface<W>,
+ public AsyncReaderInterface<R> {
+ public:
+ ClientAsyncReaderWriter(ChannelInterface *channel,
+ const RpcMethod &method, ClientContext *context)
+ : call_(channel->CreateCall(method, context, &cq_)) {}
+
+ virtual void Read(R *msg, void* tag) override {
+ read_buf_.Reset(tag);
+ read_buf_.AddRecvMessage(msg);
+ call_.PerformOps(&read_buf_);
+ }
+
+ virtual void Write(const W& msg, void* tag) override {
+ write_buf_.Reset(tag);
+ write_buf_.AddSendMessage(msg);
+ call_.PerformOps(&write_buf_);
+ }
+
+ virtual void WritesDone(void* tag) override {
+ writes_done_buf_.Reset(tag);
+ writes_done_buf_.AddClientSendClose();
+ call_.PerformOps(&writes_done_buf_);
+ }
+
+ virtual void Finish(Status* status, void* tag) override {
+ finish_buf_.Reset(tag);
+ finish_buf_.AddClientRecvStatus(nullptr, status); // TODO metadata
+ call_.PerformOps(&finish_buf_);
+ }
+
+ private:
+ CompletionQueue cq_;
+ Call call_;
+ CallOpBuffer read_buf_;
+ CallOpBuffer write_buf_;
+ CallOpBuffer writes_done_buf_;
+ CallOpBuffer finish_buf_;
+};
+
+// TODO(yangg) Move out of stream.h
+template <class W>
+class ServerAsyncResponseWriter final {
+ public:
+ explicit ServerAsyncResponseWriter(Call* call) : call_(call) {}
+
+ virtual void Write(const W& msg, void* tag) override {
+ CallOpBuffer buf;
+ buf.AddSendMessage(msg);
+ call_->PerformOps(&buf);
+ }
+
+ private:
+ Call* call_;
+};
+
+template <class R>
+class ServerAsyncReader : public AsyncReaderInterface<R> {
+ public:
+ explicit ServerAsyncReader(Call* call) : call_(call) {}
+
+ virtual void Read(R* msg, void* tag) {
+ // TODO
+ }
+
+ private:
+ Call* call_;
+};
+
+template <class W>
+class ServerAsyncWriter : public AsyncWriterInterface<W> {
+ public:
+ explicit ServerAsyncWriter(Call* call) : call_(call) {}
+
+ virtual void Write(const W& msg, void* tag) {
+ // TODO
+ }
+
+ private:
+ Call* call_;
+};
+
+// Server-side interface for bi-directional streaming.
+template <class W, class R>
+class ServerAsyncReaderWriter : public AsyncWriterInterface<W>,
+ public AsyncReaderInterface<R> {
+ public:
+ explicit ServerAsyncReaderWriter(Call* call) : call_(call) {}
+
+ virtual void Read(R* msg, void* tag) {
+ // TODO
+ }
+
+ virtual void Write(const W& msg, void* tag) {
+ // TODO
+ }
+
+ private:
+ Call* call_;
};
} // namespace grpc
diff --git a/include/grpc++/stream_context_interface.h b/include/grpc++/stream_context_interface.h
deleted file mode 100644
index a841198..0000000
--- a/include/grpc++/stream_context_interface.h
+++ /dev/null
@@ -1,64 +0,0 @@
-/*
- *
- * Copyright 2014, Google Inc.
- * All rights reserved.
- *
- * Redistribution and use in source and binary forms, with or without
- * modification, are permitted provided that the following conditions are
- * met:
- *
- * * Redistributions of source code must retain the above copyright
- * notice, this list of conditions and the following disclaimer.
- * * Redistributions in binary form must reproduce the above
- * copyright notice, this list of conditions and the following disclaimer
- * in the documentation and/or other materials provided with the
- * distribution.
- * * Neither the name of Google Inc. nor the names of its
- * contributors may be used to endorse or promote products derived from
- * this software without specific prior written permission.
- *
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
- * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
- * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
- * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
- * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
- * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
- * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
- * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
- * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
- * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
- * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
- *
- */
-
-#ifndef __GRPCPP_STREAM_CONTEXT_INTERFACE_H__
-#define __GRPCPP_STREAM_CONTEXT_INTERFACE_H__
-
-namespace google {
-namespace protobuf {
-class Message;
-}
-}
-
-namespace grpc {
-class Status;
-
-// An interface to avoid dependency on internal implementation.
-class StreamContextInterface {
- public:
- virtual ~StreamContextInterface() {}
-
- virtual void Start(bool buffered) = 0;
-
- virtual bool Read(google::protobuf::Message* msg) = 0;
- virtual bool Write(const google::protobuf::Message* msg, bool is_last) = 0;
- virtual const Status& Wait() = 0;
- virtual void Cancel() = 0;
-
- virtual google::protobuf::Message* request() = 0;
- virtual google::protobuf::Message* response() = 0;
-};
-
-} // namespace grpc
-
-#endif // __GRPCPP_STREAM_CONTEXT_INTERFACE_H__
diff --git a/include/grpc/grpc b/include/grpc/grpc
new file mode 120000
index 0000000..fc80ad1
--- /dev/null
+++ b/include/grpc/grpc
@@ -0,0 +1 @@
+/home/craig/grpc-ct/include/grpc
\ No newline at end of file
diff --git a/include/grpc/grpc.h b/include/grpc/grpc.h
index 7b33a4d..561bc9a 100644
--- a/include/grpc/grpc.h
+++ b/include/grpc/grpc.h
@@ -254,15 +254,18 @@
void grpc_call_details_destroy(grpc_call_details *details);
typedef enum {
- /* Send initial metadata: one and only one instance MUST be sent for each call,
+ /* Send initial metadata: one and only one instance MUST be sent for each
+ call,
unless the call was cancelled - in which case this can be skipped */
GRPC_OP_SEND_INITIAL_METADATA = 0,
/* Send a message: 0 or more of these operations can occur for each call */
GRPC_OP_SEND_MESSAGE,
- /* Send a close from the server: one and only one instance MUST be sent from the client,
+ /* Send a close from the server: one and only one instance MUST be sent from
+ the client,
unless the call was cancelled - in which case this can be skipped */
GRPC_OP_SEND_CLOSE_FROM_CLIENT,
- /* Send status from the server: one and only one instance MUST be sent from the server
+ /* Send status from the server: one and only one instance MUST be sent from
+ the server
unless the call was cancelled - in which case this can be skipped */
GRPC_OP_SEND_STATUS_FROM_SERVER,
/* Receive initial metadata: one and only one MUST be made on the client, must
@@ -270,13 +273,16 @@
GRPC_OP_RECV_INITIAL_METADATA,
/* Receive a message: 0 or more of these operations can occur for each call */
GRPC_OP_RECV_MESSAGE,
- /* Receive status on the client: one and only one must be made on the client */
+ /* Receive status on the client: one and only one must be made on the client
+ */
GRPC_OP_RECV_STATUS_ON_CLIENT,
- /* Receive status on the server: one and only one must be made on the server */
+ /* Receive status on the server: one and only one must be made on the server
+ */
GRPC_OP_RECV_CLOSE_ON_SERVER
} grpc_op_type;
-/* Operation data: one field for each op type (except SEND_CLOSE_FROM_CLIENT which has
+/* Operation data: one field for each op type (except SEND_CLOSE_FROM_CLIENT
+ which has
no arguments) */
typedef struct grpc_op {
grpc_op_type op;
@@ -300,29 +306,33 @@
grpc_metadata_array *recv_initial_metadata;
grpc_byte_buffer **recv_message;
struct {
- /* ownership of the array is with the caller, but ownership of the elements
+ /* ownership of the array is with the caller, but ownership of the
+ elements
stays with the call object (ie key, value members are owned by the call
object, trailing_metadata->array is owned by the caller).
After the operation completes, call grpc_metadata_array_destroy on this
value, or reuse it in a future op. */
grpc_metadata_array *trailing_metadata;
grpc_status_code *status;
- /* status_details is a buffer owned by the application before the op completes
- and after the op has completed. During the operation status_details may be
- reallocated to a size larger than *status_details_capacity, in which case
+ /* status_details is a buffer owned by the application before the op
+ completes
+ and after the op has completed. During the operation status_details may
+ be
+ reallocated to a size larger than *status_details_capacity, in which
+ case
*status_details_capacity will be updated with the new array capacity.
Pre-allocating space:
size_t my_capacity = 8;
char *my_details = gpr_malloc(my_capacity);
x.status_details = &my_details;
- x.status_details_capacity = &my_capacity;
+ x.status_details_capacity = &my_capacity;
Not pre-allocating space:
size_t my_capacity = 0;
char *my_details = NULL;
x.status_details = &my_details;
- x.status_details_capacity = &my_capacity;
+ x.status_details_capacity = &my_capacity;
After the call:
gpr_free(my_details); */
@@ -330,7 +340,8 @@
size_t *status_details_capacity;
} recv_status_on_client;
struct {
- /* out argument, set to 1 if the call failed in any way (seen as a cancellation
+ /* out argument, set to 1 if the call failed in any way (seen as a
+ cancellation
on the server), or 0 if the call succeeded */
int *cancelled;
} recv_close_on_server;
@@ -392,7 +403,7 @@
gpr_timespec deadline);
/* Start a batch of operations defined in the array ops; when complete, post a
- completion of type 'tag' to the completion queue bound to the call.
+ completion of type 'tag' to the completion queue bound to the call.
The order of ops specified in the batch has no significance.
Only one operation of each type can be active at once in any given
batch. */
@@ -542,7 +553,17 @@
grpc_call_error grpc_server_request_call(
grpc_server *server, grpc_call **call, grpc_call_details *details,
grpc_metadata_array *request_metadata,
- grpc_completion_queue *completion_queue, void *tag_new);
+ grpc_completion_queue *cq_bound_to_call,
+ void *tag_new);
+
+void *grpc_server_register_method(grpc_server *server, const char *method,
+ const char *host);
+
+grpc_call_error grpc_server_request_registered_call(
+ grpc_server *server, void *registered_method, grpc_call **call,
+ gpr_timespec *deadline, grpc_metadata_array *request_metadata,
+ grpc_byte_buffer **optional_payload,
+ grpc_completion_queue *cq_bound_to_call, void *tag_new);
/* Create a server */
grpc_server *grpc_server_create(grpc_completion_queue *cq,
diff --git a/src/core/support/cpu.h b/include/grpc/support/cpu.h
similarity index 95%
rename from src/core/support/cpu.h
rename to include/grpc/support/cpu.h
index f8ec2c6..9025f7c 100644
--- a/src/core/support/cpu.h
+++ b/include/grpc/support/cpu.h
@@ -34,6 +34,10 @@
#ifndef __GRPC_INTERNAL_SUPPORT_CPU_H__
#define __GRPC_INTERNAL_SUPPORT_CPU_H__
+#ifdef __cplusplus
+extern "C" {
+#endif
+
/* Interface providing CPU information for currently running system */
/* Return the number of CPU cores on the current system. Will return 0 if
@@ -46,4 +50,8 @@
[0, gpr_cpu_num_cores() - 1] */
unsigned gpr_cpu_current_cpu(void);
+#ifdef __cplusplus
+} // extern "C"
+#endif
+
#endif /* __GRPC_INTERNAL_SUPPORT_CPU_H__ */
diff --git a/src/compiler/cpp_generator.cc b/src/compiler/cpp_generator.cc
index 8724f97..d04c2ef 100644
--- a/src/compiler/cpp_generator.cc
+++ b/src/compiler/cpp_generator.cc
@@ -61,6 +61,17 @@
return method->client_streaming() && method->server_streaming();
}
+bool HasUnaryCalls(const google::protobuf::FileDescriptor *file) {
+ for (int i = 0; i < file->service_count(); i++) {
+ for (int j = 0; j < file->service(i)->method_count(); j++) {
+ if (NoStreaming(file->service(i)->method(j))) {
+ return true;
+ }
+ }
+ }
+ return false;
+}
+
bool HasClientOnlyStreaming(const google::protobuf::FileDescriptor *file) {
for (int i = 0; i < file->service_count(); i++) {
for (int j = 0; j < file->service(i)->method_count(); j++) {
@@ -97,20 +108,29 @@
std::string GetHeaderIncludes(const google::protobuf::FileDescriptor *file) {
std::string temp =
- "#include \"grpc++/impl/internal_stub.h\"\n"
- "#include \"grpc++/status.h\"\n"
+ "#include <grpc++/impl/internal_stub.h>\n"
+ "#include <grpc++/impl/service_type.h>\n"
+ "#include <grpc++/status.h>\n"
"\n"
"namespace grpc {\n"
"class ChannelInterface;\n"
"class RpcService;\n"
"class ServerContext;\n";
+ if (HasUnaryCalls(file)) {
+ temp.append(
+ "template <class OutMessage> class ServerAsyncResponseWriter;\n");
+ }
if (HasClientOnlyStreaming(file)) {
temp.append("template <class OutMessage> class ClientWriter;\n");
temp.append("template <class InMessage> class ServerReader;\n");
+ temp.append("template <class OutMessage> class ClientAsyncWriter;\n");
+ temp.append("template <class InMessage> class ServerAsyncReader;\n");
}
if (HasServerOnlyStreaming(file)) {
temp.append("template <class InMessage> class ClientReader;\n");
temp.append("template <class OutMessage> class ServerWriter;\n");
+ temp.append("template <class OutMessage> class ClientAsyncReader;\n");
+ temp.append("template <class InMessage> class ServerAsyncWriter;\n");
}
if (HasBidiStreaming(file)) {
temp.append(
@@ -119,16 +139,24 @@
temp.append(
"template <class OutMessage, class InMessage>\n"
"class ServerReaderWriter;\n");
+ temp.append(
+ "template <class OutMessage, class InMessage>\n"
+ "class ClientAsyncReaderWriter;\n");
+ temp.append(
+ "template <class OutMessage, class InMessage>\n"
+ "class ServerAsyncReaderWriter;\n");
}
temp.append("} // namespace grpc\n");
return temp;
}
std::string GetSourceIncludes() {
- return "#include \"grpc++/channel_interface.h\"\n"
- "#include \"grpc++/impl/rpc_method.h\"\n"
- "#include \"grpc++/impl/rpc_service_method.h\"\n"
- "#include \"grpc++/stream.h\"\n";
+ return "#include <grpc++/channel_interface.h>\n"
+ "#include <grpc++/impl/client_unary_call.h>\n"
+ "#include <grpc++/impl/rpc_method.h>\n"
+ "#include <grpc++/impl/rpc_service_method.h>\n"
+ "#include <grpc++/impl/service_type.h>\n"
+ "#include <grpc++/stream.h>\n";
}
void PrintHeaderClientMethod(google::protobuf::io::Printer *printer,
@@ -142,27 +170,45 @@
if (NoStreaming(method)) {
printer->Print(*vars,
"::grpc::Status $Method$(::grpc::ClientContext* context, "
- "const $Request$& request, $Response$* response);\n\n");
+ "const $Request$& request, $Response$* response);\n");
+ printer->Print(*vars,
+ "void $Method$(::grpc::ClientContext* context, "
+ "const $Request$& request, $Response$* response, "
+ "::grpc::Status *status, "
+ "::grpc::CompletionQueue *cq, void *tag);\n");
} else if (ClientOnlyStreaming(method)) {
- printer->Print(
- *vars,
- "::grpc::ClientWriter< $Request$>* $Method$("
- "::grpc::ClientContext* context, $Response$* response);\n\n");
+ printer->Print(*vars,
+ "::grpc::ClientWriter< $Request$>* $Method$("
+ "::grpc::ClientContext* context, $Response$* response);\n");
+ printer->Print(*vars,
+ "::grpc::ClientWriter< $Request$>* $Method$("
+ "::grpc::ClientContext* context, $Response$* response, "
+ "::grpc::Status *status, "
+ "::grpc::CompletionQueue *cq, void *tag);\n");
} else if (ServerOnlyStreaming(method)) {
printer->Print(
*vars,
"::grpc::ClientReader< $Response$>* $Method$("
- "::grpc::ClientContext* context, const $Request$* request);\n\n");
+ "::grpc::ClientContext* context, const $Request$* request);\n");
+ printer->Print(*vars,
+ "::grpc::ClientReader< $Response$>* $Method$("
+ "::grpc::ClientContext* context, const $Request$* request, "
+ "::grpc::CompletionQueue *cq, void *tag);\n");
} else if (BidiStreaming(method)) {
printer->Print(*vars,
"::grpc::ClientReaderWriter< $Request$, $Response$>* "
- "$Method$(::grpc::ClientContext* context);\n\n");
+ "$Method$(::grpc::ClientContext* context);\n");
+ printer->Print(*vars,
+ "::grpc::ClientAsyncReaderWriter< $Request$, $Response$>* "
+ "$Method$(::grpc::ClientContext* context, "
+ "::grpc::CompletionQueue *cq, void *tag);\n");
}
}
-void PrintHeaderServerMethod(google::protobuf::io::Printer *printer,
- const google::protobuf::MethodDescriptor *method,
- std::map<std::string, std::string> *vars) {
+void PrintHeaderServerMethodSync(
+ google::protobuf::io::Printer *printer,
+ const google::protobuf::MethodDescriptor *method,
+ std::map<std::string, std::string> *vars) {
(*vars)["Method"] = method->name();
(*vars)["Request"] =
grpc_cpp_generator::ClassName(method->input_type(), true);
@@ -194,19 +240,56 @@
}
}
+void PrintHeaderServerMethodAsync(
+ google::protobuf::io::Printer *printer,
+ const google::protobuf::MethodDescriptor *method,
+ std::map<std::string, std::string> *vars) {
+ (*vars)["Method"] = method->name();
+ (*vars)["Request"] =
+ grpc_cpp_generator::ClassName(method->input_type(), true);
+ (*vars)["Response"] =
+ grpc_cpp_generator::ClassName(method->output_type(), true);
+ if (NoStreaming(method)) {
+ printer->Print(*vars,
+ "void $Method$("
+ "::grpc::ServerContext* context, $Request$* request, "
+ "::grpc::ServerAsyncResponseWriter< $Response$>* response, "
+ "::grpc::CompletionQueue* cq, void *tag);\n");
+ } else if (ClientOnlyStreaming(method)) {
+ printer->Print(*vars,
+ "void $Method$("
+ "::grpc::ServerContext* context, "
+ "::grpc::ServerAsyncReader< $Request$>* reader, "
+ "::grpc::CompletionQueue* cq, void *tag);\n");
+ } else if (ServerOnlyStreaming(method)) {
+ printer->Print(*vars,
+ "void $Method$("
+ "::grpc::ServerContext* context, $Request$* request, "
+ "::grpc::ServerAsyncWriter< $Response$>* writer, "
+ "::grpc::CompletionQueue* cq, void *tag);\n");
+ } else if (BidiStreaming(method)) {
+ printer->Print(
+ *vars,
+ "void $Method$("
+ "::grpc::ServerContext* context, "
+ "::grpc::ServerAsyncReaderWriter< $Response$, $Request$>* stream, "
+ "::grpc::CompletionQueue* cq, void *tag);\n");
+ }
+}
+
void PrintHeaderService(google::protobuf::io::Printer *printer,
const google::protobuf::ServiceDescriptor *service,
std::map<std::string, std::string> *vars) {
(*vars)["Service"] = service->name();
printer->Print(*vars,
- "class $Service$ {\n"
+ "class $Service$ final {\n"
" public:\n");
printer->Indent();
// Client side
printer->Print(
- "class Stub : public ::grpc::InternalStub {\n"
+ "class Stub final : public ::grpc::InternalStub {\n"
" public:\n");
printer->Indent();
for (int i = 0; i < service->method_count(); ++i) {
@@ -220,17 +303,34 @@
printer->Print("\n");
- // Server side
+ // Server side - Synchronous
printer->Print(
- "class Service {\n"
+ "class Service : public ::grpc::SynchronousService {\n"
" public:\n");
printer->Indent();
printer->Print("Service() : service_(nullptr) {}\n");
printer->Print("virtual ~Service();\n");
for (int i = 0; i < service->method_count(); ++i) {
- PrintHeaderServerMethod(printer, service->method(i), vars);
+ PrintHeaderServerMethodSync(printer, service->method(i), vars);
}
- printer->Print("::grpc::RpcService* service();\n");
+ printer->Print("::grpc::RpcService* service() override final;\n");
+ printer->Outdent();
+ printer->Print(
+ " private:\n"
+ " ::grpc::RpcService* service_;\n");
+ printer->Print("};\n");
+
+ // Server side - Asynchronous
+ printer->Print(
+ "class AsyncService final : public ::grpc::AsynchronousService {\n"
+ " public:\n");
+ printer->Indent();
+ printer->Print("AsyncService() : service_(nullptr) {}\n");
+ printer->Print("~AsyncService();\n");
+ for (int i = 0; i < service->method_count(); ++i) {
+ PrintHeaderServerMethodAsync(printer, service->method(i), vars);
+ }
+ printer->Print("::grpc::RpcService* service() override;\n");
printer->Outdent();
printer->Print(
" private:\n"
@@ -268,7 +368,7 @@
"::grpc::ClientContext* context, "
"const $Request$& request, $Response$* response) {\n");
printer->Print(*vars,
- " return channel()->StartBlockingRpc("
+ "return ::grpc::BlockingUnaryCall(channel(),"
"::grpc::RpcMethod(\"/$Package$$Service$/$Method$\"), "
"context, request, response);\n"
"}\n\n");
@@ -279,10 +379,10 @@
"::grpc::ClientContext* context, $Response$* response) {\n");
printer->Print(*vars,
" return new ::grpc::ClientWriter< $Request$>("
- "channel()->CreateStream("
+ "channel(),"
"::grpc::RpcMethod(\"/$Package$$Service$/$Method$\", "
"::grpc::RpcMethod::RpcType::CLIENT_STREAMING), "
- "context, nullptr, response));\n"
+ "context, response);\n"
"}\n\n");
} else if (ServerOnlyStreaming(method)) {
printer->Print(
@@ -291,10 +391,10 @@
"::grpc::ClientContext* context, const $Request$* request) {\n");
printer->Print(*vars,
" return new ::grpc::ClientReader< $Response$>("
- "channel()->CreateStream("
+ "channel(),"
"::grpc::RpcMethod(\"/$Package$$Service$/$Method$\", "
"::grpc::RpcMethod::RpcType::SERVER_STREAMING), "
- "context, request, nullptr));\n"
+ "context, *request);\n"
"}\n\n");
} else if (BidiStreaming(method)) {
printer->Print(
@@ -304,10 +404,10 @@
printer->Print(
*vars,
" return new ::grpc::ClientReaderWriter< $Request$, $Response$>("
- "channel()->CreateStream("
+ "channel(),"
"::grpc::RpcMethod(\"/$Package$$Service$/$Method$\", "
"::grpc::RpcMethod::RpcType::BIDI_STREAMING), "
- "context, nullptr, nullptr));\n"
+ "context);\n"
"}\n\n");
}
}
@@ -362,6 +462,47 @@
}
}
+void PrintSourceServerAsyncMethod(google::protobuf::io::Printer *printer,
+ const google::protobuf::MethodDescriptor *method,
+ std::map<std::string, std::string> *vars) {
+ (*vars)["Method"] = method->name();
+ (*vars)["Request"] =
+ grpc_cpp_generator::ClassName(method->input_type(), true);
+ (*vars)["Response"] =
+ grpc_cpp_generator::ClassName(method->output_type(), true);
+ if (NoStreaming(method)) {
+ printer->Print(*vars,
+ "void $Service$::AsyncService::$Method$("
+ "::grpc::ServerContext* context, "
+ "$Request$* request, "
+ "::grpc::ServerAsyncResponseWriter< $Response$>* response, "
+ "::grpc::CompletionQueue* cq, void* tag) {\n");
+ printer->Print("}\n\n");
+ } else if (ClientOnlyStreaming(method)) {
+ printer->Print(*vars,
+ "void $Service$::AsyncService::$Method$("
+ "::grpc::ServerContext* context, "
+ "::grpc::ServerAsyncReader< $Request$>* reader, "
+ "::grpc::CompletionQueue* cq, void* tag) {\n");
+ printer->Print("}\n\n");
+ } else if (ServerOnlyStreaming(method)) {
+ printer->Print(*vars,
+ "void $Service$::AsyncService::$Method$("
+ "::grpc::ServerContext* context, "
+ "$Request$* request, "
+ "::grpc::ServerAsyncWriter< $Response$>* writer, "
+ "::grpc::CompletionQueue* cq, void* tag) {\n");
+ printer->Print("}\n\n");
+ } else if (BidiStreaming(method)) {
+ printer->Print(*vars,
+ "void $Service$::AsyncService::$Method$("
+ "::grpc::ServerContext* context, "
+ "::grpc::ServerAsyncReaderWriter< $Response$, $Request$>* stream, "
+ "::grpc::CompletionQueue* cq, void *tag) {\n");
+ printer->Print("}\n\n");
+ }
+}
+
void PrintSourceService(google::protobuf::io::Printer *printer,
const google::protobuf::ServiceDescriptor *service,
std::map<std::string, std::string> *vars) {
@@ -384,6 +525,7 @@
"}\n\n");
for (int i = 0; i < service->method_count(); ++i) {
PrintSourceServerMethod(printer, service->method(i), vars);
+ PrintSourceServerAsyncMethod(printer, service->method(i), vars);
}
printer->Print(*vars,
"::grpc::RpcService* $Service$::Service::service() {\n");
diff --git a/src/core/statistics/census_log.c b/src/core/statistics/census_log.c
index aea0a33..1504c02 100644
--- a/src/core/statistics/census_log.c
+++ b/src/core/statistics/census_log.c
@@ -91,9 +91,9 @@
*/
#include "src/core/statistics/census_log.h"
#include <string.h>
-#include "src/core/support/cpu.h"
#include <grpc/support/alloc.h>
#include <grpc/support/atm.h>
+#include <grpc/support/cpu.h>
#include <grpc/support/log.h>
#include <grpc/support/port_platform.h>
#include <grpc/support/sync.h>
diff --git a/src/core/support/cpu_linux.c b/src/core/support/cpu_linux.c
index ad82174..c8375e6 100644
--- a/src/core/support/cpu_linux.c
+++ b/src/core/support/cpu_linux.c
@@ -39,7 +39,7 @@
#ifdef GPR_CPU_LINUX
-#include "src/core/support/cpu.h"
+#include <grpc/support/cpu.h>
#include <sched.h>
#include <errno.h>
diff --git a/src/core/surface/call.c b/src/core/surface/call.c
index 0af2154..b31f4f1 100644
--- a/src/core/surface/call.c
+++ b/src/core/surface/call.c
@@ -258,6 +258,10 @@
call->cq = cq;
}
+grpc_completion_queue *grpc_call_get_completion_queue(grpc_call *call) {
+ return call->cq;
+}
+
void grpc_call_internal_ref(grpc_call *c) { gpr_ref(&c->internal_refcount); }
static void destroy_call(void *call, int ignored_success) {
diff --git a/src/core/surface/call.h b/src/core/surface/call.h
index 05014c6..55e4344 100644
--- a/src/core/surface/call.h
+++ b/src/core/surface/call.h
@@ -89,6 +89,7 @@
const void *server_transport_data);
void grpc_call_set_completion_queue(grpc_call *call, grpc_completion_queue *cq);
+grpc_completion_queue *grpc_call_get_completion_queue(grpc_call *call);
void grpc_call_internal_ref(grpc_call *call);
void grpc_call_internal_unref(grpc_call *call, int allow_immediate_deletion);
diff --git a/src/core/surface/lame_client.c b/src/core/surface/lame_client.c
index 411dbab..a8fdeed 100644
--- a/src/core/surface/lame_client.c
+++ b/src/core/surface/lame_client.c
@@ -47,6 +47,7 @@
} call_data;
typedef struct {
+ grpc_mdelem *status;
grpc_mdelem *message;
} channel_data;
@@ -57,6 +58,7 @@
switch (op->type) {
case GRPC_SEND_START:
+ grpc_call_recv_metadata(elem, grpc_mdelem_ref(channeld->status));
grpc_call_recv_metadata(elem, grpc_mdelem_ref(channeld->message));
grpc_call_stream_closed(elem);
break;
@@ -93,18 +95,22 @@
const grpc_channel_args *args, grpc_mdctx *mdctx,
int is_first, int is_last) {
channel_data *channeld = elem->channel_data;
+ char status[12];
GPR_ASSERT(is_first);
GPR_ASSERT(is_last);
channeld->message = grpc_mdelem_from_strings(mdctx, "grpc-message",
"Rpc sent on a lame channel.");
+ gpr_ltoa(GRPC_STATUS_UNKNOWN, status);
+ channeld->status = grpc_mdelem_from_strings(mdctx, "grpc-status", status);
}
static void destroy_channel_elem(grpc_channel_element *elem) {
channel_data *channeld = elem->channel_data;
grpc_mdelem_unref(channeld->message);
+ grpc_mdelem_unref(channeld->status);
}
static const grpc_channel_filter lame_filter = {
diff --git a/src/core/surface/server.c b/src/core/surface/server.c
index ee0f96a..80b248e 100644
--- a/src/core/surface/server.c
+++ b/src/core/surface/server.c
@@ -60,6 +60,55 @@
typedef struct call_data call_data;
typedef struct channel_data channel_data;
+typedef struct registered_method registered_method;
+
+typedef struct {
+ call_data *next;
+ call_data *prev;
+} call_link;
+
+typedef enum { LEGACY_CALL, BATCH_CALL, REGISTERED_CALL } requested_call_type;
+
+typedef struct {
+ requested_call_type type;
+ void *tag;
+ union {
+ struct {
+ grpc_completion_queue *cq_bind;
+ grpc_call **call;
+ grpc_call_details *details;
+ grpc_metadata_array *initial_metadata;
+ } batch;
+ struct {
+ grpc_completion_queue *cq_bind;
+ grpc_call **call;
+ registered_method *registered_method;
+ gpr_timespec *deadline;
+ grpc_metadata_array *initial_metadata;
+ grpc_byte_buffer **optional_payload;
+ } registered;
+ } data;
+} requested_call;
+
+typedef struct {
+ requested_call *calls;
+ size_t count;
+ size_t capacity;
+} requested_call_array;
+
+struct registered_method {
+ char *method;
+ char *host;
+ call_data *pending;
+ requested_call_array requested;
+ registered_method *next;
+};
+
+typedef struct channel_registered_method {
+ registered_method *server_registered_method;
+ grpc_mdstr *method;
+ grpc_mdstr *host;
+} channel_registered_method;
struct channel_data {
grpc_server *server;
@@ -69,22 +118,11 @@
/* linked list of all channels on a server */
channel_data *next;
channel_data *prev;
+ channel_registered_method *registered_methods;
+ gpr_uint32 registered_method_slots;
+ gpr_uint32 registered_method_max_probes;
};
-typedef void (*new_call_cb)(grpc_server *server, grpc_completion_queue *cq,
- grpc_call **call, grpc_call_details *details,
- grpc_metadata_array *initial_metadata,
- call_data *calld, void *user_data);
-
-typedef struct {
- void *user_data;
- grpc_completion_queue *cq;
- grpc_call **call;
- grpc_call_details *details;
- grpc_metadata_array *initial_metadata;
- new_call_cb cb;
-} requested_call;
-
struct grpc_server {
size_t channel_filter_count;
const grpc_channel_filter **channel_filters;
@@ -93,9 +131,8 @@
gpr_mu mu;
- requested_call *requested_calls;
- size_t requested_call_count;
- size_t requested_call_capacity;
+ registered_method *registered_methods;
+ requested_call_array requested_calls;
gpr_uint8 shutdown;
gpr_uint8 have_shutdown_tag;
@@ -108,11 +145,6 @@
gpr_refcount internal_refcount;
};
-typedef struct {
- call_data *next;
- call_data *prev;
-} call_link;
-
typedef enum {
/* waiting for metadata */
NOT_STARTED,
@@ -125,7 +157,7 @@
} call_state;
typedef struct legacy_data {
- grpc_metadata_array *initial_metadata;
+ grpc_metadata_array initial_metadata;
} legacy_data;
struct call_data {
@@ -137,9 +169,8 @@
grpc_mdstr *host;
legacy_data *legacy;
- grpc_call_details *details;
- gpr_uint8 included[CALL_LIST_COUNT];
+ call_data **root[CALL_LIST_COUNT];
call_link links[CALL_LIST_COUNT];
};
@@ -148,30 +179,33 @@
static void do_nothing(void *unused, grpc_op_error ignored) {}
-static int call_list_join(grpc_server *server, call_data *call,
- call_list list) {
- if (call->included[list]) return 0;
- call->included[list] = 1;
- if (!server->lists[list]) {
- server->lists[list] = call;
+static void begin_call(grpc_server *server, call_data *calld,
+ requested_call *rc);
+static void fail_call(grpc_server *server, requested_call *rc);
+
+static int call_list_join(call_data **root, call_data *call, call_list list) {
+ GPR_ASSERT(!call->root[list]);
+ call->root[list] = root;
+ if (!*root) {
+ *root = call;
call->links[list].next = call->links[list].prev = call;
} else {
- call->links[list].next = server->lists[list];
- call->links[list].prev = server->lists[list]->links[list].prev;
+ call->links[list].next = *root;
+ call->links[list].prev = (*root)->links[list].prev;
call->links[list].next->links[list].prev =
call->links[list].prev->links[list].next = call;
}
return 1;
}
-static call_data *call_list_remove_head(grpc_server *server, call_list list) {
- call_data *out = server->lists[list];
+static call_data *call_list_remove_head(call_data **root, call_list list) {
+ call_data *out = *root;
if (out) {
- out->included[list] = 0;
+ out->root[list] = NULL;
if (out->links[list].next == out) {
- server->lists[list] = NULL;
+ *root = NULL;
} else {
- server->lists[list] = out->links[list].next;
+ *root = out->links[list].next;
out->links[list].next->links[list].prev = out->links[list].prev;
out->links[list].prev->links[list].next = out->links[list].next;
}
@@ -179,23 +213,39 @@
return out;
}
-static int call_list_remove(grpc_server *server, call_data *call,
- call_list list) {
- if (!call->included[list]) return 0;
- call->included[list] = 0;
- if (server->lists[list] == call) {
- server->lists[list] = call->links[list].next;
- if (server->lists[list] == call) {
- server->lists[list] = NULL;
+static int call_list_remove(call_data *call, call_list list) {
+ call_data **root = call->root[list];
+ if (root == NULL) return 0;
+ call->root[list] = NULL;
+ if (*root == call) {
+ *root = call->links[list].next;
+ if (*root == call) {
+ *root = NULL;
return 1;
}
}
- GPR_ASSERT(server->lists[list] != call);
+ GPR_ASSERT(*root != call);
call->links[list].next->links[list].prev = call->links[list].prev;
call->links[list].prev->links[list].next = call->links[list].next;
return 1;
}
+static void requested_call_array_destroy(requested_call_array *array) {
+ gpr_free(array->calls);
+}
+
+static requested_call *requested_call_array_add(requested_call_array *array) {
+ requested_call *rc;
+ if (array->count == array->capacity) {
+ array->capacity = GPR_MAX(array->capacity + 8, array->capacity * 2);
+ array->calls =
+ gpr_realloc(array->calls, sizeof(requested_call) * array->capacity);
+ }
+ rc = &array->calls[array->count++];
+ memset(rc, 0, sizeof(*rc));
+ return rc;
+}
+
static void server_ref(grpc_server *server) {
gpr_ref(&server->internal_refcount);
}
@@ -205,7 +255,7 @@
grpc_channel_args_destroy(server->channel_args);
gpr_mu_destroy(&server->mu);
gpr_free(server->channel_filters);
- gpr_free(server->requested_calls);
+ requested_call_array_destroy(&server->requested_calls);
gpr_free(server);
}
}
@@ -223,7 +273,6 @@
static void finish_destroy_channel(void *cd, int success) {
channel_data *chand = cd;
grpc_server *server = chand->server;
- /*gpr_log(GPR_INFO, "destroy channel %p", chand->channel);*/
grpc_channel_destroy(chand->channel);
server_unref(server);
}
@@ -236,23 +285,63 @@
grpc_iomgr_add_callback(finish_destroy_channel, chand);
}
+static void finish_start_new_rpc_and_unlock(grpc_server *server,
+ grpc_call_element *elem,
+ call_data **pending_root,
+ requested_call_array *array) {
+ requested_call rc;
+ call_data *calld = elem->call_data;
+ if (array->count == 0) {
+ calld->state = PENDING;
+ call_list_join(pending_root, calld, PENDING_START);
+ gpr_mu_unlock(&server->mu);
+ } else {
+ rc = array->calls[--array->count];
+ calld->state = ACTIVATED;
+ gpr_mu_unlock(&server->mu);
+ begin_call(server, calld, &rc);
+ }
+}
+
static void start_new_rpc(grpc_call_element *elem) {
channel_data *chand = elem->channel_data;
call_data *calld = elem->call_data;
grpc_server *server = chand->server;
+ gpr_uint32 i;
+ gpr_uint32 hash;
+ channel_registered_method *rm;
gpr_mu_lock(&server->mu);
- if (server->requested_call_count > 0) {
- requested_call rc = server->requested_calls[--server->requested_call_count];
- calld->state = ACTIVATED;
- gpr_mu_unlock(&server->mu);
- rc.cb(server, rc.cq, rc.call, rc.details, rc.initial_metadata, calld,
- rc.user_data);
- } else {
- calld->state = PENDING;
- call_list_join(server, calld, PENDING_START);
- gpr_mu_unlock(&server->mu);
+ if (chand->registered_methods && calld->path && calld->host) {
+ /* check for an exact match with host */
+ hash = GRPC_MDSTR_KV_HASH(calld->host->hash, calld->path->hash);
+ for (i = 0; i < chand->registered_method_max_probes; i++) {
+ rm = &chand->registered_methods[(hash + i) %
+ chand->registered_method_slots];
+ if (!rm) break;
+ if (rm->host != calld->host) continue;
+ if (rm->method != calld->path) continue;
+ finish_start_new_rpc_and_unlock(server, elem,
+ &rm->server_registered_method->pending,
+ &rm->server_registered_method->requested);
+ return;
+ }
+ /* check for a wildcard method definition (no host set) */
+ hash = GRPC_MDSTR_KV_HASH(0, calld->path->hash);
+ for (i = 0; i <= chand->registered_method_max_probes; i++) {
+ rm = &chand->registered_methods[(hash + i) %
+ chand->registered_method_slots];
+ if (!rm) break;
+ if (rm->host != NULL) continue;
+ if (rm->method != calld->path) continue;
+ finish_start_new_rpc_and_unlock(server, elem,
+ &rm->server_registered_method->pending,
+ &rm->server_registered_method->requested);
+ return;
+ }
}
+ finish_start_new_rpc_and_unlock(server, elem, &server->lists[PENDING_START],
+ &server->requested_calls);
}
static void kill_zombie(void *elem, int success) {
@@ -267,7 +356,7 @@
case ACTIVATED:
break;
case PENDING:
- call_list_remove(chand->server, calld, PENDING_START);
+ call_list_remove(calld, PENDING_START);
/* fallthrough intended */
case NOT_STARTED:
calld->state = ZOMBIED;
@@ -398,7 +487,7 @@
calld->call = grpc_call_from_top_element(elem);
gpr_mu_lock(&chand->server->mu);
- call_list_join(chand->server, calld, ALL_CALLS);
+ call_list_join(&chand->server->lists[ALL_CALLS], calld, ALL_CALLS);
gpr_mu_unlock(&chand->server->mu);
server_ref(chand->server);
@@ -411,7 +500,7 @@
gpr_mu_lock(&chand->server->mu);
for (i = 0; i < CALL_LIST_COUNT; i++) {
- call_list_remove(chand->server, elem->call_data, i);
+ call_list_remove(elem->call_data, i);
}
if (chand->server->shutdown && chand->server->have_shutdown_tag &&
chand->server->lists[ALL_CALLS] == NULL) {
@@ -427,8 +516,7 @@
}
if (calld->legacy) {
- gpr_free(calld->legacy->initial_metadata->metadata);
- gpr_free(calld->legacy->initial_metadata);
+ gpr_free(calld->legacy->initial_metadata.metadata);
gpr_free(calld->legacy);
}
@@ -447,6 +535,7 @@
chand->path_key = grpc_mdstr_from_string(metadata_context, ":path");
chand->authority_key = grpc_mdstr_from_string(metadata_context, ":authority");
chand->next = chand->prev = chand;
+ chand->registered_methods = NULL;
}
static void destroy_channel_elem(grpc_channel_element *elem) {
@@ -464,9 +553,8 @@
}
static const grpc_channel_filter server_surface_filter = {
- call_op, channel_op, sizeof(call_data),
- init_call_elem, destroy_call_elem, sizeof(channel_data),
- init_channel_elem, destroy_channel_elem, "server",
+ call_op, channel_op, sizeof(call_data), init_call_elem, destroy_call_elem,
+ sizeof(channel_data), init_channel_elem, destroy_channel_elem, "server",
};
grpc_server *grpc_server_create_from_filters(grpc_completion_queue *cq,
@@ -509,6 +597,36 @@
return server;
}
+static int streq(const char *a, const char *b) {
+ if (a == NULL && b == NULL) return 1;
+ if (a == NULL) return 0;
+ if (b == NULL) return 0;
+ return 0 == strcmp(a, b);
+}
+
+void *grpc_server_register_method(grpc_server *server, const char *method,
+ const char *host) {
+ registered_method *m;
+ if (!method) {
+ gpr_log(GPR_ERROR, "%s method string cannot be NULL", __FUNCTION__);
+ return NULL;
+ }
+ for (m = server->registered_methods; m; m = m->next) {
+ if (streq(m->method, method) && streq(m->host, host)) {
+ gpr_log(GPR_ERROR, "duplicate registration for %s@%s", method,
+ host ? host : "*");
+ return NULL;
+ }
+ }
+ m = gpr_malloc(sizeof(registered_method));
+ memset(m, 0, sizeof(*m));
+ m->method = gpr_strdup(method);
+ m->host = gpr_strdup(host);
+ m->next = server->registered_methods;
+ server->registered_methods = m;
+ return m;
+}
+
void grpc_server_start(grpc_server *server) {
listener *l;
@@ -525,8 +643,18 @@
grpc_channel_filter const **filters =
gpr_malloc(sizeof(grpc_channel_filter *) * num_filters);
size_t i;
+ size_t num_registered_methods;
+ size_t alloc;
+ registered_method *rm;
+ channel_registered_method *crm;
grpc_channel *channel;
channel_data *chand;
+ grpc_mdstr *host;
+ grpc_mdstr *method;
+ gpr_uint32 hash;
+ gpr_uint32 slots;
+ gpr_uint32 probes;
+ gpr_uint32 max_probes = 0;
for (i = 0; i < s->channel_filter_count; i++) {
filters[i] = s->channel_filters[i];
@@ -546,6 +674,35 @@
server_ref(s);
chand->channel = channel;
+ num_registered_methods = 0;
+ for (rm = s->registered_methods; rm; rm = rm->next) {
+ num_registered_methods++;
+ }
+ /* build a lookup table phrased in terms of mdstr's in this channels context
+ to quickly find registered methods */
+ if (num_registered_methods > 0) {
+ slots = 2 * num_registered_methods;
+ alloc = sizeof(channel_registered_method) * slots;
+ chand->registered_methods = gpr_malloc(alloc);
+ memset(chand->registered_methods, 0, alloc);
+ for (rm = s->registered_methods; rm; rm = rm->next) {
+ host = rm->host ? grpc_mdstr_from_string(mdctx, rm->host) : NULL;
+ method = grpc_mdstr_from_string(mdctx, rm->method);
+ hash = GRPC_MDSTR_KV_HASH(host ? host->hash : 0, method->hash);
+ for (probes = 0; chand->registered_methods[(hash + probes) % slots]
+ .server_registered_method != NULL;
+ probes++)
+ ;
+ if (probes > max_probes) max_probes = probes;
+ crm = &chand->registered_methods[(hash + probes) % slots];
+ crm->server_registered_method = rm;
+ crm->host = host;
+ crm->method = method;
+ }
+ chand->registered_method_slots = slots;
+ chand->registered_method_max_probes = max_probes;
+ }
+
gpr_mu_lock(&s->mu);
chand->next = &s->root_channel_data;
chand->prev = chand->next->prev;
@@ -558,17 +715,17 @@
grpc_channel_get_channel_stack(channel), transport);
}
-void shutdown_internal(grpc_server *server, gpr_uint8 have_shutdown_tag,
+static void shutdown_internal(grpc_server *server, gpr_uint8 have_shutdown_tag,
void *shutdown_tag) {
listener *l;
- requested_call *requested_calls;
- size_t requested_call_count;
+ requested_call_array requested_calls;
channel_data **channels;
channel_data *c;
size_t nchannels;
size_t i;
grpc_channel_op op;
grpc_channel_element *elem;
+ registered_method *rm;
/* lock, and gather up some stuff to do */
gpr_mu_lock(&server->mu);
@@ -591,10 +748,18 @@
i++;
}
+ /* collect all unregistered then registered calls */
requested_calls = server->requested_calls;
- requested_call_count = server->requested_call_count;
- server->requested_calls = NULL;
- server->requested_call_count = 0;
+ memset(&server->requested_calls, 0, sizeof(server->requested_calls));
+ for (rm = server->registered_methods; rm; rm = rm->next) {
+ if (requested_calls.count + rm->requested.count > requested_calls.capacity) {
+ requested_calls.capacity = GPR_MAX(requested_calls.count + rm->requested.count, 2 * requested_calls.capacity);
+ requested_calls.calls = gpr_realloc(requested_calls.calls, sizeof(*requested_calls.calls) * requested_calls.capacity);
+ }
+ memcpy(requested_calls.calls + requested_calls.count, rm->requested.calls, sizeof(*requested_calls.calls) * rm->requested.count);
+ requested_calls.count += rm->requested.count;
+ memset(&rm->requested, 0, sizeof(rm->requested));
+ }
server->shutdown = 1;
server->have_shutdown_tag = have_shutdown_tag;
@@ -623,13 +788,10 @@
gpr_free(channels);
/* terminate all the requested calls */
- for (i = 0; i < requested_call_count; i++) {
- requested_calls[i].cb(server, requested_calls[i].cq,
- requested_calls[i].call, requested_calls[i].details,
- requested_calls[i].initial_metadata, NULL,
- requested_calls[i].user_data);
+ for (i = 0; i < requested_calls.count; i++) {
+ fail_call(server, &requested_calls.calls[i]);
}
- gpr_free(requested_calls);
+ gpr_free(requested_calls.calls);
/* Shutdown listeners */
for (l = server->listeners; l; l = l->next) {
@@ -675,47 +837,92 @@
}
static grpc_call_error queue_call_request(grpc_server *server,
- grpc_completion_queue *cq,
- grpc_call **call,
- grpc_call_details *details,
- grpc_metadata_array *initial_metadata,
- new_call_cb cb, void *user_data) {
- call_data *calld;
- requested_call *rc;
+ requested_call *rc) {
+ call_data *calld = NULL;
+ requested_call_array *requested_calls = NULL;
gpr_mu_lock(&server->mu);
if (server->shutdown) {
gpr_mu_unlock(&server->mu);
- cb(server, cq, call, details, initial_metadata, NULL, user_data);
+ fail_call(server, rc);
return GRPC_CALL_OK;
}
- calld = call_list_remove_head(server, PENDING_START);
+ switch (rc->type) {
+ case LEGACY_CALL:
+ case BATCH_CALL:
+ calld =
+ call_list_remove_head(&server->lists[PENDING_START], PENDING_START);
+ requested_calls = &server->requested_calls;
+ break;
+ case REGISTERED_CALL:
+ calld = call_list_remove_head(
+ &rc->data.registered.registered_method->pending, PENDING_START);
+ requested_calls = &rc->data.registered.registered_method->requested;
+ break;
+ }
if (calld) {
GPR_ASSERT(calld->state == PENDING);
calld->state = ACTIVATED;
gpr_mu_unlock(&server->mu);
- cb(server, cq, call, details, initial_metadata, calld, user_data);
+ begin_call(server, calld, rc);
return GRPC_CALL_OK;
} else {
- if (server->requested_call_count == server->requested_call_capacity) {
- server->requested_call_capacity =
- GPR_MAX(server->requested_call_capacity + 8,
- server->requested_call_capacity * 2);
- server->requested_calls =
- gpr_realloc(server->requested_calls,
- sizeof(requested_call) * server->requested_call_capacity);
- }
- rc = &server->requested_calls[server->requested_call_count++];
- rc->cb = cb;
- rc->cq = cq;
- rc->call = call;
- rc->details = details;
- rc->user_data = user_data;
- rc->initial_metadata = initial_metadata;
+ *requested_call_array_add(requested_calls) = *rc;
gpr_mu_unlock(&server->mu);
return GRPC_CALL_OK;
}
}
+grpc_call_error grpc_server_request_call(grpc_server *server, grpc_call **call,
+ grpc_call_details *details,
+ grpc_metadata_array *initial_metadata,
+ grpc_completion_queue *cq_bind,
+ void *tag) {
+ requested_call rc;
+ grpc_cq_begin_op(server->cq, NULL, GRPC_OP_COMPLETE);
+ rc.type = BATCH_CALL;
+ rc.tag = tag;
+ rc.data.batch.cq_bind = cq_bind;
+ rc.data.batch.call = call;
+ rc.data.batch.details = details;
+ rc.data.batch.initial_metadata = initial_metadata;
+ return queue_call_request(server, &rc);
+}
+
+grpc_call_error grpc_server_request_registered_call(
+ grpc_server *server, void *registered_method, grpc_call **call,
+ gpr_timespec *deadline, grpc_metadata_array *initial_metadata,
+ grpc_byte_buffer **optional_payload, grpc_completion_queue *cq_bind,
+ void *tag) {
+ requested_call rc;
+ grpc_cq_begin_op(server->cq, NULL, GRPC_OP_COMPLETE);
+ rc.type = REGISTERED_CALL;
+ rc.tag = tag;
+ rc.data.registered.cq_bind = cq_bind;
+ rc.data.registered.call = call;
+ rc.data.registered.registered_method = registered_method;
+ rc.data.registered.deadline = deadline;
+ rc.data.registered.initial_metadata = initial_metadata;
+ rc.data.registered.optional_payload = optional_payload;
+ return queue_call_request(server, &rc);
+}
+
+grpc_call_error grpc_server_request_call_old(grpc_server *server,
+ void *tag_new) {
+ requested_call rc;
+ grpc_cq_begin_op(server->cq, NULL, GRPC_SERVER_RPC_NEW);
+ rc.type = LEGACY_CALL;
+ rc.tag = tag_new;
+ return queue_call_request(server, &rc);
+}
+
+static void publish_legacy(grpc_call *call, grpc_op_error status, void *tag);
+static void publish_registered_or_batch(grpc_call *call, grpc_op_error status,
+ void *tag);
+static void publish_was_not_set(grpc_call *call, grpc_op_error status,
+ void *tag) {
+ abort();
+}
+
static void cpstr(char **dest, size_t *capacity, grpc_mdstr *value) {
gpr_slice slice = value->slice;
size_t len = GPR_SLICE_LENGTH(slice);
@@ -727,57 +934,82 @@
memcpy(*dest, grpc_mdstr_as_c_string(value), len + 1);
}
-static void publish_request(grpc_call *call, grpc_op_error status, void *tag) {
- grpc_call_element *elem =
- grpc_call_stack_element(grpc_call_get_call_stack(call), 0);
- call_data *calld = elem->call_data;
- channel_data *chand = elem->channel_data;
- grpc_server *server = chand->server;
+static void begin_call(grpc_server *server, call_data *calld,
+ requested_call *rc) {
+ grpc_ioreq_completion_func publish = publish_was_not_set;
+ grpc_ioreq req[2];
+ grpc_ioreq *r = req;
- if (status == GRPC_OP_OK) {
- cpstr(&calld->details->host, &calld->details->host_capacity, calld->host);
- cpstr(&calld->details->method, &calld->details->method_capacity,
- calld->path);
- calld->details->deadline = calld->deadline;
- grpc_cq_end_op_complete(server->cq, tag, call, do_nothing, NULL,
- GRPC_OP_OK);
- } else {
- abort();
- }
-}
+ /* called once initial metadata has been read by the call, but BEFORE
+ the ioreq to fetch it out of the call has been executed.
+ This means metadata related fields can be relied on in calld, but to
+ fill in the metadata array passed by the client, we need to perform
+ an ioreq op, that should complete immediately. */
-static void begin_request(grpc_server *server, grpc_completion_queue *cq,
- grpc_call **call, grpc_call_details *details,
- grpc_metadata_array *initial_metadata,
- call_data *calld, void *tag) {
- grpc_ioreq req;
- if (!calld) {
- *call = NULL;
- initial_metadata->count = 0;
- grpc_cq_end_op_complete(cq, tag, NULL, do_nothing, NULL, GRPC_OP_ERROR);
- return;
+ switch (rc->type) {
+ case LEGACY_CALL:
+ calld->legacy = gpr_malloc(sizeof(legacy_data));
+ memset(calld->legacy, 0, sizeof(legacy_data));
+ r->op = GRPC_IOREQ_RECV_INITIAL_METADATA;
+ r->data.recv_metadata = &calld->legacy->initial_metadata;
+ r++;
+ publish = publish_legacy;
+ break;
+ case BATCH_CALL:
+ cpstr(&rc->data.batch.details->host,
+ &rc->data.batch.details->host_capacity, calld->host);
+ cpstr(&rc->data.batch.details->method,
+ &rc->data.batch.details->method_capacity, calld->path);
+ grpc_call_set_completion_queue(calld->call, rc->data.batch.cq_bind);
+ *rc->data.batch.call = calld->call;
+ r->op = GRPC_IOREQ_RECV_INITIAL_METADATA;
+ r->data.recv_metadata = rc->data.batch.initial_metadata;
+ r++;
+ publish = publish_registered_or_batch;
+ break;
+ case REGISTERED_CALL:
+ *rc->data.registered.deadline = calld->deadline;
+ grpc_call_set_completion_queue(calld->call, rc->data.registered.cq_bind);
+ *rc->data.registered.call = calld->call;
+ r->op = GRPC_IOREQ_RECV_INITIAL_METADATA;
+ r->data.recv_metadata = rc->data.registered.initial_metadata;
+ r++;
+ if (rc->data.registered.optional_payload) {
+ r->op = GRPC_IOREQ_RECV_MESSAGE;
+ r->data.recv_message = rc->data.registered.optional_payload;
+ r++;
+ }
+ publish = publish_registered_or_batch;
+ break;
}
- calld->details = details;
- grpc_call_set_completion_queue(calld->call, cq);
- *call = calld->call;
- req.op = GRPC_IOREQ_RECV_INITIAL_METADATA;
- req.data.recv_metadata = initial_metadata;
+
grpc_call_internal_ref(calld->call);
- grpc_call_start_ioreq_and_call_back(calld->call, &req, 1, publish_request,
- tag);
+ grpc_call_start_ioreq_and_call_back(calld->call, req, r - req, publish,
+ rc->tag);
}
-grpc_call_error grpc_server_request_call(grpc_server *server, grpc_call **call,
- grpc_call_details *details,
- grpc_metadata_array *initial_metadata,
- grpc_completion_queue *cq, void *tag) {
- grpc_cq_begin_op(cq, NULL, GRPC_OP_COMPLETE);
- return queue_call_request(server, cq, call, details, initial_metadata,
- begin_request, tag);
+static void fail_call(grpc_server *server, requested_call *rc) {
+ switch (rc->type) {
+ case LEGACY_CALL:
+ grpc_cq_end_new_rpc(server->cq, rc->tag, NULL, do_nothing, NULL, NULL,
+ NULL, gpr_inf_past, 0, NULL);
+ break;
+ case BATCH_CALL:
+ *rc->data.batch.call = NULL;
+ rc->data.batch.initial_metadata->count = 0;
+ grpc_cq_end_op_complete(server->cq, rc->tag, NULL, do_nothing, NULL,
+ GRPC_OP_ERROR);
+ break;
+ case REGISTERED_CALL:
+ *rc->data.registered.call = NULL;
+ rc->data.registered.initial_metadata->count = 0;
+ grpc_cq_end_op_complete(server->cq, rc->tag, NULL, do_nothing, NULL,
+ GRPC_OP_ERROR);
+ break;
+ }
}
-static void publish_legacy_request(grpc_call *call, grpc_op_error status,
- void *tag) {
+static void publish_legacy(grpc_call *call, grpc_op_error status, void *tag) {
grpc_call_element *elem =
grpc_call_stack_element(grpc_call_get_call_stack(call), 0);
call_data *calld = elem->call_data;
@@ -788,44 +1020,21 @@
grpc_cq_end_new_rpc(server->cq, tag, call, do_nothing, NULL,
grpc_mdstr_as_c_string(calld->path),
grpc_mdstr_as_c_string(calld->host), calld->deadline,
- calld->legacy->initial_metadata->count,
- calld->legacy->initial_metadata->metadata);
+ calld->legacy->initial_metadata.count,
+ calld->legacy->initial_metadata.metadata);
} else {
+ gpr_log(GPR_ERROR, "should never reach here");
abort();
}
}
-static void begin_legacy_request(grpc_server *server, grpc_completion_queue *cq,
- grpc_call **call, grpc_call_details *details,
- grpc_metadata_array *initial_metadata,
- call_data *calld, void *tag) {
- grpc_ioreq req;
- GPR_ASSERT(call == NULL);
- GPR_ASSERT(details == NULL);
- if (!calld) {
- gpr_free(initial_metadata);
- grpc_cq_end_new_rpc(cq, tag, NULL, do_nothing, NULL, NULL, NULL,
- gpr_inf_past, 0, NULL);
- return;
- }
- req.op = GRPC_IOREQ_RECV_INITIAL_METADATA;
- req.data.recv_metadata = initial_metadata;
- calld->legacy = gpr_malloc(sizeof(legacy_data));
- memset(calld->legacy, 0, sizeof(legacy_data));
- calld->legacy->initial_metadata = initial_metadata;
- grpc_call_internal_ref(calld->call);
- grpc_call_start_ioreq_and_call_back(calld->call, &req, 1,
- publish_legacy_request, tag);
-}
-
-grpc_call_error grpc_server_request_call_old(grpc_server *server,
- void *tag_new) {
- grpc_metadata_array *client_metadata =
- gpr_malloc(sizeof(grpc_metadata_array));
- memset(client_metadata, 0, sizeof(*client_metadata));
- grpc_cq_begin_op(server->cq, NULL, GRPC_SERVER_RPC_NEW);
- return queue_call_request(server, server->cq, NULL, NULL, client_metadata,
- begin_legacy_request, tag_new);
+static void publish_registered_or_batch(grpc_call *call, grpc_op_error status,
+ void *tag) {
+ grpc_call_element *elem =
+ grpc_call_stack_element(grpc_call_get_call_stack(call), 0);
+ channel_data *chand = elem->channel_data;
+ grpc_server *server = chand->server;
+ grpc_cq_end_op_complete(server->cq, tag, call, do_nothing, NULL, status);
}
const grpc_channel_args *grpc_server_get_channel_args(grpc_server *server) {
diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c
index ea579cf..37cdb9d 100644
--- a/src/core/transport/chttp2_transport.c
+++ b/src/core/transport/chttp2_transport.c
@@ -478,9 +478,6 @@
ref_transport(t);
gpr_mu_unlock(&t->mu);
- ref_transport(t);
- recv_data(t, slices, nslices, GRPC_ENDPOINT_CB_OK);
-
sr = setup(arg, &t->base, t->metadata_context);
lock(t);
@@ -488,6 +485,10 @@
t->cb_user_data = sr.user_data;
t->calling_back = 0;
unlock(t);
+
+ ref_transport(t);
+ recv_data(t, slices, nslices, GRPC_ENDPOINT_CB_OK);
+
unref_transport(t);
}
@@ -1014,6 +1015,8 @@
int had_outgoing;
char buffer[GPR_LTOA_MIN_BUFSIZE];
+ gpr_log(GPR_DEBUG, "cancel %d", id);
+
if (s) {
/* clear out any unreported input & output: nobody cares anymore */
had_outgoing = s->outgoing_sopb.nops != 0;
diff --git a/src/cpp/client/channel.cc b/src/cpp/client/channel.cc
index 3f39364..da94739 100644
--- a/src/cpp/client/channel.cc
+++ b/src/cpp/client/channel.cc
@@ -42,11 +42,12 @@
#include <grpc/support/slice.h>
#include "src/cpp/proto/proto_utils.h"
-#include "src/cpp/stream/stream_context.h"
#include <grpc++/channel_arguments.h>
#include <grpc++/client_context.h>
+#include <grpc++/completion_queue.h>
#include <grpc++/config.h>
#include <grpc++/credentials.h>
+#include <grpc++/impl/call.h>
#include <grpc++/impl/rpc_method.h>
#include <grpc++/status.h>
#include <google/protobuf/message.h>
@@ -77,103 +78,23 @@
Channel::~Channel() { grpc_channel_destroy(c_channel_); }
-namespace {
-// Pluck the finished event and set to status when it is not nullptr.
-void GetFinalStatus(grpc_completion_queue *cq, void *finished_tag,
- Status *status) {
- grpc_event *ev =
- grpc_completion_queue_pluck(cq, finished_tag, gpr_inf_future);
- if (status) {
- StatusCode error_code = static_cast<StatusCode>(ev->data.finished.status);
- grpc::string details(ev->data.finished.details ? ev->data.finished.details
- : "");
- *status = Status(error_code, details);
- }
- grpc_event_finish(ev);
-}
-} // namespace
-
-// TODO(yangg) more error handling
-Status Channel::StartBlockingRpc(const RpcMethod &method,
- ClientContext *context,
- const google::protobuf::Message &request,
- google::protobuf::Message *result) {
- Status status;
- grpc_call *call = grpc_channel_create_call_old(
- c_channel_, method.name(), target_.c_str(), context->RawDeadline());
- context->set_call(call);
-
- grpc_event *ev;
- void *finished_tag = reinterpret_cast<char *>(call);
- void *metadata_read_tag = reinterpret_cast<char *>(call) + 2;
- void *write_tag = reinterpret_cast<char *>(call) + 3;
- void *halfclose_tag = reinterpret_cast<char *>(call) + 4;
- void *read_tag = reinterpret_cast<char *>(call) + 5;
-
- grpc_completion_queue *cq = grpc_completion_queue_create();
- context->set_cq(cq);
- // add_metadata from context
- //
- // invoke
- GPR_ASSERT(grpc_call_invoke_old(call, cq, metadata_read_tag, finished_tag,
- GRPC_WRITE_BUFFER_HINT) == GRPC_CALL_OK);
- // write request
- grpc_byte_buffer *write_buffer = nullptr;
- bool success = SerializeProto(request, &write_buffer);
- if (!success) {
- grpc_call_cancel(call);
- status =
- Status(StatusCode::DATA_LOSS, "Failed to serialize request proto.");
- GetFinalStatus(cq, finished_tag, nullptr);
- return status;
- }
- GPR_ASSERT(grpc_call_start_write_old(call, write_buffer, write_tag,
- GRPC_WRITE_BUFFER_HINT) == GRPC_CALL_OK);
- grpc_byte_buffer_destroy(write_buffer);
- ev = grpc_completion_queue_pluck(cq, write_tag, gpr_inf_future);
-
- success = ev->data.write_accepted == GRPC_OP_OK;
- grpc_event_finish(ev);
- if (!success) {
- GetFinalStatus(cq, finished_tag, &status);
- return status;
- }
- // writes done
- GPR_ASSERT(grpc_call_writes_done_old(call, halfclose_tag) == GRPC_CALL_OK);
- ev = grpc_completion_queue_pluck(cq, halfclose_tag, gpr_inf_future);
- grpc_event_finish(ev);
- // start read metadata
- //
- ev = grpc_completion_queue_pluck(cq, metadata_read_tag, gpr_inf_future);
- grpc_event_finish(ev);
- // start read
- GPR_ASSERT(grpc_call_start_read_old(call, read_tag) == GRPC_CALL_OK);
- ev = grpc_completion_queue_pluck(cq, read_tag, gpr_inf_future);
- if (ev->data.read) {
- if (!DeserializeProto(ev->data.read, result)) {
- grpc_event_finish(ev);
- status = Status(StatusCode::DATA_LOSS, "Failed to parse response proto.");
- GetFinalStatus(cq, finished_tag, nullptr);
- return status;
- }
- }
- grpc_event_finish(ev);
-
- // wait status
- GetFinalStatus(cq, finished_tag, &status);
- return status;
+Call Channel::CreateCall(const RpcMethod &method, ClientContext *context,
+ CompletionQueue *cq) {
+ auto c_call =
+ grpc_channel_create_call(c_channel_, cq->cq(), method.name(),
+ target_.c_str(), context->RawDeadline());
+ context->set_call(c_call);
+ return Call(c_call, this, cq);
}
-StreamContextInterface *Channel::CreateStream(
- const RpcMethod &method, ClientContext *context,
- const google::protobuf::Message *request,
- google::protobuf::Message *result) {
- grpc_call *call = grpc_channel_create_call_old(
- c_channel_, method.name(), target_.c_str(), context->RawDeadline());
- context->set_call(call);
- grpc_completion_queue *cq = grpc_completion_queue_create();
- context->set_cq(cq);
- return new StreamContext(method, context, request, result);
+void Channel::PerformOpsOnCall(CallOpBuffer *buf, Call *call) {
+ static const size_t MAX_OPS = 8;
+ size_t nops = MAX_OPS;
+ grpc_op ops[MAX_OPS];
+ buf->FillOps(ops, &nops);
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_call_start_batch(call->call(), ops, nops,
+ buf));
}
} // namespace grpc
diff --git a/src/cpp/client/channel.h b/src/cpp/client/channel.h
index 67d18bf..894f876 100644
--- a/src/cpp/client/channel.h
+++ b/src/cpp/client/channel.h
@@ -42,11 +42,14 @@
struct grpc_channel;
namespace grpc {
+class Call;
+class CallOpBuffer;
class ChannelArguments;
+class CompletionQueue;
class Credentials;
class StreamContextInterface;
-class Channel : public ChannelInterface {
+class Channel final : public ChannelInterface {
public:
Channel(const grpc::string &target, const ChannelArguments &args);
Channel(const grpc::string &target, const std::unique_ptr<Credentials> &creds,
@@ -54,14 +57,10 @@
~Channel() override;
- Status StartBlockingRpc(const RpcMethod &method, ClientContext *context,
- const google::protobuf::Message &request,
- google::protobuf::Message *result) override;
-
- StreamContextInterface *CreateStream(
- const RpcMethod &method, ClientContext *context,
- const google::protobuf::Message *request,
- google::protobuf::Message *result) override;
+ virtual Call CreateCall(const RpcMethod &method, ClientContext *context,
+ CompletionQueue *cq) override;
+ virtual void PerformOpsOnCall(CallOpBuffer *ops,
+ Call *call) override;
private:
const grpc::string target_;
diff --git a/src/cpp/client/client_context.cc b/src/cpp/client/client_context.cc
index 7bda2d0..afacc81 100644
--- a/src/cpp/client/client_context.cc
+++ b/src/cpp/client/client_context.cc
@@ -72,7 +72,7 @@
void ClientContext::AddMetadata(const grpc::string &meta_key,
const grpc::string &meta_value) {
- return;
+ send_initial_metadata_.insert(std::make_pair(meta_key, meta_value));
}
void ClientContext::StartCancel() {}
diff --git a/src/cpp/server/server_rpc_handler.h b/src/cpp/client/client_unary_call.cc
similarity index 64%
rename from src/cpp/server/server_rpc_handler.h
rename to src/cpp/client/client_unary_call.cc
index a43e07d..bc0e837 100644
--- a/src/cpp/server/server_rpc_handler.h
+++ b/src/cpp/client/client_unary_call.cc
@@ -31,36 +31,33 @@
*
*/
-#ifndef __GRPCPP_INTERNAL_SERVER_SERVER_RPC_HANDLER_H__
-#define __GRPCPP_INTERNAL_SERVER_SERVER_RPC_HANDLER_H__
-
-#include <memory>
-
+#include <grpc++/impl/client_unary_call.h>
+#include <grpc++/impl/call.h>
+#include <grpc++/channel_interface.h>
#include <grpc++/completion_queue.h>
#include <grpc++/status.h>
+#include <grpc/support/log.h>
namespace grpc {
-class AsyncServerContext;
-class RpcServiceMethod;
-
-class ServerRpcHandler {
- public:
- // Takes ownership of async_server_context.
- ServerRpcHandler(AsyncServerContext *async_server_context,
- RpcServiceMethod *method);
-
- void StartRpc();
-
- private:
- CompletionQueue::CompletionType WaitForNextEvent();
- void FinishRpc(const Status &status);
-
- std::unique_ptr<AsyncServerContext> async_server_context_;
- RpcServiceMethod *method_;
- CompletionQueue cq_;
-};
+// Wrapper that performs a blocking unary call
+Status BlockingUnaryCall(ChannelInterface *channel, const RpcMethod &method,
+ ClientContext *context,
+ const google::protobuf::Message &request,
+ google::protobuf::Message *result) {
+ CompletionQueue cq;
+ Call call(channel->CreateCall(method, context, &cq));
+ CallOpBuffer buf;
+ Status status;
+ buf.AddSendInitialMetadata(context);
+ buf.AddSendMessage(request);
+ bool got_message;
+ buf.AddRecvMessage(result, &got_message);
+ buf.AddClientSendClose();
+ buf.AddClientRecvStatus(nullptr, &status); // TODO metadata
+ call.PerformOps(&buf);
+ GPR_ASSERT(cq.Pluck(&buf) && (got_message || !status.IsOk()));
+ return status;
+}
} // namespace grpc
-
-#endif // __GRPCPP_INTERNAL_SERVER_SERVER_RPC_HANDLER_H__
diff --git a/src/cpp/common/call.cc b/src/cpp/common/call.cc
new file mode 100644
index 0000000..a20d4a0
--- /dev/null
+++ b/src/cpp/common/call.cc
@@ -0,0 +1,274 @@
+/*
+ *
+ * Copyright 2014, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ * * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#include <grpc/support/alloc.h>
+#include <grpc++/impl/call.h>
+#include <grpc++/client_context.h>
+#include <grpc++/channel_interface.h>
+
+#include "src/cpp/proto/proto_utils.h"
+
+namespace grpc {
+
+void CallOpBuffer::Reset(void* next_return_tag) {
+ return_tag_ = next_return_tag;
+
+ send_initial_metadata_ = false;
+ initial_metadata_count_ = 0;
+ gpr_free(initial_metadata_);
+
+ recv_initial_metadata_ = nullptr;
+ gpr_free(recv_initial_metadata_arr_.metadata);
+ recv_initial_metadata_arr_ = {0, 0, nullptr};
+
+ send_message_ = nullptr;
+ if (send_message_buf_) {
+ grpc_byte_buffer_destroy(send_message_buf_);
+ send_message_buf_ = nullptr;
+ }
+
+ recv_message_ = nullptr;
+ got_message_ = nullptr;
+ if (recv_message_buf_) {
+ grpc_byte_buffer_destroy(recv_message_buf_);
+ recv_message_buf_ = nullptr;
+ }
+
+ client_send_close_ = false;
+
+ recv_trailing_metadata_ = nullptr;
+ recv_status_ = nullptr;
+ gpr_free(recv_trailing_metadata_arr_.metadata);
+ recv_trailing_metadata_arr_ = {0, 0, nullptr};
+
+ status_code_ = GRPC_STATUS_OK;
+ gpr_free(status_details_);
+ status_details_ = nullptr;
+ status_details_capacity_ = 0;
+
+ send_status_ = nullptr;
+ trailing_metadata_count_ = 0;
+ trailing_metadata_ = nullptr;
+
+ recv_closed_ = nullptr;
+}
+
+namespace {
+// TODO(yangg) if the map is changed before we send, the pointers will be a
+// mess. Make sure it does not happen.
+grpc_metadata* FillMetadataArray(
+ std::multimap<grpc::string, grpc::string>* metadata) {
+ if (metadata->empty()) { return nullptr; }
+ grpc_metadata* metadata_array = (grpc_metadata*)gpr_malloc(
+ metadata->size()* sizeof(grpc_metadata));
+ size_t i = 0;
+ for (auto iter = metadata->cbegin();
+ iter != metadata->cend();
+ ++iter, ++i) {
+ metadata_array[i].key = iter->first.c_str();
+ metadata_array[i].value = iter->second.c_str();
+ metadata_array[i].value_length = iter->second.size();
+ }
+ return metadata_array;
+}
+
+void FillMetadataMap(grpc_metadata_array* arr,
+ std::multimap<grpc::string, grpc::string>* metadata) {
+ for (size_t i = 0; i < arr->count; i++) {
+ // TODO(yangg) handle duplicates?
+ metadata->insert(std::pair<grpc::string, grpc::string>(
+ arr->metadata[i].key, {arr->metadata[i].value, arr->metadata[i].value_length}));
+ }
+ grpc_metadata_array_destroy(arr);
+ grpc_metadata_array_init(arr);
+}
+} // namespace
+
+void CallOpBuffer::AddSendInitialMetadata(
+ std::multimap<grpc::string, grpc::string>* metadata) {
+ send_initial_metadata_ = true;
+ initial_metadata_count_ = metadata->size();
+ initial_metadata_ = FillMetadataArray(metadata);
+}
+
+void CallOpBuffer::AddSendInitialMetadata(ClientContext *ctx) {
+ AddSendInitialMetadata(&ctx->send_initial_metadata_);
+}
+
+void CallOpBuffer::AddSendMessage(const google::protobuf::Message& message) {
+ send_message_ = &message;
+}
+
+void CallOpBuffer::AddRecvMessage(google::protobuf::Message *message, bool* got_message) {
+ recv_message_ = message;
+ got_message_ = got_message;
+}
+
+void CallOpBuffer::AddClientSendClose() {
+ client_send_close_ = true;
+}
+
+void CallOpBuffer::AddServerRecvClose(bool* cancelled) {
+ recv_closed_ = cancelled;
+}
+
+void CallOpBuffer::AddClientRecvStatus(
+ std::multimap<grpc::string, grpc::string>* metadata, Status *status) {
+ recv_trailing_metadata_ = metadata;
+ recv_status_ = status;
+}
+
+void CallOpBuffer::AddServerSendStatus(
+ std::multimap<grpc::string, grpc::string>* metadata, const Status& status) {
+ if (metadata != NULL) {
+ trailing_metadata_count_ = metadata->size();
+ trailing_metadata_ = FillMetadataArray(metadata);
+ } else {
+ trailing_metadata_count_ = 0;
+ }
+ send_status_ = &status;
+}
+
+void CallOpBuffer::FillOps(grpc_op *ops, size_t *nops) {
+ *nops = 0;
+ if (send_initial_metadata_) {
+ ops[*nops].op = GRPC_OP_SEND_INITIAL_METADATA;
+ ops[*nops].data.send_initial_metadata.count = initial_metadata_count_;
+ ops[*nops].data.send_initial_metadata.metadata = initial_metadata_;
+ (*nops)++;
+ }
+ if (recv_initial_metadata_) {
+ ops[*nops].op = GRPC_OP_RECV_INITIAL_METADATA;
+ ops[*nops].data.recv_initial_metadata = &recv_initial_metadata_arr_;
+ (*nops)++;
+ }
+ if (send_message_) {
+ bool success = SerializeProto(*send_message_, &send_message_buf_);
+ if (!success) {
+ abort();
+ // TODO handle parse failure
+ }
+ ops[*nops].op = GRPC_OP_SEND_MESSAGE;
+ ops[*nops].data.send_message = send_message_buf_;
+ (*nops)++;
+ }
+ if (recv_message_) {
+ ops[*nops].op = GRPC_OP_RECV_MESSAGE;
+ ops[*nops].data.recv_message = &recv_message_buf_;
+ (*nops)++;
+ }
+ if (client_send_close_) {
+ ops[*nops].op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+ (*nops)++;
+ }
+ if (recv_status_) {
+ ops[*nops].op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+ ops[*nops].data.recv_status_on_client.trailing_metadata =
+ &recv_trailing_metadata_arr_;
+ ops[*nops].data.recv_status_on_client.status = &status_code_;
+ ops[*nops].data.recv_status_on_client.status_details = &status_details_;
+ ops[*nops].data.recv_status_on_client.status_details_capacity =
+ &status_details_capacity_;
+ (*nops)++;
+ }
+ if (send_status_) {
+ ops[*nops].op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+ ops[*nops].data.send_status_from_server.trailing_metadata_count =
+ trailing_metadata_count_;
+ ops[*nops].data.send_status_from_server.trailing_metadata =
+ trailing_metadata_;
+ ops[*nops].data.send_status_from_server.status =
+ static_cast<grpc_status_code>(send_status_->code());
+ ops[*nops].data.send_status_from_server.status_details =
+ send_status_->details().c_str();
+ (*nops)++;
+ }
+ if (recv_closed_) {
+ ops[*nops].op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+ ops[*nops].data.recv_close_on_server.cancelled = &cancelled_buf_;
+ (*nops)++;
+ }
+}
+
+void CallOpBuffer::FinalizeResult(void **tag, bool *status) {
+ // Release send buffers.
+ if (send_message_buf_) {
+ grpc_byte_buffer_destroy(send_message_buf_);
+ send_message_buf_ = nullptr;
+ }
+ if (initial_metadata_) {
+ gpr_free(initial_metadata_);
+ initial_metadata_ = nullptr;
+ }
+ if (trailing_metadata_count_) {
+ gpr_free(trailing_metadata_);
+ trailing_metadata_ = nullptr;
+ }
+ // Set user-facing tag.
+ *tag = return_tag_;
+ // Process received initial metadata
+ if (recv_initial_metadata_) {
+ FillMetadataMap(&recv_initial_metadata_arr_, recv_initial_metadata_);
+ }
+ // Parse received message if any.
+ if (recv_message_) {
+ if (recv_message_buf_) {
+ *got_message_ = true;
+ *status = DeserializeProto(recv_message_buf_, recv_message_);
+ grpc_byte_buffer_destroy(recv_message_buf_);
+ recv_message_buf_ = nullptr;
+ } else {
+ *got_message_ = false;
+ }
+ }
+ // Parse received status.
+ if (recv_status_) {
+ FillMetadataMap(&recv_trailing_metadata_arr_, recv_trailing_metadata_);
+ *recv_status_ = Status(
+ static_cast<StatusCode>(status_code_),
+ status_details_ ? grpc::string(status_details_)
+ : grpc::string());
+ }
+ if (recv_closed_) {
+ *recv_closed_ = cancelled_buf_ != 0;
+ }
+}
+
+Call::Call(grpc_call* call, CallHook *call_hook, CompletionQueue* cq)
+ : call_hook_(call_hook), cq_(cq), call_(call) {}
+
+void Call::PerformOps(CallOpBuffer* buffer) {
+ call_hook_->PerformOpsOnCall(buffer, this);
+}
+
+} // namespace grpc
diff --git a/src/cpp/common/completion_queue.cc b/src/cpp/common/completion_queue.cc
index f06da9b..f69ddc3 100644
--- a/src/cpp/common/completion_queue.cc
+++ b/src/cpp/common/completion_queue.cc
@@ -1,5 +1,4 @@
/*
- *
* Copyright 2014, Google Inc.
* All rights reserved.
*
@@ -33,80 +32,52 @@
#include <grpc++/completion_queue.h>
+#include <memory>
+
#include <grpc/grpc.h>
#include <grpc/support/log.h>
#include <grpc/support/time.h>
#include "src/cpp/util/time.h"
-#include <grpc++/async_server_context.h>
namespace grpc {
CompletionQueue::CompletionQueue() { cq_ = grpc_completion_queue_create(); }
+CompletionQueue::CompletionQueue(grpc_completion_queue *take) : cq_(take) {}
+
CompletionQueue::~CompletionQueue() { grpc_completion_queue_destroy(cq_); }
void CompletionQueue::Shutdown() { grpc_completion_queue_shutdown(cq_); }
-CompletionQueue::CompletionType CompletionQueue::Next(void **tag) {
- grpc_event *ev;
- CompletionType return_type;
- bool success;
+// Helper class so we can declare a unique_ptr with grpc_event
+class EventDeleter {
+ public:
+ void operator()(grpc_event *ev) { if (ev) grpc_event_finish(ev); }
+};
- ev = grpc_completion_queue_next(cq_, gpr_inf_future);
- if (!ev) {
- gpr_log(GPR_ERROR, "no next event in queue");
- abort();
+bool CompletionQueue::Next(void **tag, bool *ok) {
+ std::unique_ptr<grpc_event, EventDeleter> ev;
+
+ ev.reset(grpc_completion_queue_next(cq_, gpr_inf_future));
+ if (ev->type == GRPC_QUEUE_SHUTDOWN) {
+ return false;
}
- switch (ev->type) {
- case GRPC_QUEUE_SHUTDOWN:
- return_type = QUEUE_CLOSED;
- break;
- case GRPC_READ:
- *tag = ev->tag;
- if (ev->data.read) {
- success = static_cast<AsyncServerContext *>(ev->tag)
- ->ParseRead(ev->data.read);
- return_type = success ? SERVER_READ_OK : SERVER_READ_ERROR;
- } else {
- return_type = SERVER_READ_ERROR;
- }
- break;
- case GRPC_WRITE_ACCEPTED:
- *tag = ev->tag;
- if (ev->data.write_accepted != GRPC_OP_ERROR) {
- return_type = SERVER_WRITE_OK;
- } else {
- return_type = SERVER_WRITE_ERROR;
- }
- break;
- case GRPC_SERVER_RPC_NEW:
- GPR_ASSERT(!ev->tag);
- // Finishing the pending new rpcs after the server has been shutdown.
- if (!ev->call) {
- *tag = nullptr;
- } else {
- *tag = new AsyncServerContext(
- ev->call, ev->data.server_rpc_new.method,
- ev->data.server_rpc_new.host,
- Timespec2Timepoint(ev->data.server_rpc_new.deadline));
- }
- return_type = SERVER_RPC_NEW;
- break;
- case GRPC_FINISHED:
- *tag = ev->tag;
- return_type = RPC_END;
- break;
- case GRPC_FINISH_ACCEPTED:
- *tag = ev->tag;
- return_type = HALFCLOSE_OK;
- break;
- default:
- // We do not handle client side messages now
- gpr_log(GPR_ERROR, "client-side messages aren't supported yet");
- abort();
- }
- grpc_event_finish(ev);
- return return_type;
+ auto cq_tag = static_cast<CompletionQueueTag *>(ev->tag);
+ *ok = ev->data.op_complete == GRPC_OP_OK;
+ *tag = cq_tag;
+ cq_tag->FinalizeResult(tag, ok);
+ return true;
+}
+
+bool CompletionQueue::Pluck(CompletionQueueTag *tag) {
+ std::unique_ptr<grpc_event, EventDeleter> ev;
+
+ ev.reset(grpc_completion_queue_pluck(cq_, tag, gpr_inf_future));
+ bool ok = ev->data.op_complete == GRPC_OP_OK;
+ void *ignored = tag;
+ tag->FinalizeResult(&ignored, &ok);
+ GPR_ASSERT(ignored == tag);
+ return ok;
}
} // namespace grpc
diff --git a/src/cpp/server/async_server.cc b/src/cpp/server/async_server.cc
deleted file mode 100644
index 86faa07..0000000
--- a/src/cpp/server/async_server.cc
+++ /dev/null
@@ -1,89 +0,0 @@
-/*
- *
- * Copyright 2014, Google Inc.
- * All rights reserved.
- *
- * Redistribution and use in source and binary forms, with or without
- * modification, are permitted provided that the following conditions are
- * met:
- *
- * * Redistributions of source code must retain the above copyright
- * notice, this list of conditions and the following disclaimer.
- * * Redistributions in binary form must reproduce the above
- * copyright notice, this list of conditions and the following disclaimer
- * in the documentation and/or other materials provided with the
- * distribution.
- * * Neither the name of Google Inc. nor the names of its
- * contributors may be used to endorse or promote products derived from
- * this software without specific prior written permission.
- *
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
- * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
- * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
- * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
- * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
- * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
- * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
- * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
- * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
- * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
- * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
- *
- */
-
-#include <grpc++/async_server.h>
-
-#include <grpc/grpc.h>
-#include <grpc/support/log.h>
-#include <grpc++/completion_queue.h>
-
-namespace grpc {
-
-AsyncServer::AsyncServer(CompletionQueue *cc)
- : started_(false), shutdown_(false) {
- server_ = grpc_server_create(cc->cq(), nullptr);
-}
-
-AsyncServer::~AsyncServer() {
- std::unique_lock<std::mutex> lock(shutdown_mu_);
- if (started_ && !shutdown_) {
- lock.unlock();
- Shutdown();
- }
- grpc_server_destroy(server_);
-}
-
-void AsyncServer::AddPort(const grpc::string &addr) {
- GPR_ASSERT(!started_);
- int success = grpc_server_add_http2_port(server_, addr.c_str());
- GPR_ASSERT(success);
-}
-
-void AsyncServer::Start() {
- GPR_ASSERT(!started_);
- started_ = true;
- grpc_server_start(server_);
-}
-
-void AsyncServer::RequestOneRpc() {
- GPR_ASSERT(started_);
- std::unique_lock<std::mutex> lock(shutdown_mu_);
- if (shutdown_) {
- return;
- }
- lock.unlock();
- grpc_call_error err = grpc_server_request_call_old(server_, nullptr);
- GPR_ASSERT(err == GRPC_CALL_OK);
-}
-
-void AsyncServer::Shutdown() {
- std::unique_lock<std::mutex> lock(shutdown_mu_);
- if (started_ && !shutdown_) {
- shutdown_ = true;
- lock.unlock();
- // TODO(yangg) should we shutdown without start?
- grpc_server_shutdown(server_);
- }
-}
-
-} // namespace grpc
diff --git a/src/cpp/server/server.cc b/src/cpp/server/server.cc
index 1abdf70..17b0543 100644
--- a/src/cpp/server/server.cc
+++ b/src/cpp/server/server.cc
@@ -37,29 +37,26 @@
#include <grpc/grpc.h>
#include <grpc/grpc_security.h>
#include <grpc/support/log.h>
-#include "src/cpp/server/server_rpc_handler.h"
-#include "src/cpp/server/thread_pool.h"
-#include <grpc++/async_server_context.h>
#include <grpc++/completion_queue.h>
#include <grpc++/impl/rpc_service_method.h>
+#include <grpc++/server_context.h>
#include <grpc++/server_credentials.h>
+#include <grpc++/thread_pool_interface.h>
+
+#include "src/cpp/proto/proto_utils.h"
namespace grpc {
-// TODO(rocking): consider a better default value like num of cores.
-static const int kNumThreads = 4;
-
-Server::Server(ThreadPoolInterface *thread_pool, ServerCredentials *creds)
+Server::Server(ThreadPoolInterface *thread_pool, bool thread_pool_owned,
+ ServerCredentials *creds)
: started_(false),
shutdown_(false),
num_running_cb_(0),
- thread_pool_(thread_pool == nullptr ? new ThreadPool(kNumThreads)
- : thread_pool),
- thread_pool_owned_(thread_pool == nullptr),
+ thread_pool_(thread_pool),
+ thread_pool_owned_(thread_pool_owned),
secure_(creds != nullptr) {
if (creds) {
- server_ =
- grpc_secure_server_create(creds->GetRawCreds(), cq_.cq(), nullptr);
+ server_ = grpc_secure_server_create(creds->GetRawCreds(), cq_.cq(), nullptr);
} else {
server_ = grpc_server_create(cq_.cq(), nullptr);
}
@@ -75,6 +72,8 @@
if (started_ && !shutdown_) {
lock.unlock();
Shutdown();
+ } else {
+ lock.unlock();
}
grpc_server_destroy(server_);
if (thread_pool_owned_) {
@@ -82,37 +81,151 @@
}
}
-void Server::RegisterService(RpcService *service) {
+bool Server::RegisterService(RpcService *service) {
for (int i = 0; i < service->GetMethodCount(); ++i) {
RpcServiceMethod *method = service->GetMethod(i);
- method_map_.insert(std::make_pair(method->name(), method));
+ void *tag = grpc_server_register_method(server_, method->name(), nullptr);
+ if (!tag) {
+ gpr_log(GPR_DEBUG, "Attempt to register %s multiple times",
+ method->name());
+ return false;
+ }
+ methods_.emplace_back(method, tag);
}
+ return true;
}
-void Server::AddPort(const grpc::string &addr) {
+int Server::AddPort(const grpc::string &addr) {
GPR_ASSERT(!started_);
- int success;
if (secure_) {
- success = grpc_server_add_secure_http2_port(server_, addr.c_str());
+ return grpc_server_add_secure_http2_port(server_, addr.c_str());
} else {
- success = grpc_server_add_http2_port(server_, addr.c_str());
+ return grpc_server_add_http2_port(server_, addr.c_str());
}
- GPR_ASSERT(success);
}
-void Server::Start() {
+class Server::MethodRequestData final : public CompletionQueueTag {
+ public:
+ MethodRequestData(RpcServiceMethod *method, void *tag)
+ : method_(method),
+ tag_(tag),
+ has_request_payload_(method->method_type() == RpcMethod::NORMAL_RPC ||
+ method->method_type() ==
+ RpcMethod::SERVER_STREAMING),
+ has_response_payload_(method->method_type() == RpcMethod::NORMAL_RPC ||
+ method->method_type() ==
+ RpcMethod::CLIENT_STREAMING) {
+ grpc_metadata_array_init(&request_metadata_);
+ }
+
+ static MethodRequestData *Wait(CompletionQueue *cq, bool *ok) {
+ void *tag = nullptr;
+ *ok = false;
+ if (!cq->Next(&tag, ok)) {
+ return nullptr;
+ }
+ auto *mrd = static_cast<MethodRequestData *>(tag);
+ GPR_ASSERT(mrd->in_flight_);
+ return mrd;
+ }
+
+ void Request(grpc_server *server) {
+ GPR_ASSERT(!in_flight_);
+ in_flight_ = true;
+ cq_ = grpc_completion_queue_create();
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_server_request_registered_call(
+ server, tag_, &call_, &deadline_, &request_metadata_,
+ has_request_payload_ ? &request_payload_ : nullptr,
+ cq_, this));
+ }
+
+ void FinalizeResult(void **tag, bool *status) override {}
+
+ class CallData {
+ public:
+ explicit CallData(Server *server, MethodRequestData *mrd)
+ : cq_(mrd->cq_),
+ call_(mrd->call_, server, &cq_),
+ ctx_(mrd->deadline_, mrd->request_metadata_.metadata,
+ mrd->request_metadata_.count),
+ has_request_payload_(mrd->has_request_payload_),
+ has_response_payload_(mrd->has_response_payload_),
+ request_payload_(mrd->request_payload_),
+ method_(mrd->method_) {
+ GPR_ASSERT(mrd->in_flight_);
+ mrd->in_flight_ = false;
+ mrd->request_metadata_.count = 0;
+ }
+
+ ~CallData() {
+ if (call_.call()) grpc_call_destroy(call_.call());
+ }
+
+ void Run() {
+ std::unique_ptr<google::protobuf::Message> req;
+ std::unique_ptr<google::protobuf::Message> res;
+ if (has_request_payload_) {
+ req.reset(method_->AllocateRequestProto());
+ if (!DeserializeProto(request_payload_, req.get())) {
+ abort(); // for now
+ }
+ }
+ if (has_response_payload_) {
+ res.reset(method_->AllocateResponseProto());
+ }
+ auto status = method_->handler()->RunHandler(
+ MethodHandler::HandlerParameter(&call_, &ctx_, req.get(), res.get()));
+ CallOpBuffer buf;
+ ctx_.SendInitialMetadataIfNeeded(&buf);
+ if (has_response_payload_) {
+ buf.AddSendMessage(*res);
+ }
+ buf.AddServerSendStatus(&ctx_.trailing_metadata_, status);
+ bool cancelled;
+ buf.AddServerRecvClose(&cancelled);
+ call_.PerformOps(&buf);
+ GPR_ASSERT(cq_.Pluck(&buf));
+ }
+
+ private:
+ CompletionQueue cq_;
+ Call call_;
+ ServerContext ctx_;
+ const bool has_request_payload_;
+ const bool has_response_payload_;
+ grpc_byte_buffer *request_payload_;
+ RpcServiceMethod *const method_;
+ };
+
+ private:
+ RpcServiceMethod *const method_;
+ void *const tag_;
+ bool in_flight_ = false;
+ const bool has_request_payload_;
+ const bool has_response_payload_;
+ grpc_call *call_;
+ gpr_timespec deadline_;
+ grpc_metadata_array request_metadata_;
+ grpc_byte_buffer *request_payload_;
+ grpc_completion_queue *cq_;
+};
+
+bool Server::Start() {
GPR_ASSERT(!started_);
started_ = true;
grpc_server_start(server_);
// Start processing rpcs.
- ScheduleCallback();
-}
+ if (!methods_.empty()) {
+ for (auto &m : methods_) {
+ m.Request(server_);
+ }
-void Server::AllowOneRpc() {
- GPR_ASSERT(started_);
- grpc_call_error err = grpc_server_request_call_old(server_, nullptr);
- GPR_ASSERT(err == GRPC_CALL_OK);
+ ScheduleCallback();
+ }
+
+ return true;
}
void Server::Shutdown() {
@@ -121,6 +234,7 @@
if (started_ && !shutdown_) {
shutdown_ = true;
grpc_server_shutdown(server_);
+ cq_.Shutdown();
// Wait for running callbacks to finish.
while (num_running_cb_ != 0) {
@@ -128,12 +242,16 @@
}
}
}
+}
- // Shutdown the completion queue.
- cq_.Shutdown();
- void *tag = nullptr;
- CompletionQueue::CompletionType t = cq_.Next(&tag);
- GPR_ASSERT(t == CompletionQueue::QUEUE_CLOSED);
+void Server::PerformOpsOnCall(CallOpBuffer *buf, Call *call) {
+ static const size_t MAX_OPS = 8;
+ size_t nops = MAX_OPS;
+ grpc_op ops[MAX_OPS];
+ buf->FillOps(ops, &nops);
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_call_start_batch(call->call(), ops, nops,
+ buf));
}
void Server::ScheduleCallback() {
@@ -141,30 +259,21 @@
std::unique_lock<std::mutex> lock(mu_);
num_running_cb_++;
}
- std::function<void()> callback = std::bind(&Server::RunRpc, this);
- thread_pool_->ScheduleCallback(callback);
+ thread_pool_->ScheduleCallback(std::bind(&Server::RunRpc, this));
}
void Server::RunRpc() {
// Wait for one more incoming rpc.
- void *tag = nullptr;
- AllowOneRpc();
- CompletionQueue::CompletionType t = cq_.Next(&tag);
- GPR_ASSERT(t == CompletionQueue::SERVER_RPC_NEW);
-
- AsyncServerContext *server_context = static_cast<AsyncServerContext *>(tag);
- // server_context could be nullptr during server shutdown.
- if (server_context != nullptr) {
- // Schedule a new callback to handle more rpcs.
+ bool ok;
+ auto *mrd = MethodRequestData::Wait(&cq_, &ok);
+ if (mrd) {
ScheduleCallback();
+ if (ok) {
+ MethodRequestData::CallData cd(this, mrd);
+ mrd->Request(server_);
- RpcServiceMethod *method = nullptr;
- auto iter = method_map_.find(server_context->method());
- if (iter != method_map_.end()) {
- method = iter->second;
+ cd.Run();
}
- ServerRpcHandler rpc_handler(server_context, method);
- rpc_handler.StartRpc();
}
{
diff --git a/src/cpp/server/server_builder.cc b/src/cpp/server/server_builder.cc
index add22cc..d6bcb93 100644
--- a/src/cpp/server/server_builder.cc
+++ b/src/cpp/server/server_builder.cc
@@ -33,15 +33,22 @@
#include <grpc++/server_builder.h>
+#include <grpc/support/cpu.h>
#include <grpc/support/log.h>
+#include <grpc++/impl/service_type.h>
#include <grpc++/server.h>
+#include "src/cpp/server/thread_pool.h"
namespace grpc {
-ServerBuilder::ServerBuilder() : thread_pool_(nullptr) {}
+ServerBuilder::ServerBuilder() {}
-void ServerBuilder::RegisterService(RpcService *service) {
- services_.push_back(service);
+void ServerBuilder::RegisterService(SynchronousService *service) {
+ services_.push_back(service->service());
+}
+
+void ServerBuilder::RegisterAsyncService(AsynchronousService *service) {
+ async_services_.push_back(service);
}
void ServerBuilder::AddPort(const grpc::string &addr) {
@@ -59,14 +66,31 @@
}
std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
- std::unique_ptr<Server> server(new Server(thread_pool_, creds_.get()));
+ bool thread_pool_owned = false;
+ if (!async_services_.empty() && !services_.empty()) {
+ gpr_log(GPR_ERROR, "Mixing async and sync services is unsupported for now");
+ return nullptr;
+ }
+ if (!thread_pool_ && services_.size()) {
+ int cores = gpr_cpu_num_cores();
+ if (!cores) cores = 4;
+ thread_pool_ = new ThreadPool(cores);
+ thread_pool_owned = true;
+ }
+ std::unique_ptr<Server> server(new Server(thread_pool_, thread_pool_owned, creds_.get()));
for (auto *service : services_) {
- server->RegisterService(service);
+ if (!server->RegisterService(service)) {
+ return nullptr;
+ }
}
for (auto &port : ports_) {
- server->AddPort(port);
+ if (!server->AddPort(port)) {
+ return nullptr;
+ }
}
- server->Start();
+ if (!server->Start()) {
+ return nullptr;
+ }
return server;
}
diff --git a/src/cpp/server/server_context_impl.h b/src/cpp/server/server_context.cc
similarity index 70%
rename from src/cpp/server/server_context_impl.h
rename to src/cpp/server/server_context.cc
index c6016b7..06eeb39 100644
--- a/src/cpp/server/server_context_impl.h
+++ b/src/cpp/server/server_context.cc
@@ -31,31 +31,29 @@
*
*/
-#ifndef __GRPCPP_INTERNAL_SERVER_SERVER_CONTEXT_IMPL_H_
-#define __GRPCPP_INTERNAL_SERVER_SERVER_CONTEXT_IMPL_H_
-
#include <grpc++/server_context.h>
-
-#include <chrono>
-
-#include <grpc/support/time.h>
+#include <grpc++/impl/call.h>
+#include <grpc/grpc.h>
+#include "src/cpp/util/time.h"
namespace grpc {
-class ServerContextImpl : public ServerContext {
- public:
- explicit ServerContextImpl(std::chrono::system_clock::time_point deadline)
- : absolute_deadline_(deadline) {}
- ~ServerContextImpl() {}
-
- std::chrono::system_clock::time_point absolute_deadline() const {
- return absolute_deadline_;
+ServerContext::ServerContext(gpr_timespec deadline, grpc_metadata *metadata,
+ size_t metadata_count)
+ : deadline_(Timespec2Timepoint(deadline)) {
+ for (size_t i = 0; i < metadata_count; i++) {
+ client_metadata_.insert(std::make_pair(
+ grpc::string(metadata[i].key),
+ grpc::string(metadata[i].value,
+ metadata[i].value + metadata[i].value_length)));
}
+}
- private:
- std::chrono::system_clock::time_point absolute_deadline_;
-};
+void ServerContext::SendInitialMetadataIfNeeded(CallOpBuffer* buf) {
+ if (!sent_initial_metadata_) {
+ buf->AddSendInitialMetadata(&initial_metadata_);
+ sent_initial_metadata_ = true;
+ }
+}
} // namespace grpc
-
-#endif // __GRPCPP_INTERNAL_SERVER_SERVER_CONTEXT_IMPL_H_
diff --git a/src/cpp/server/server_rpc_handler.cc b/src/cpp/server/server_rpc_handler.cc
deleted file mode 100644
index bf02de8..0000000
--- a/src/cpp/server/server_rpc_handler.cc
+++ /dev/null
@@ -1,140 +0,0 @@
-/*
- *
- * Copyright 2014, Google Inc.
- * All rights reserved.
- *
- * Redistribution and use in source and binary forms, with or without
- * modification, are permitted provided that the following conditions are
- * met:
- *
- * * Redistributions of source code must retain the above copyright
- * notice, this list of conditions and the following disclaimer.
- * * Redistributions in binary form must reproduce the above
- * copyright notice, this list of conditions and the following disclaimer
- * in the documentation and/or other materials provided with the
- * distribution.
- * * Neither the name of Google Inc. nor the names of its
- * contributors may be used to endorse or promote products derived from
- * this software without specific prior written permission.
- *
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
- * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
- * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
- * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
- * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
- * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
- * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
- * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
- * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
- * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
- * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
- *
- */
-
-#include "src/cpp/server/server_rpc_handler.h"
-
-#include <grpc/support/log.h>
-#include "src/cpp/server/server_context_impl.h"
-#include "src/cpp/stream/stream_context.h"
-#include <grpc++/async_server_context.h>
-#include <grpc++/impl/rpc_service_method.h>
-
-namespace grpc {
-
-ServerRpcHandler::ServerRpcHandler(AsyncServerContext *async_server_context,
- RpcServiceMethod *method)
- : async_server_context_(async_server_context), method_(method) {}
-
-void ServerRpcHandler::StartRpc() {
- if (method_ == nullptr) {
- // Method not supported, finish the rpc with error.
- // TODO(rocking): do we need to call read to consume the request?
- FinishRpc(Status(StatusCode::UNIMPLEMENTED, "No such method."));
- return;
- }
-
- ServerContextImpl user_context(async_server_context_->absolute_deadline());
-
- if (method_->method_type() == RpcMethod::NORMAL_RPC) {
- // Start the rpc on this dedicated completion queue.
- async_server_context_->Accept(cq_.cq());
-
- // Allocate request and response.
- std::unique_ptr<google::protobuf::Message> request(
- method_->AllocateRequestProto());
- std::unique_ptr<google::protobuf::Message> response(
- method_->AllocateResponseProto());
-
- // Read request
- async_server_context_->StartRead(request.get());
- auto type = WaitForNextEvent();
- GPR_ASSERT(type == CompletionQueue::SERVER_READ_OK);
-
- // Run the application's rpc handler
- MethodHandler *handler = method_->handler();
- Status status = handler->RunHandler(MethodHandler::HandlerParameter(
- &user_context, request.get(), response.get()));
-
- if (status.IsOk()) {
- // Send the response if we get an ok status.
- async_server_context_->StartWrite(*response, GRPC_WRITE_BUFFER_HINT);
- type = WaitForNextEvent();
- if (type != CompletionQueue::SERVER_WRITE_OK) {
- status = Status(StatusCode::INTERNAL, "Error writing response.");
- }
- }
-
- FinishRpc(status);
- } else {
- // Allocate request and response.
- // TODO(yangg) maybe not allocate both when not needed?
- std::unique_ptr<google::protobuf::Message> request(
- method_->AllocateRequestProto());
- std::unique_ptr<google::protobuf::Message> response(
- method_->AllocateResponseProto());
-
- StreamContext stream_context(*method_, async_server_context_->call(),
- cq_.cq(), request.get(), response.get());
-
- // Run the application's rpc handler
- MethodHandler *handler = method_->handler();
- Status status = handler->RunHandler(MethodHandler::HandlerParameter(
- &user_context, request.get(), response.get(), &stream_context));
- if (status.IsOk() &&
- method_->method_type() == RpcMethod::CLIENT_STREAMING) {
- stream_context.Write(response.get(), false);
- }
- // TODO(yangg) Do we need to consider the status in stream_context?
- FinishRpc(status);
- }
-}
-
-CompletionQueue::CompletionType ServerRpcHandler::WaitForNextEvent() {
- void *tag = nullptr;
- CompletionQueue::CompletionType type = cq_.Next(&tag);
- if (type != CompletionQueue::QUEUE_CLOSED &&
- type != CompletionQueue::RPC_END) {
- GPR_ASSERT(static_cast<AsyncServerContext *>(tag) ==
- async_server_context_.get());
- }
- return type;
-}
-
-void ServerRpcHandler::FinishRpc(const Status &status) {
- async_server_context_->StartWriteStatus(status);
- CompletionQueue::CompletionType type;
-
- // HALFCLOSE_OK and RPC_END events come in either order.
- type = WaitForNextEvent();
- GPR_ASSERT(type == CompletionQueue::HALFCLOSE_OK ||
- type == CompletionQueue::RPC_END);
- type = WaitForNextEvent();
- GPR_ASSERT(type == CompletionQueue::HALFCLOSE_OK ||
- type == CompletionQueue::RPC_END);
-
- cq_.Shutdown();
- type = WaitForNextEvent();
- GPR_ASSERT(type == CompletionQueue::QUEUE_CLOSED);
-}
-
-} // namespace grpc
diff --git a/src/cpp/server/thread_pool.h b/src/cpp/server/thread_pool.h
index c53f7a7..8a28c87 100644
--- a/src/cpp/server/thread_pool.h
+++ b/src/cpp/server/thread_pool.h
@@ -44,12 +44,12 @@
namespace grpc {
-class ThreadPool : public ThreadPoolInterface {
+class ThreadPool final : public ThreadPoolInterface {
public:
explicit ThreadPool(int num_threads);
~ThreadPool();
- void ScheduleCallback(const std::function<void()> &callback) final;
+ void ScheduleCallback(const std::function<void()> &callback) override;
private:
std::mutex mu_;
diff --git a/src/cpp/stream/stream_context.cc b/src/cpp/stream/stream_context.cc
deleted file mode 100644
index e4f344d..0000000
--- a/src/cpp/stream/stream_context.cc
+++ /dev/null
@@ -1,179 +0,0 @@
-/*
- *
- * Copyright 2014, Google Inc.
- * All rights reserved.
- *
- * Redistribution and use in source and binary forms, with or without
- * modification, are permitted provided that the following conditions are
- * met:
- *
- * * Redistributions of source code must retain the above copyright
- * notice, this list of conditions and the following disclaimer.
- * * Redistributions in binary form must reproduce the above
- * copyright notice, this list of conditions and the following disclaimer
- * in the documentation and/or other materials provided with the
- * distribution.
- * * Neither the name of Google Inc. nor the names of its
- * contributors may be used to endorse or promote products derived from
- * this software without specific prior written permission.
- *
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
- * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
- * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
- * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
- * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
- * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
- * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
- * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
- * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
- * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
- * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
- *
- */
-
-#include "src/cpp/stream/stream_context.h"
-
-#include <grpc/support/log.h>
-#include "src/cpp/proto/proto_utils.h"
-#include "src/cpp/util/time.h"
-#include <grpc++/client_context.h>
-#include <grpc++/config.h>
-#include <grpc++/impl/rpc_method.h>
-#include <google/protobuf/message.h>
-
-namespace grpc {
-
-// Client only ctor
-StreamContext::StreamContext(const RpcMethod &method, ClientContext *context,
- const google::protobuf::Message *request,
- google::protobuf::Message *result)
- : is_client_(true),
- method_(&method),
- call_(context->call()),
- cq_(context->cq()),
- request_(const_cast<google::protobuf::Message *>(request)),
- result_(result),
- peer_halfclosed_(false),
- self_halfclosed_(false) {
- GPR_ASSERT(method_->method_type() != RpcMethod::RpcType::NORMAL_RPC);
-}
-
-// Server only ctor
-StreamContext::StreamContext(const RpcMethod &method, grpc_call *call,
- grpc_completion_queue *cq,
- google::protobuf::Message *request,
- google::protobuf::Message *result)
- : is_client_(false),
- method_(&method),
- call_(call),
- cq_(cq),
- request_(request),
- result_(result),
- peer_halfclosed_(false),
- self_halfclosed_(false) {
- GPR_ASSERT(method_->method_type() != RpcMethod::RpcType::NORMAL_RPC);
-}
-
-StreamContext::~StreamContext() {}
-
-void StreamContext::Start(bool buffered) {
- if (is_client_) {
- // TODO(yangg) handle metadata send path
- int flag = buffered ? GRPC_WRITE_BUFFER_HINT : 0;
- grpc_call_error error = grpc_call_invoke_old(
- call(), cq(), client_metadata_read_tag(), finished_tag(), flag);
- GPR_ASSERT(GRPC_CALL_OK == error);
- } else {
- // TODO(yangg) metadata needs to be added before accept
- // TODO(yangg) correctly set flag to accept
- GPR_ASSERT(grpc_call_server_accept_old(call(), cq(), finished_tag()) ==
- GRPC_CALL_OK);
- GPR_ASSERT(grpc_call_server_end_initial_metadata_old(call(), 0) ==
- GRPC_CALL_OK);
- }
-}
-
-bool StreamContext::Read(google::protobuf::Message *msg) {
- // TODO(yangg) check peer_halfclosed_ here for possible early return.
- grpc_call_error err = grpc_call_start_read_old(call(), read_tag());
- GPR_ASSERT(err == GRPC_CALL_OK);
- grpc_event *read_ev =
- grpc_completion_queue_pluck(cq(), read_tag(), gpr_inf_future);
- GPR_ASSERT(read_ev->type == GRPC_READ);
- bool ret = true;
- if (read_ev->data.read) {
- if (!DeserializeProto(read_ev->data.read, msg)) {
- ret = false;
- grpc_call_cancel_with_status(call(), GRPC_STATUS_DATA_LOSS,
- "Failed to parse incoming proto");
- }
- } else {
- ret = false;
- peer_halfclosed_ = true;
- }
- grpc_event_finish(read_ev);
- return ret;
-}
-
-bool StreamContext::Write(const google::protobuf::Message *msg, bool is_last) {
- // TODO(yangg) check self_halfclosed_ for possible early return.
- bool ret = true;
- grpc_event *ev = nullptr;
-
- if (msg) {
- grpc_byte_buffer *out_buf = nullptr;
- if (!SerializeProto(*msg, &out_buf)) {
- grpc_call_cancel_with_status(call(), GRPC_STATUS_INVALID_ARGUMENT,
- "Failed to serialize outgoing proto");
- return false;
- }
- int flag = is_last ? GRPC_WRITE_BUFFER_HINT : 0;
- grpc_call_error err =
- grpc_call_start_write_old(call(), out_buf, write_tag(), flag);
- grpc_byte_buffer_destroy(out_buf);
- GPR_ASSERT(err == GRPC_CALL_OK);
-
- ev = grpc_completion_queue_pluck(cq(), write_tag(), gpr_inf_future);
- GPR_ASSERT(ev->type == GRPC_WRITE_ACCEPTED);
-
- ret = ev->data.write_accepted == GRPC_OP_OK;
- grpc_event_finish(ev);
- }
- if (ret && is_last) {
- grpc_call_error err = grpc_call_writes_done_old(call(), halfclose_tag());
- GPR_ASSERT(err == GRPC_CALL_OK);
- ev = grpc_completion_queue_pluck(cq(), halfclose_tag(), gpr_inf_future);
- GPR_ASSERT(ev->type == GRPC_FINISH_ACCEPTED);
- grpc_event_finish(ev);
-
- self_halfclosed_ = true;
- } else if (!ret) { // Stream broken
- self_halfclosed_ = true;
- peer_halfclosed_ = true;
- }
-
- return ret;
-}
-
-const Status &StreamContext::Wait() {
- // TODO(yangg) properly support metadata
- grpc_event *metadata_ev = grpc_completion_queue_pluck(
- cq(), client_metadata_read_tag(), gpr_inf_future);
- grpc_event_finish(metadata_ev);
- // TODO(yangg) protect states by a mutex, including other places.
- if (!self_halfclosed_ || !peer_halfclosed_) {
- Cancel();
- }
- grpc_event *finish_ev =
- grpc_completion_queue_pluck(cq(), finished_tag(), gpr_inf_future);
- GPR_ASSERT(finish_ev->type == GRPC_FINISHED);
- final_status_ = Status(
- static_cast<StatusCode>(finish_ev->data.finished.status),
- finish_ev->data.finished.details ? finish_ev->data.finished.details : "");
- grpc_event_finish(finish_ev);
- return final_status_;
-}
-
-void StreamContext::Cancel() { grpc_call_cancel(call()); }
-
-} // namespace grpc
diff --git a/src/cpp/stream/stream_context.h b/src/cpp/stream/stream_context.h
deleted file mode 100644
index 8def589..0000000
--- a/src/cpp/stream/stream_context.h
+++ /dev/null
@@ -1,99 +0,0 @@
-/*
- *
- * Copyright 2014, Google Inc.
- * All rights reserved.
- *
- * Redistribution and use in source and binary forms, with or without
- * modification, are permitted provided that the following conditions are
- * met:
- *
- * * Redistributions of source code must retain the above copyright
- * notice, this list of conditions and the following disclaimer.
- * * Redistributions in binary form must reproduce the above
- * copyright notice, this list of conditions and the following disclaimer
- * in the documentation and/or other materials provided with the
- * distribution.
- * * Neither the name of Google Inc. nor the names of its
- * contributors may be used to endorse or promote products derived from
- * this software without specific prior written permission.
- *
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
- * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
- * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
- * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
- * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
- * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
- * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
- * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
- * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
- * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
- * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
- *
- */
-
-#ifndef __GRPCPP_INTERNAL_STREAM_STREAM_CONTEXT_H__
-#define __GRPCPP_INTERNAL_STREAM_STREAM_CONTEXT_H__
-
-#include <grpc/grpc.h>
-#include <grpc++/status.h>
-#include <grpc++/stream_context_interface.h>
-
-namespace google {
-namespace protobuf {
-class Message;
-}
-}
-
-namespace grpc {
-class ClientContext;
-class RpcMethod;
-
-class StreamContext final : public StreamContextInterface {
- public:
- StreamContext(const RpcMethod &method, ClientContext *context,
- const google::protobuf::Message *request,
- google::protobuf::Message *result);
- StreamContext(const RpcMethod &method, grpc_call *call,
- grpc_completion_queue *cq, google::protobuf::Message *request,
- google::protobuf::Message *result);
- ~StreamContext();
- // Start the stream, if there is a final write following immediately, set
- // buffered so that the messages can be sent in batch.
- void Start(bool buffered) override;
- bool Read(google::protobuf::Message *msg) override;
- bool Write(const google::protobuf::Message *msg, bool is_last) override;
- const Status &Wait() override;
- void Cancel() override;
-
- google::protobuf::Message *request() override { return request_; }
- google::protobuf::Message *response() override { return result_; }
-
- private:
- // Unique tags for plucking events from the c layer. this pointer is casted
- // to char* to create single byte step between tags. It implicitly relies on
- // that StreamContext is large enough to contain all the pointers.
- void *finished_tag() { return reinterpret_cast<char *>(this); }
- void *read_tag() { return reinterpret_cast<char *>(this) + 1; }
- void *write_tag() { return reinterpret_cast<char *>(this) + 2; }
- void *halfclose_tag() { return reinterpret_cast<char *>(this) + 3; }
- void *client_metadata_read_tag() {
- return reinterpret_cast<char *>(this) + 5;
- }
- grpc_call *call() { return call_; }
- grpc_completion_queue *cq() { return cq_; }
-
- bool is_client_;
- const RpcMethod *method_; // not owned
- grpc_call *call_; // not owned
- grpc_completion_queue *cq_; // not owned
- google::protobuf::Message *request_; // first request, not owned
- google::protobuf::Message *result_; // last response, not owned
-
- bool peer_halfclosed_;
- bool self_halfclosed_;
- Status final_status_;
-};
-
-} // namespace grpc
-
-#endif // __GRPCPP_INTERNAL_STREAM_STREAM_CONTEXT_H__
diff --git a/test/core/end2end/tests/cancel_after_accept.c b/test/core/end2end/tests/cancel_after_accept.c
index 18d6bce..6527fef 100644
--- a/test/core/end2end/tests/cancel_after_accept.c
+++ b/test/core/end2end/tests/cancel_after_accept.c
@@ -166,7 +166,8 @@
GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
&call_details,
&request_metadata_recv,
- f.server_cq, tag(2)));
+ f.server_cq, f.server_cq,
+ tag(2)));
cq_expect_completion(v_server, tag(2), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c b/test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c
index 940e327..695ddbc 100644
--- a/test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c
+++ b/test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c
@@ -175,7 +175,8 @@
GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
&call_details,
&request_metadata_recv,
- f.server_cq, tag(101)));
+ f.server_cq, f.server_cq,
+ tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/request_response_with_metadata_and_payload.c b/test/core/end2end/tests/request_response_with_metadata_and_payload.c
index 80cb629..d8624a3 100644
--- a/test/core/end2end/tests/request_response_with_metadata_and_payload.c
+++ b/test/core/end2end/tests/request_response_with_metadata_and_payload.c
@@ -168,7 +168,8 @@
GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
&call_details,
&request_metadata_recv,
- f.server_cq, tag(101)));
+ f.server_cq, f.server_cq,
+ tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/request_response_with_payload.c b/test/core/end2end/tests/request_response_with_payload.c
index b07f51d..6a6c792 100644
--- a/test/core/end2end/tests/request_response_with_payload.c
+++ b/test/core/end2end/tests/request_response_with_payload.c
@@ -162,7 +162,8 @@
GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
&call_details,
&request_metadata_recv,
- f.server_cq, tag(101)));
+ f.server_cq, f.server_cq,
+ tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/request_response_with_trailing_metadata_and_payload.c b/test/core/end2end/tests/request_response_with_trailing_metadata_and_payload.c
index e547604..76e298d 100644
--- a/test/core/end2end/tests/request_response_with_trailing_metadata_and_payload.c
+++ b/test/core/end2end/tests/request_response_with_trailing_metadata_and_payload.c
@@ -169,7 +169,8 @@
GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
&call_details,
&request_metadata_recv,
- f.server_cq, tag(101)));
+ f.server_cq, f.server_cq,
+ tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/request_with_large_metadata.c b/test/core/end2end/tests/request_with_large_metadata.c
index eb6180c..693f81e 100644
--- a/test/core/end2end/tests/request_with_large_metadata.c
+++ b/test/core/end2end/tests/request_with_large_metadata.c
@@ -166,7 +166,8 @@
GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
&call_details,
&request_metadata_recv,
- f.server_cq, tag(101)));
+ f.server_cq, f.server_cq,
+ tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/request_with_payload.c b/test/core/end2end/tests/request_with_payload.c
index 2bf0fa3..3924991 100644
--- a/test/core/end2end/tests/request_with_payload.c
+++ b/test/core/end2end/tests/request_with_payload.c
@@ -157,7 +157,8 @@
GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
&call_details,
&request_metadata_recv,
- f.server_cq, tag(101)));
+ f.server_cq, f.server_cq,
+ tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/simple_delayed_request.c b/test/core/end2end/tests/simple_delayed_request.c
index 80763fe..851d0cb 100644
--- a/test/core/end2end/tests/simple_delayed_request.c
+++ b/test/core/end2end/tests/simple_delayed_request.c
@@ -144,7 +144,8 @@
GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f->server, &s,
&call_details,
&request_metadata_recv,
- f->server_cq, tag(101)));
+ f->server_cq, f->server_cq,
+ tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/simple_request.c b/test/core/end2end/tests/simple_request.c
index 968be74..9b6230d 100644
--- a/test/core/end2end/tests/simple_request.c
+++ b/test/core/end2end/tests/simple_request.c
@@ -150,7 +150,8 @@
GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
&call_details,
&request_metadata_recv,
- f.server_cq, tag(101)));
+ f.server_cq, f.server_cq,
+ tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/statistics/census_log_tests.c b/test/core/statistics/census_log_tests.c
index c7b2b2e..e2ad78a 100644
--- a/test/core/statistics/census_log_tests.c
+++ b/test/core/statistics/census_log_tests.c
@@ -35,7 +35,7 @@
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
-#include "src/core/support/cpu.h"
+#include <grpc/support/cpu.h>
#include <grpc/support/log.h>
#include <grpc/support/port_platform.h>
#include <grpc/support/sync.h>
diff --git a/test/cpp/end2end/async_test_server.cc b/test/cpp/end2end/async_test_server.cc
deleted file mode 100644
index f18b6c0..0000000
--- a/test/cpp/end2end/async_test_server.cc
+++ /dev/null
@@ -1,154 +0,0 @@
-/*
- *
- * Copyright 2014, Google Inc.
- * All rights reserved.
- *
- * Redistribution and use in source and binary forms, with or without
- * modification, are permitted provided that the following conditions are
- * met:
- *
- * * Redistributions of source code must retain the above copyright
- * notice, this list of conditions and the following disclaimer.
- * * Redistributions in binary form must reproduce the above
- * copyright notice, this list of conditions and the following disclaimer
- * in the documentation and/or other materials provided with the
- * distribution.
- * * Neither the name of Google Inc. nor the names of its
- * contributors may be used to endorse or promote products derived from
- * this software without specific prior written permission.
- *
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
- * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
- * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
- * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
- * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
- * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
- * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
- * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
- * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
- * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
- * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
- *
- */
-
-#include "test/cpp/end2end/async_test_server.h"
-
-#include <chrono>
-
-#include <grpc/support/log.h>
-#include "src/cpp/proto/proto_utils.h"
-#include "test/cpp/util/echo.pb.h"
-#include <grpc++/async_server.h>
-#include <grpc++/async_server_context.h>
-#include <grpc++/completion_queue.h>
-#include <grpc++/status.h>
-#include <gtest/gtest.h>
-
-using grpc::cpp::test::util::EchoRequest;
-using grpc::cpp::test::util::EchoResponse;
-
-using std::chrono::duration_cast;
-using std::chrono::microseconds;
-using std::chrono::seconds;
-using std::chrono::system_clock;
-
-namespace grpc {
-namespace testing {
-
-AsyncTestServer::AsyncTestServer() : server_(&cq_), cq_drained_(false) {}
-
-AsyncTestServer::~AsyncTestServer() {}
-
-void AsyncTestServer::AddPort(const grpc::string& addr) {
- server_.AddPort(addr);
-}
-
-void AsyncTestServer::Start() { server_.Start(); }
-
-// Return true if deadline actual is within 0.5s from expected.
-bool DeadlineMatched(const system_clock::time_point& actual,
- const system_clock::time_point& expected) {
- microseconds diff_usecs = duration_cast<microseconds>(expected - actual);
- gpr_log(GPR_INFO, "diff_usecs= %d", diff_usecs.count());
- return diff_usecs.count() < 500000 && diff_usecs.count() > -500000;
-}
-
-void AsyncTestServer::RequestOneRpc() { server_.RequestOneRpc(); }
-
-void AsyncTestServer::MainLoop() {
- EchoRequest request;
- EchoResponse response;
- void* tag = nullptr;
-
- RequestOneRpc();
-
- while (true) {
- CompletionQueue::CompletionType t = cq_.Next(&tag);
- AsyncServerContext* server_context = static_cast<AsyncServerContext*>(tag);
- switch (t) {
- case CompletionQueue::SERVER_RPC_NEW:
- gpr_log(GPR_INFO, "SERVER_RPC_NEW %p", server_context);
- if (server_context) {
- EXPECT_EQ(server_context->method(), "/foo");
- // TODO(ctiller): verify deadline
- server_context->Accept(cq_.cq());
- // Handle only one rpc at a time.
- RequestOneRpc();
- server_context->StartRead(&request);
- }
- break;
- case CompletionQueue::RPC_END:
- gpr_log(GPR_INFO, "RPC_END %p", server_context);
- delete server_context;
- break;
- case CompletionQueue::SERVER_READ_OK:
- gpr_log(GPR_INFO, "SERVER_READ_OK %p", server_context);
- response.set_message(request.message());
- server_context->StartWrite(response, 0);
- break;
- case CompletionQueue::SERVER_READ_ERROR:
- gpr_log(GPR_INFO, "SERVER_READ_ERROR %p", server_context);
- server_context->StartWriteStatus(Status::OK);
- break;
- case CompletionQueue::HALFCLOSE_OK:
- gpr_log(GPR_INFO, "HALFCLOSE_OK %p", server_context);
- // Do nothing, just wait for RPC_END.
- break;
- case CompletionQueue::SERVER_WRITE_OK:
- gpr_log(GPR_INFO, "SERVER_WRITE_OK %p", server_context);
- server_context->StartRead(&request);
- break;
- case CompletionQueue::SERVER_WRITE_ERROR:
- EXPECT_TRUE(0);
- break;
- case CompletionQueue::QUEUE_CLOSED: {
- gpr_log(GPR_INFO, "QUEUE_CLOSED");
- HandleQueueClosed();
- return;
- }
- default:
- EXPECT_TRUE(0);
- break;
- }
- }
-}
-
-void AsyncTestServer::HandleQueueClosed() {
- std::unique_lock<std::mutex> lock(cq_drained_mu_);
- cq_drained_ = true;
- cq_drained_cv_.notify_all();
-}
-
-void AsyncTestServer::Shutdown() {
- // The server need to be shut down before cq_ as grpc_server flushes all
- // pending requested calls to the completion queue at shutdown.
- server_.Shutdown();
- cq_.Shutdown();
- std::unique_lock<std::mutex> lock(cq_drained_mu_);
- while (!cq_drained_) {
- cq_drained_cv_.wait(lock);
- }
-}
-
-} // namespace testing
-} // namespace grpc
diff --git a/test/cpp/end2end/async_test_server.h b/test/cpp/end2end/async_test_server.h
deleted file mode 100644
index a277061..0000000
--- a/test/cpp/end2end/async_test_server.h
+++ /dev/null
@@ -1,75 +0,0 @@
-/*
- *
- * Copyright 2014, Google Inc.
- * All rights reserved.
- *
- * Redistribution and use in source and binary forms, with or without
- * modification, are permitted provided that the following conditions are
- * met:
- *
- * * Redistributions of source code must retain the above copyright
- * notice, this list of conditions and the following disclaimer.
- * * Redistributions in binary form must reproduce the above
- * copyright notice, this list of conditions and the following disclaimer
- * in the documentation and/or other materials provided with the
- * distribution.
- * * Neither the name of Google Inc. nor the names of its
- * contributors may be used to endorse or promote products derived from
- * this software without specific prior written permission.
- *
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
- * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
- * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
- * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
- * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
- * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
- * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
- * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
- * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
- * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
- * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
- *
- */
-
-#ifndef __GRPCPP_TEST_END2END_ASYNC_TEST_SERVER_H__
-#define __GRPCPP_TEST_END2END_ASYNC_TEST_SERVER_H__
-
-#include <condition_variable>
-#include <mutex>
-#include <string>
-
-#include <grpc++/async_server.h>
-#include <grpc++/completion_queue.h>
-
-namespace grpc {
-
-namespace testing {
-
-class AsyncTestServer {
- public:
- AsyncTestServer();
- virtual ~AsyncTestServer();
-
- void AddPort(const grpc::string& addr);
- void Start();
- void RequestOneRpc();
- virtual void MainLoop();
- void Shutdown();
-
- CompletionQueue* completion_queue() { return &cq_; }
-
- protected:
- void HandleQueueClosed();
-
- private:
- CompletionQueue cq_;
- AsyncServer server_;
- bool cq_drained_;
- std::mutex cq_drained_mu_;
- std::condition_variable cq_drained_cv_;
-};
-
-} // namespace testing
-} // namespace grpc
-
-#endif // __GRPCPP_TEST_END2END_ASYNC_TEST_SERVER_H__
diff --git a/test/cpp/end2end/end2end_test.cc b/test/cpp/end2end/end2end_test.cc
index 4dea77e..2aaecb4 100644
--- a/test/cpp/end2end/end2end_test.cc
+++ b/test/cpp/end2end/end2end_test.cc
@@ -38,6 +38,7 @@
#include "test/cpp/util/echo_duplicate.pb.h"
#include "test/cpp/util/echo.pb.h"
#include "src/cpp/util/time.h"
+#include "src/cpp/server/thread_pool.h"
#include <grpc++/channel_arguments.h>
#include <grpc++/channel_interface.h>
#include <grpc++/client_context.h>
@@ -76,6 +77,7 @@
response->mutable_param()->set_request_deadline(deadline.tv_sec);
}
}
+
} // namespace
class TestServiceImpl : public ::grpc::cpp::test::util::TestService::Service {
@@ -141,14 +143,17 @@
class End2endTest : public ::testing::Test {
protected:
+ End2endTest() : thread_pool_(2) {}
+
void SetUp() override {
int port = grpc_pick_unused_port_or_die();
server_address_ << "localhost:" << port;
// Setup server
ServerBuilder builder;
builder.AddPort(server_address_.str());
- builder.RegisterService(service_.service());
- builder.RegisterService(dup_pkg_service_.service());
+ builder.RegisterService(&service_);
+ builder.RegisterService(&dup_pkg_service_);
+ builder.SetThreadPool(&thread_pool_);
server_ = builder.BuildAndStart();
}
@@ -165,6 +170,7 @@
std::ostringstream server_address_;
TestServiceImpl service_;
TestServiceImplDupPkg dup_pkg_service_;
+ ThreadPool thread_pool_;
};
static void SendRpc(grpc::cpp::test::util::TestService::Stub* stub,
@@ -290,7 +296,7 @@
request.set_message("hello");
EXPECT_TRUE(stream->Write(request));
stream->WritesDone();
- Status s = stream->Wait();
+ Status s = stream->Finish();
EXPECT_EQ(response.message(), request.message());
EXPECT_TRUE(s.IsOk());
@@ -308,7 +314,7 @@
EXPECT_TRUE(stream->Write(request));
EXPECT_TRUE(stream->Write(request));
stream->WritesDone();
- Status s = stream->Wait();
+ Status s = stream->Finish();
EXPECT_EQ(response.message(), "hellohello");
EXPECT_TRUE(s.IsOk());
@@ -332,7 +338,7 @@
EXPECT_EQ(response.message(), request.message() + "2");
EXPECT_FALSE(stream->Read(&response));
- Status s = stream->Wait();
+ Status s = stream->Finish();
EXPECT_TRUE(s.IsOk());
delete stream;
@@ -366,7 +372,7 @@
stream->WritesDone();
EXPECT_FALSE(stream->Read(&response));
- Status s = stream->Wait();
+ Status s = stream->Finish();
EXPECT_TRUE(s.IsOk());
delete stream;
@@ -422,7 +428,7 @@
ClientContext context2;
ClientReaderWriter<EchoRequest, EchoResponse>* stream =
stub->BidiStream(&context2);
- s = stream->Wait();
+ s = stream->Finish();
EXPECT_FALSE(s.IsOk());
EXPECT_EQ(StatusCode::UNKNOWN, s.code());
EXPECT_EQ("Rpc sent on a lame channel.", s.details());
diff --git a/test/cpp/end2end/sync_client_async_server_test.cc b/test/cpp/end2end/sync_client_async_server_test.cc
deleted file mode 100644
index 9955eb3..0000000
--- a/test/cpp/end2end/sync_client_async_server_test.cc
+++ /dev/null
@@ -1,236 +0,0 @@
-/*
- *
- * Copyright 2014, Google Inc.
- * All rights reserved.
- *
- * Redistribution and use in source and binary forms, with or without
- * modification, are permitted provided that the following conditions are
- * met:
- *
- * * Redistributions of source code must retain the above copyright
- * notice, this list of conditions and the following disclaimer.
- * * Redistributions in binary form must reproduce the above
- * copyright notice, this list of conditions and the following disclaimer
- * in the documentation and/or other materials provided with the
- * distribution.
- * * Neither the name of Google Inc. nor the names of its
- * contributors may be used to endorse or promote products derived from
- * this software without specific prior written permission.
- *
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
- * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
- * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
- * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
- * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
- * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
- * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
- * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
- * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
- * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
- * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
- *
- */
-
-#include <chrono>
-#include <memory>
-#include <sstream>
-#include <string>
-
-#include <grpc/grpc.h>
-#include <grpc/support/thd.h>
-#include "test/cpp/util/echo.pb.h"
-#include <grpc++/channel_arguments.h>
-#include <grpc++/channel_interface.h>
-#include <grpc++/client_context.h>
-#include <grpc++/create_channel.h>
-#include <grpc++/impl/internal_stub.h>
-#include <grpc++/impl/rpc_method.h>
-#include <grpc++/status.h>
-#include <grpc++/stream.h>
-#include "test/cpp/end2end/async_test_server.h"
-#include "test/core/util/port.h"
-#include <gtest/gtest.h>
-
-using grpc::cpp::test::util::EchoRequest;
-using grpc::cpp::test::util::EchoResponse;
-
-using std::chrono::duration_cast;
-using std::chrono::microseconds;
-using std::chrono::seconds;
-using std::chrono::system_clock;
-
-using grpc::testing::AsyncTestServer;
-
-namespace grpc {
-namespace {
-
-void ServerLoop(void* s) {
- AsyncTestServer* server = static_cast<AsyncTestServer*>(s);
- server->MainLoop();
-}
-
-class End2endTest : public ::testing::Test {
- protected:
- void SetUp() override {
- int port = grpc_pick_unused_port_or_die();
- // TODO(yangg) protobuf has a StringPrintf, maybe use that
- std::ostringstream oss;
- oss << "[::]:" << port;
- // Setup server
- server_.reset(new AsyncTestServer());
- server_->AddPort(oss.str());
- server_->Start();
-
- RunServerThread();
-
- // Setup client
- oss.str("");
- oss << "127.0.0.1:" << port;
- std::shared_ptr<ChannelInterface> channel =
- CreateChannel(oss.str(), ChannelArguments());
- stub_.set_channel(channel);
- }
-
- void RunServerThread() {
- gpr_thd_id id;
- EXPECT_TRUE(gpr_thd_new(&id, ServerLoop, server_.get(), NULL));
- }
-
- void TearDown() override { server_->Shutdown(); }
-
- std::unique_ptr<AsyncTestServer> server_;
- InternalStub stub_;
-};
-
-TEST_F(End2endTest, NoOpTest) { EXPECT_TRUE(stub_.channel() != nullptr); }
-
-TEST_F(End2endTest, SimpleRpc) {
- EchoRequest request;
- request.set_message("hello");
- EchoResponse result;
- ClientContext context;
- RpcMethod method("/foo");
- std::chrono::system_clock::time_point deadline =
- std::chrono::system_clock::now() + std::chrono::seconds(10);
- context.set_absolute_deadline(deadline);
- Status s =
- stub_.channel()->StartBlockingRpc(method, &context, request, &result);
- EXPECT_EQ(result.message(), request.message());
- EXPECT_TRUE(s.IsOk());
-}
-
-TEST_F(End2endTest, KSequentialSimpleRpcs) {
- int k = 3;
- for (int i = 0; i < k; i++) {
- EchoRequest request;
- request.set_message("hello");
- EchoResponse result;
- ClientContext context;
- RpcMethod method("/foo");
- std::chrono::system_clock::time_point deadline =
- std::chrono::system_clock::now() + std::chrono::seconds(10);
- context.set_absolute_deadline(deadline);
- Status s =
- stub_.channel()->StartBlockingRpc(method, &context, request, &result);
- EXPECT_EQ(result.message(), request.message());
- EXPECT_TRUE(s.IsOk());
- }
-}
-
-TEST_F(End2endTest, OnePingpongBidiStream) {
- EchoRequest request;
- request.set_message("hello");
- EchoResponse result;
- ClientContext context;
- RpcMethod method("/foo", RpcMethod::RpcType::BIDI_STREAMING);
- std::chrono::system_clock::time_point deadline =
- std::chrono::system_clock::now() + std::chrono::seconds(10);
- context.set_absolute_deadline(deadline);
- StreamContextInterface* stream_interface =
- stub_.channel()->CreateStream(method, &context, nullptr, nullptr);
- std::unique_ptr<ClientReaderWriter<EchoRequest, EchoResponse>> stream(
- new ClientReaderWriter<EchoRequest, EchoResponse>(stream_interface));
- EXPECT_TRUE(stream->Write(request));
- EXPECT_TRUE(stream->Read(&result));
- stream->WritesDone();
- EXPECT_FALSE(stream->Read(&result));
- Status s = stream->Wait();
- EXPECT_EQ(result.message(), request.message());
- EXPECT_TRUE(s.IsOk());
-}
-
-TEST_F(End2endTest, TwoPingpongBidiStream) {
- EchoRequest request;
- request.set_message("hello");
- EchoResponse result;
- ClientContext context;
- RpcMethod method("/foo", RpcMethod::RpcType::BIDI_STREAMING);
- std::chrono::system_clock::time_point deadline =
- std::chrono::system_clock::now() + std::chrono::seconds(10);
- context.set_absolute_deadline(deadline);
- StreamContextInterface* stream_interface =
- stub_.channel()->CreateStream(method, &context, nullptr, nullptr);
- std::unique_ptr<ClientReaderWriter<EchoRequest, EchoResponse>> stream(
- new ClientReaderWriter<EchoRequest, EchoResponse>(stream_interface));
- EXPECT_TRUE(stream->Write(request));
- EXPECT_TRUE(stream->Read(&result));
- EXPECT_EQ(result.message(), request.message());
- EXPECT_TRUE(stream->Write(request));
- EXPECT_TRUE(stream->Read(&result));
- EXPECT_EQ(result.message(), request.message());
- stream->WritesDone();
- EXPECT_FALSE(stream->Read(&result));
- Status s = stream->Wait();
- EXPECT_TRUE(s.IsOk());
-}
-
-TEST_F(End2endTest, OnePingpongClientStream) {
- EchoRequest request;
- request.set_message("hello");
- EchoResponse result;
- ClientContext context;
- RpcMethod method("/foo", RpcMethod::RpcType::CLIENT_STREAMING);
- std::chrono::system_clock::time_point deadline =
- std::chrono::system_clock::now() + std::chrono::seconds(10);
- context.set_absolute_deadline(deadline);
- StreamContextInterface* stream_interface =
- stub_.channel()->CreateStream(method, &context, nullptr, &result);
- std::unique_ptr<ClientWriter<EchoRequest>> stream(
- new ClientWriter<EchoRequest>(stream_interface));
- EXPECT_TRUE(stream->Write(request));
- stream->WritesDone();
- Status s = stream->Wait();
- EXPECT_EQ(result.message(), request.message());
- EXPECT_TRUE(s.IsOk());
-}
-
-TEST_F(End2endTest, OnePingpongServerStream) {
- EchoRequest request;
- request.set_message("hello");
- EchoResponse result;
- ClientContext context;
- RpcMethod method("/foo", RpcMethod::RpcType::SERVER_STREAMING);
- std::chrono::system_clock::time_point deadline =
- std::chrono::system_clock::now() + std::chrono::seconds(10);
- context.set_absolute_deadline(deadline);
- StreamContextInterface* stream_interface =
- stub_.channel()->CreateStream(method, &context, &request, nullptr);
- std::unique_ptr<ClientReader<EchoResponse>> stream(
- new ClientReader<EchoResponse>(stream_interface));
- EXPECT_TRUE(stream->Read(&result));
- EXPECT_FALSE(stream->Read(nullptr));
- Status s = stream->Wait();
- EXPECT_EQ(result.message(), request.message());
- EXPECT_TRUE(s.IsOk());
-}
-
-} // namespace
-} // namespace grpc
-
-int main(int argc, char** argv) {
- grpc_init();
- ::testing::InitGoogleTest(&argc, argv);
- int result = RUN_ALL_TESTS();
- grpc_shutdown();
- return result;
-}
diff --git a/test/cpp/interop/client.cc b/test/cpp/interop/client.cc
index 0fa76f0..29bbe4d 100644
--- a/test/cpp/interop/client.cc
+++ b/test/cpp/interop/client.cc
@@ -248,7 +248,7 @@
aggregated_payload_size += request_stream_sizes[i];
}
stream->WritesDone();
- grpc::Status s = stream->Wait();
+ grpc::Status s = stream->Finish();
GPR_ASSERT(response.aggregated_payload_size() == aggregated_payload_size);
GPR_ASSERT(s.IsOk());
@@ -278,7 +278,7 @@
++i;
}
GPR_ASSERT(response_stream_sizes.size() == i);
- grpc::Status s = stream->Wait();
+ grpc::Status s = stream->Finish();
GPR_ASSERT(s.IsOk());
gpr_log(GPR_INFO, "Response streaming done.");
@@ -311,7 +311,7 @@
++i;
}
GPR_ASSERT(kNumResponseMessages == i);
- grpc::Status s = stream->Wait();
+ grpc::Status s = stream->Finish();
GPR_ASSERT(s.IsOk());
gpr_log(GPR_INFO, "Response streaming done.");
@@ -345,7 +345,7 @@
++i;
}
GPR_ASSERT(response_stream_sizes.size() == i);
- grpc::Status s = stream->Wait();
+ grpc::Status s = stream->Finish();
GPR_ASSERT(s.IsOk());
gpr_log(GPR_INFO, "Half-duplex streaming rpc done.");
}
@@ -378,7 +378,7 @@
stream->WritesDone();
GPR_ASSERT(!stream->Read(&response));
- grpc::Status s = stream->Wait();
+ grpc::Status s = stream->Finish();
GPR_ASSERT(s.IsOk());
gpr_log(GPR_INFO, "Ping pong streaming done.");
}
diff --git a/test/cpp/interop/server.cc b/test/cpp/interop/server.cc
index 8a6be57..a839977 100644
--- a/test/cpp/interop/server.cc
+++ b/test/cpp/interop/server.cc
@@ -200,7 +200,7 @@
ServerBuilder builder;
builder.AddPort(server_address.str());
- builder.RegisterService(service.service());
+ builder.RegisterService(&service);
if (FLAGS_enable_ssl) {
SslServerCredentialsOptions ssl_opts = {
"", {{test_server1_key, test_server1_cert}}};
diff --git a/test/cpp/qps/server.cc b/test/cpp/qps/server.cc
index 3a432b6..7180461 100644
--- a/test/cpp/qps/server.cc
+++ b/test/cpp/qps/server.cc
@@ -128,7 +128,7 @@
ServerBuilder builder;
builder.AddPort(server_address);
- builder.RegisterService(service.service());
+ builder.RegisterService(&service);
std::unique_ptr<ThreadPool> pool(new ThreadPool(FLAGS_server_threads));
builder.SetThreadPool(pool.get());
diff --git a/tools/run_tests/tests.json b/tools/run_tests/tests.json
index 197dc3b..2ab3f50 100644
--- a/tools/run_tests/tests.json
+++ b/tools/run_tests/tests.json
@@ -283,10 +283,6 @@
},
{
"language": "c++",
- "name": "sync_client_async_server_test"
- },
- {
- "language": "c++",
"name": "thread_pool_test"
},
{
diff --git a/vsprojects/vs2013/grpc_shared.vcxproj b/vsprojects/vs2013/grpc_shared.vcxproj
index b1890cf..71b4a0c 100644
--- a/vsprojects/vs2013/grpc_shared.vcxproj
+++ b/vsprojects/vs2013/grpc_shared.vcxproj
@@ -277,6 +277,8 @@
</ClCompile>
<ClCompile Include="..\..\src\core\iomgr\pollset_multipoller_with_poll_posix.c">
</ClCompile>
+ <ClCompile Include="..\..\src\core\iomgr\pollset_multipoller_with_epoll.c">
+ </ClCompile>
<ClCompile Include="..\..\src\core\iomgr\pollset_posix.c">
</ClCompile>
<ClCompile Include="..\..\src\core\iomgr\pollset_windows.c">
diff --git a/vsprojects/vs2013/grpc_shared.vcxproj.filters b/vsprojects/vs2013/grpc_shared.vcxproj.filters
index fed8fb1..75ecc7a 100644
--- a/vsprojects/vs2013/grpc_shared.vcxproj.filters
+++ b/vsprojects/vs2013/grpc_shared.vcxproj.filters
@@ -130,6 +130,9 @@
<ClCompile Include="..\..\src\core\iomgr\pollset_multipoller_with_poll_posix.c">
<Filter>src\core\iomgr</Filter>
</ClCompile>
+ <ClCompile Include="..\..\src\core\iomgr\pollset_multipoller_with_epoll.c">
+ <Filter>src\core\iomgr</Filter>
+ </ClCompile>
<ClCompile Include="..\..\src\core\iomgr\pollset_posix.c">
<Filter>src\core\iomgr</Filter>
</ClCompile>