/* * * Copyright 2015-2016 gRPC authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. * */ #include #include #include #include #include #include #include #include #include #include #include #include #include "src/core/lib/transport/byte_stream.h" #include "src/proto/grpc/testing/empty.pb.h" #include "src/proto/grpc/testing/messages.pb.h" #include "src/proto/grpc/testing/test.grpc.pb.h" #include "test/cpp/interop/client_helper.h" #include "test/cpp/interop/interop_client.h" namespace grpc { namespace testing { namespace { // The same value is defined by the Java client. const std::vector request_stream_sizes = {27182, 8, 1828, 45904}; const std::vector response_stream_sizes = {31415, 9, 2653, 58979}; const int kNumResponseMessages = 2000; const int kResponseMessageSize = 1030; const int kReceiveDelayMilliSeconds = 20; const int kLargeRequestSize = 271828; const int kLargeResponseSize = 314159; void NoopChecks(const InteropClientContextInspector& inspector, const SimpleRequest* request, const SimpleResponse* response) {} void UnaryCompressionChecks(const InteropClientContextInspector& inspector, const SimpleRequest* request, const SimpleResponse* response) { const grpc_compression_algorithm received_compression = inspector.GetCallCompressionAlgorithm(); if (request->response_compressed().value()) { if (received_compression == GRPC_COMPRESS_NONE) { // Requested some compression, got NONE. This is an error. gpr_log(GPR_ERROR, "Failure: Requested compression but got uncompressed response " "from server."); abort(); } GPR_ASSERT(inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS); } else { // Didn't request compression -> make sure the response is uncompressed GPR_ASSERT(!(inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS)); } } } // namespace InteropClient::ServiceStub::ServiceStub( ChannelCreationFunc channel_creation_func, bool new_stub_every_call) : channel_creation_func_(std::move(channel_creation_func)), channel_(channel_creation_func_()), new_stub_every_call_(new_stub_every_call) { // If new_stub_every_call is false, then this is our chance to initialize // stub_. (see Get()) if (!new_stub_every_call) { stub_ = TestService::NewStub(channel_); } } TestService::Stub* InteropClient::ServiceStub::Get() { if (new_stub_every_call_) { stub_ = TestService::NewStub(channel_); } return stub_.get(); } UnimplementedService::Stub* InteropClient::ServiceStub::GetUnimplementedServiceStub() { if (unimplemented_service_stub_ == nullptr) { unimplemented_service_stub_ = UnimplementedService::NewStub(channel_); } return unimplemented_service_stub_.get(); } void InteropClient::ServiceStub::ResetChannel() { channel_ = channel_creation_func_(); if (!new_stub_every_call_) { stub_ = TestService::NewStub(channel_); } } InteropClient::InteropClient(ChannelCreationFunc channel_creation_func, bool new_stub_every_test_case, bool do_not_abort_on_transient_failures) : serviceStub_(std::move(channel_creation_func), new_stub_every_test_case), do_not_abort_on_transient_failures_(do_not_abort_on_transient_failures) {} bool InteropClient::AssertStatusOk(const Status& s, const grpc::string& optional_debug_string) { if (s.ok()) { return true; } // Note: At this point, s.error_code is definitely not StatusCode::OK (we // already checked for s.ok() above). So, the following will call abort() // (unless s.error_code() corresponds to a transient failure and // 'do_not_abort_on_transient_failures' is true) return AssertStatusCode(s, StatusCode::OK, optional_debug_string); } bool InteropClient::AssertStatusCode( const Status& s, StatusCode expected_code, const grpc::string& optional_debug_string) { if (s.error_code() == expected_code) { return true; } gpr_log(GPR_ERROR, "Error status code: %d (expected: %d), message: %s," " debug string: %s", s.error_code(), expected_code, s.error_message().c_str(), optional_debug_string.c_str()); // In case of transient transient/retryable failures (like a broken // connection) we may or may not abort (see TransientFailureOrAbort()) if (s.error_code() == grpc::StatusCode::UNAVAILABLE) { return TransientFailureOrAbort(); } abort(); } bool InteropClient::DoEmpty() { gpr_log(GPR_DEBUG, "Sending an empty rpc..."); Empty request; Empty response; ClientContext context; Status s = serviceStub_.Get()->EmptyCall(&context, request, &response); if (!AssertStatusOk(s, context.debug_error_string())) { return false; } gpr_log(GPR_DEBUG, "Empty rpc done."); return true; } bool InteropClient::PerformLargeUnary(SimpleRequest* request, SimpleResponse* response) { return PerformLargeUnary(request, response, NoopChecks); } bool InteropClient::PerformLargeUnary(SimpleRequest* request, SimpleResponse* response, const CheckerFn& custom_checks_fn) { ClientContext context; InteropClientContextInspector inspector(context); request->set_response_size(kLargeResponseSize); grpc::string payload(kLargeRequestSize, '\0'); request->mutable_payload()->set_body(payload.c_str(), kLargeRequestSize); if (request->has_expect_compressed()) { if (request->expect_compressed().value()) { context.set_compression_algorithm(GRPC_COMPRESS_GZIP); } else { context.set_compression_algorithm(GRPC_COMPRESS_NONE); } } Status s = serviceStub_.Get()->UnaryCall(&context, *request, response); if (!AssertStatusOk(s, context.debug_error_string())) { return false; } custom_checks_fn(inspector, request, response); // Payload related checks. GPR_ASSERT(response->payload().body() == grpc::string(kLargeResponseSize, '\0')); return true; } bool InteropClient::DoComputeEngineCreds( const grpc::string& default_service_account, const grpc::string& oauth_scope) { gpr_log(GPR_DEBUG, "Sending a large unary rpc with compute engine credentials ..."); SimpleRequest request; SimpleResponse response; request.set_fill_username(true); request.set_fill_oauth_scope(true); if (!PerformLargeUnary(&request, &response)) { return false; } gpr_log(GPR_DEBUG, "Got username %s", response.username().c_str()); gpr_log(GPR_DEBUG, "Got oauth_scope %s", response.oauth_scope().c_str()); GPR_ASSERT(!response.username().empty()); GPR_ASSERT(response.username().c_str() == default_service_account); GPR_ASSERT(!response.oauth_scope().empty()); const char* oauth_scope_str = response.oauth_scope().c_str(); GPR_ASSERT(oauth_scope.find(oauth_scope_str) != grpc::string::npos); gpr_log(GPR_DEBUG, "Large unary with compute engine creds done."); return true; } bool InteropClient::DoOauth2AuthToken(const grpc::string& username, const grpc::string& oauth_scope) { gpr_log(GPR_DEBUG, "Sending a unary rpc with raw oauth2 access token credentials ..."); SimpleRequest request; SimpleResponse response; request.set_fill_username(true); request.set_fill_oauth_scope(true); ClientContext context; Status s = serviceStub_.Get()->UnaryCall(&context, request, &response); if (!AssertStatusOk(s, context.debug_error_string())) { return false; } GPR_ASSERT(!response.username().empty()); GPR_ASSERT(!response.oauth_scope().empty()); GPR_ASSERT(username == response.username()); const char* oauth_scope_str = response.oauth_scope().c_str(); GPR_ASSERT(oauth_scope.find(oauth_scope_str) != grpc::string::npos); gpr_log(GPR_DEBUG, "Unary with oauth2 access token credentials done."); return true; } bool InteropClient::DoPerRpcCreds(const grpc::string& json_key) { gpr_log(GPR_DEBUG, "Sending a unary rpc with per-rpc JWT access token ..."); SimpleRequest request; SimpleResponse response; request.set_fill_username(true); ClientContext context; std::chrono::seconds token_lifetime = std::chrono::hours(1); std::shared_ptr creds = ServiceAccountJWTAccessCredentials(json_key, token_lifetime.count()); context.set_credentials(creds); Status s = serviceStub_.Get()->UnaryCall(&context, request, &response); if (!AssertStatusOk(s, context.debug_error_string())) { return false; } GPR_ASSERT(!response.username().empty()); GPR_ASSERT(json_key.find(response.username()) != grpc::string::npos); gpr_log(GPR_DEBUG, "Unary with per-rpc JWT access token done."); return true; } bool InteropClient::DoJwtTokenCreds(const grpc::string& username) { gpr_log(GPR_DEBUG, "Sending a large unary rpc with JWT token credentials ..."); SimpleRequest request; SimpleResponse response; request.set_fill_username(true); if (!PerformLargeUnary(&request, &response)) { return false; } GPR_ASSERT(!response.username().empty()); GPR_ASSERT(username.find(response.username()) != grpc::string::npos); gpr_log(GPR_DEBUG, "Large unary with JWT token creds done."); return true; } bool InteropClient::DoLargeUnary() { gpr_log(GPR_DEBUG, "Sending a large unary rpc..."); SimpleRequest request; SimpleResponse response; if (!PerformLargeUnary(&request, &response)) { return false; } gpr_log(GPR_DEBUG, "Large unary done."); return true; } bool InteropClient::DoClientCompressedUnary() { // Probing for compression-checks support. ClientContext probe_context; SimpleRequest probe_req; SimpleResponse probe_res; probe_context.set_compression_algorithm(GRPC_COMPRESS_NONE); probe_req.mutable_expect_compressed()->set_value(true); // lies! probe_req.set_response_size(kLargeResponseSize); probe_req.mutable_payload()->set_body(grpc::string(kLargeRequestSize, '\0')); gpr_log(GPR_DEBUG, "Sending probe for compressed unary request."); const Status s = serviceStub_.Get()->UnaryCall(&probe_context, probe_req, &probe_res); if (s.error_code() != grpc::StatusCode::INVALID_ARGUMENT) { // The server isn't able to evaluate incoming compression, making the rest // of this test moot. gpr_log(GPR_DEBUG, "Compressed unary request probe failed"); return false; } gpr_log(GPR_DEBUG, "Compressed unary request probe succeeded. Proceeding."); const std::vector compressions = {true, false}; for (size_t i = 0; i < compressions.size(); i++) { char* log_suffix; gpr_asprintf(&log_suffix, "(compression=%s)", compressions[i] ? "true" : "false"); gpr_log(GPR_DEBUG, "Sending compressed unary request %s.", log_suffix); SimpleRequest request; SimpleResponse response; request.mutable_expect_compressed()->set_value(compressions[i]); if (!PerformLargeUnary(&request, &response, UnaryCompressionChecks)) { gpr_log(GPR_ERROR, "Compressed unary request failed %s", log_suffix); gpr_free(log_suffix); return false; } gpr_log(GPR_DEBUG, "Compressed unary request failed %s", log_suffix); gpr_free(log_suffix); } return true; } bool InteropClient::DoServerCompressedUnary() { const std::vector compressions = {true, false}; for (size_t i = 0; i < compressions.size(); i++) { char* log_suffix; gpr_asprintf(&log_suffix, "(compression=%s)", compressions[i] ? "true" : "false"); gpr_log(GPR_DEBUG, "Sending unary request for compressed response %s.", log_suffix); SimpleRequest request; SimpleResponse response; request.mutable_response_compressed()->set_value(compressions[i]); if (!PerformLargeUnary(&request, &response, UnaryCompressionChecks)) { gpr_log(GPR_ERROR, "Request for compressed unary failed %s", log_suffix); gpr_free(log_suffix); return false; } gpr_log(GPR_DEBUG, "Request for compressed unary failed %s", log_suffix); gpr_free(log_suffix); } return true; } // Either abort() (unless do_not_abort_on_transient_failures_ is true) or return // false bool InteropClient::TransientFailureOrAbort() { if (do_not_abort_on_transient_failures_) { return false; } abort(); } bool InteropClient::DoRequestStreaming() { gpr_log(GPR_DEBUG, "Sending request steaming rpc ..."); ClientContext context; StreamingInputCallRequest request; StreamingInputCallResponse response; std::unique_ptr> stream( serviceStub_.Get()->StreamingInputCall(&context, &response)); int aggregated_payload_size = 0; for (size_t i = 0; i < request_stream_sizes.size(); ++i) { Payload* payload = request.mutable_payload(); payload->set_body(grpc::string(request_stream_sizes[i], '\0')); if (!stream->Write(request)) { gpr_log(GPR_ERROR, "DoRequestStreaming(): stream->Write() failed"); return TransientFailureOrAbort(); } aggregated_payload_size += request_stream_sizes[i]; } GPR_ASSERT(stream->WritesDone()); Status s = stream->Finish(); if (!AssertStatusOk(s, context.debug_error_string())) { return false; } GPR_ASSERT(response.aggregated_payload_size() == aggregated_payload_size); return true; } bool InteropClient::DoResponseStreaming() { gpr_log(GPR_DEBUG, "Receiving response streaming rpc ..."); ClientContext context; StreamingOutputCallRequest request; for (unsigned int i = 0; i < response_stream_sizes.size(); ++i) { ResponseParameters* response_parameter = request.add_response_parameters(); response_parameter->set_size(response_stream_sizes[i]); } StreamingOutputCallResponse response; std::unique_ptr> stream( serviceStub_.Get()->StreamingOutputCall(&context, request)); unsigned int i = 0; while (stream->Read(&response)) { GPR_ASSERT(response.payload().body() == grpc::string(response_stream_sizes[i], '\0')); ++i; } if (i < response_stream_sizes.size()) { // stream->Read() failed before reading all the expected messages. This is // most likely due to connection failure. gpr_log(GPR_ERROR, "DoResponseStreaming(): Read fewer streams (%d) than " "response_stream_sizes.size() (%" PRIuPTR ")", i, response_stream_sizes.size()); return TransientFailureOrAbort(); } Status s = stream->Finish(); if (!AssertStatusOk(s, context.debug_error_string())) { return false; } gpr_log(GPR_DEBUG, "Response streaming done."); return true; } bool InteropClient::DoClientCompressedStreaming() { // Probing for compression-checks support. ClientContext probe_context; StreamingInputCallRequest probe_req; StreamingInputCallResponse probe_res; probe_context.set_compression_algorithm(GRPC_COMPRESS_NONE); probe_req.mutable_expect_compressed()->set_value(true); // lies! probe_req.mutable_payload()->set_body(grpc::string(27182, '\0')); gpr_log(GPR_DEBUG, "Sending probe for compressed streaming request."); std::unique_ptr> probe_stream( serviceStub_.Get()->StreamingInputCall(&probe_context, &probe_res)); if (!probe_stream->Write(probe_req)) { gpr_log(GPR_ERROR, "%s(): stream->Write() failed", __func__); return TransientFailureOrAbort(); } Status s = probe_stream->Finish(); if (s.error_code() != grpc::StatusCode::INVALID_ARGUMENT) { // The server isn't able to evaluate incoming compression, making the rest // of this test moot. gpr_log(GPR_DEBUG, "Compressed streaming request probe failed"); return false; } gpr_log(GPR_DEBUG, "Compressed streaming request probe succeeded. Proceeding."); ClientContext context; StreamingInputCallRequest request; StreamingInputCallResponse response; context.set_compression_algorithm(GRPC_COMPRESS_GZIP); std::unique_ptr> stream( serviceStub_.Get()->StreamingInputCall(&context, &response)); request.mutable_payload()->set_body(grpc::string(27182, '\0')); request.mutable_expect_compressed()->set_value(true); gpr_log(GPR_DEBUG, "Sending streaming request with compression enabled"); if (!stream->Write(request)) { gpr_log(GPR_ERROR, "%s(): stream->Write() failed", __func__); return TransientFailureOrAbort(); } WriteOptions wopts; wopts.set_no_compression(); request.mutable_payload()->set_body(grpc::string(45904, '\0')); request.mutable_expect_compressed()->set_value(false); gpr_log(GPR_DEBUG, "Sending streaming request with compression disabled"); if (!stream->Write(request, wopts)) { gpr_log(GPR_ERROR, "%s(): stream->Write() failed", __func__); return TransientFailureOrAbort(); } GPR_ASSERT(stream->WritesDone()); s = stream->Finish(); if (!AssertStatusOk(s, context.debug_error_string())) { return false; } return true; } bool InteropClient::DoServerCompressedStreaming() { const std::vector compressions = {true, false}; const std::vector sizes = {31415, 92653}; ClientContext context; InteropClientContextInspector inspector(context); StreamingOutputCallRequest request; GPR_ASSERT(compressions.size() == sizes.size()); for (size_t i = 0; i < sizes.size(); i++) { char* log_suffix; gpr_asprintf(&log_suffix, "(compression=%s; size=%d)", compressions[i] ? "true" : "false", sizes[i]); gpr_log(GPR_DEBUG, "Sending request streaming rpc %s.", log_suffix); gpr_free(log_suffix); ResponseParameters* const response_parameter = request.add_response_parameters(); response_parameter->mutable_compressed()->set_value(compressions[i]); response_parameter->set_size(sizes[i]); } std::unique_ptr> stream( serviceStub_.Get()->StreamingOutputCall(&context, request)); size_t k = 0; StreamingOutputCallResponse response; while (stream->Read(&response)) { // Payload size checks. GPR_ASSERT(response.payload().body() == grpc::string(request.response_parameters(k).size(), '\0')); // Compression checks. GPR_ASSERT(request.response_parameters(k).has_compressed()); if (request.response_parameters(k).compressed().value()) { GPR_ASSERT(inspector.GetCallCompressionAlgorithm() > GRPC_COMPRESS_NONE); GPR_ASSERT(inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS); } else { // requested *no* compression. GPR_ASSERT(!(inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS)); } ++k; } if (k < sizes.size()) { // stream->Read() failed before reading all the expected messages. This // is most likely due to a connection failure. gpr_log(GPR_ERROR, "%s(): Responses read (k=%" PRIuPTR ") is less than the expected number of messages (%" PRIuPTR ").", __func__, k, sizes.size()); return TransientFailureOrAbort(); } Status s = stream->Finish(); if (!AssertStatusOk(s, context.debug_error_string())) { return false; } return true; } bool InteropClient::DoResponseStreamingWithSlowConsumer() { gpr_log(GPR_DEBUG, "Receiving response streaming rpc with slow consumer ..."); ClientContext context; StreamingOutputCallRequest request; for (int i = 0; i < kNumResponseMessages; ++i) { ResponseParameters* response_parameter = request.add_response_parameters(); response_parameter->set_size(kResponseMessageSize); } StreamingOutputCallResponse response; std::unique_ptr> stream( serviceStub_.Get()->StreamingOutputCall(&context, request)); int i = 0; while (stream->Read(&response)) { GPR_ASSERT(response.payload().body() == grpc::string(kResponseMessageSize, '\0')); gpr_log(GPR_DEBUG, "received message %d", i); gpr_sleep_until(gpr_time_add( gpr_now(GPR_CLOCK_REALTIME), gpr_time_from_millis(kReceiveDelayMilliSeconds, GPR_TIMESPAN))); ++i; } if (i < kNumResponseMessages) { gpr_log(GPR_ERROR, "DoResponseStreamingWithSlowConsumer(): Responses read (i=%d) is " "less than the expected messages (i.e kNumResponseMessages = %d)", i, kNumResponseMessages); return TransientFailureOrAbort(); } Status s = stream->Finish(); if (!AssertStatusOk(s, context.debug_error_string())) { return false; } gpr_log(GPR_DEBUG, "Response streaming done."); return true; } bool InteropClient::DoHalfDuplex() { gpr_log(GPR_DEBUG, "Sending half-duplex streaming rpc ..."); ClientContext context; std::unique_ptr> stream(serviceStub_.Get()->HalfDuplexCall(&context)); StreamingOutputCallRequest request; ResponseParameters* response_parameter = request.add_response_parameters(); for (unsigned int i = 0; i < response_stream_sizes.size(); ++i) { response_parameter->set_size(response_stream_sizes[i]); if (!stream->Write(request)) { gpr_log(GPR_ERROR, "DoHalfDuplex(): stream->Write() failed. i=%d", i); return TransientFailureOrAbort(); } } stream->WritesDone(); unsigned int i = 0; StreamingOutputCallResponse response; while (stream->Read(&response)) { GPR_ASSERT(response.payload().body() == grpc::string(response_stream_sizes[i], '\0')); ++i; } if (i < response_stream_sizes.size()) { // stream->Read() failed before reading all the expected messages. This is // most likely due to a connection failure gpr_log(GPR_ERROR, "DoHalfDuplex(): Responses read (i=%d) are less than the expected " "number of messages response_stream_sizes.size() (%" PRIuPTR ")", i, response_stream_sizes.size()); return TransientFailureOrAbort(); } Status s = stream->Finish(); if (!AssertStatusOk(s, context.debug_error_string())) { return false; } gpr_log(GPR_DEBUG, "Half-duplex streaming rpc done."); return true; } bool InteropClient::DoPingPong() { gpr_log(GPR_DEBUG, "Sending Ping Pong streaming rpc ..."); ClientContext context; std::unique_ptr> stream(serviceStub_.Get()->FullDuplexCall(&context)); StreamingOutputCallRequest request; ResponseParameters* response_parameter = request.add_response_parameters(); Payload* payload = request.mutable_payload(); StreamingOutputCallResponse response; for (unsigned int i = 0; i < request_stream_sizes.size(); ++i) { response_parameter->set_size(response_stream_sizes[i]); payload->set_body(grpc::string(request_stream_sizes[i], '\0')); if (!stream->Write(request)) { gpr_log(GPR_ERROR, "DoPingPong(): stream->Write() failed. i: %d", i); return TransientFailureOrAbort(); } if (!stream->Read(&response)) { gpr_log(GPR_ERROR, "DoPingPong(): stream->Read() failed. i:%d", i); return TransientFailureOrAbort(); } GPR_ASSERT(response.payload().body() == grpc::string(response_stream_sizes[i], '\0')); } stream->WritesDone(); GPR_ASSERT(!stream->Read(&response)); Status s = stream->Finish(); if (!AssertStatusOk(s, context.debug_error_string())) { return false; } gpr_log(GPR_DEBUG, "Ping pong streaming done."); return true; } bool InteropClient::DoCancelAfterBegin() { gpr_log(GPR_DEBUG, "Sending request streaming rpc ..."); ClientContext context; StreamingInputCallRequest request; StreamingInputCallResponse response; std::unique_ptr> stream( serviceStub_.Get()->StreamingInputCall(&context, &response)); gpr_log(GPR_DEBUG, "Trying to cancel..."); context.TryCancel(); Status s = stream->Finish(); if (!AssertStatusCode(s, StatusCode::CANCELLED, context.debug_error_string())) { return false; } gpr_log(GPR_DEBUG, "Canceling streaming done."); return true; } bool InteropClient::DoCancelAfterFirstResponse() { gpr_log(GPR_DEBUG, "Sending Ping Pong streaming rpc ..."); ClientContext context; std::unique_ptr> stream(serviceStub_.Get()->FullDuplexCall(&context)); StreamingOutputCallRequest request; ResponseParameters* response_parameter = request.add_response_parameters(); response_parameter->set_size(31415); request.mutable_payload()->set_body(grpc::string(27182, '\0')); StreamingOutputCallResponse response; if (!stream->Write(request)) { gpr_log(GPR_ERROR, "DoCancelAfterFirstResponse(): stream->Write() failed"); return TransientFailureOrAbort(); } if (!stream->Read(&response)) { gpr_log(GPR_ERROR, "DoCancelAfterFirstResponse(): stream->Read failed"); return TransientFailureOrAbort(); } GPR_ASSERT(response.payload().body() == grpc::string(31415, '\0')); gpr_log(GPR_DEBUG, "Trying to cancel..."); context.TryCancel(); Status s = stream->Finish(); gpr_log(GPR_DEBUG, "Canceling pingpong streaming done."); return true; } bool InteropClient::DoTimeoutOnSleepingServer() { gpr_log(GPR_DEBUG, "Sending Ping Pong streaming rpc with a short deadline..."); ClientContext context; std::chrono::system_clock::time_point deadline = std::chrono::system_clock::now() + std::chrono::milliseconds(1); context.set_deadline(deadline); std::unique_ptr> stream(serviceStub_.Get()->FullDuplexCall(&context)); StreamingOutputCallRequest request; request.mutable_payload()->set_body(grpc::string(27182, '\0')); stream->Write(request); Status s = stream->Finish(); if (!AssertStatusCode(s, StatusCode::DEADLINE_EXCEEDED, context.debug_error_string())) { return false; } gpr_log(GPR_DEBUG, "Pingpong streaming timeout done."); return true; } bool InteropClient::DoEmptyStream() { gpr_log(GPR_DEBUG, "Starting empty_stream."); ClientContext context; std::unique_ptr> stream(serviceStub_.Get()->FullDuplexCall(&context)); stream->WritesDone(); StreamingOutputCallResponse response; GPR_ASSERT(stream->Read(&response) == false); Status s = stream->Finish(); if (!AssertStatusOk(s, context.debug_error_string())) { return false; } gpr_log(GPR_DEBUG, "empty_stream done."); return true; } bool InteropClient::DoStatusWithMessage() { gpr_log(GPR_DEBUG, "Sending RPC with a request for status code 2 and message"); const grpc::StatusCode test_code = grpc::StatusCode::UNKNOWN; const grpc::string test_msg = "This is a test message"; // Test UnaryCall. ClientContext context; SimpleRequest request; SimpleResponse response; EchoStatus* requested_status = request.mutable_response_status(); requested_status->set_code(test_code); requested_status->set_message(test_msg); Status s = serviceStub_.Get()->UnaryCall(&context, request, &response); if (!AssertStatusCode(s, grpc::StatusCode::UNKNOWN, context.debug_error_string())) { return false; } GPR_ASSERT(s.error_message() == test_msg); // Test FullDuplexCall. ClientContext stream_context; std::shared_ptr> stream(serviceStub_.Get()->FullDuplexCall(&stream_context)); StreamingOutputCallRequest streaming_request; requested_status = streaming_request.mutable_response_status(); requested_status->set_code(test_code); requested_status->set_message(test_msg); stream->Write(streaming_request); stream->WritesDone(); StreamingOutputCallResponse streaming_response; while (stream->Read(&streaming_response)) ; s = stream->Finish(); if (!AssertStatusCode(s, grpc::StatusCode::UNKNOWN, context.debug_error_string())) { return false; } GPR_ASSERT(s.error_message() == test_msg); gpr_log(GPR_DEBUG, "Done testing Status and Message"); return true; } bool InteropClient::DoCacheableUnary() { gpr_log(GPR_DEBUG, "Sending RPC with cacheable response"); // Create request with current timestamp gpr_timespec ts = gpr_now(GPR_CLOCK_PRECISE); std::string timestamp = std::to_string(static_cast(ts.tv_nsec)); SimpleRequest request; request.mutable_payload()->set_body(timestamp.c_str(), timestamp.size()); // Request 1 ClientContext context1; SimpleResponse response1; context1.set_cacheable(true); // Add fake user IP since some proxy's (GFE) won't cache requests from // localhost. context1.AddMetadata("x-user-ip", "1.2.3.4"); Status s1 = serviceStub_.Get()->CacheableUnaryCall(&context1, request, &response1); if (!AssertStatusOk(s1, context1.debug_error_string())) { return false; } gpr_log(GPR_DEBUG, "response 1 payload: %s", response1.payload().body().c_str()); // Request 2 ClientContext context2; SimpleResponse response2; context2.set_cacheable(true); context2.AddMetadata("x-user-ip", "1.2.3.4"); Status s2 = serviceStub_.Get()->CacheableUnaryCall(&context2, request, &response2); if (!AssertStatusOk(s2, context2.debug_error_string())) { return false; } gpr_log(GPR_DEBUG, "response 2 payload: %s", response2.payload().body().c_str()); // Check that the body is same for both requests. It will be the same if the // second response is a cached copy of the first response GPR_ASSERT(response2.payload().body() == response1.payload().body()); // Request 3 // Modify the request body so it will not get a cache hit ts = gpr_now(GPR_CLOCK_PRECISE); timestamp = std::to_string(static_cast(ts.tv_nsec)); SimpleRequest request1; request1.mutable_payload()->set_body(timestamp.c_str(), timestamp.size()); ClientContext context3; SimpleResponse response3; context3.set_cacheable(true); context3.AddMetadata("x-user-ip", "1.2.3.4"); Status s3 = serviceStub_.Get()->CacheableUnaryCall(&context3, request1, &response3); if (!AssertStatusOk(s3, context3.debug_error_string())) { return false; } gpr_log(GPR_DEBUG, "response 3 payload: %s", response3.payload().body().c_str()); // Check that the response is different from the previous response. GPR_ASSERT(response3.payload().body() != response1.payload().body()); return true; } bool InteropClient::DoCustomMetadata() { const grpc::string kEchoInitialMetadataKey("x-grpc-test-echo-initial"); const grpc::string kInitialMetadataValue("test_initial_metadata_value"); const grpc::string kEchoTrailingBinMetadataKey( "x-grpc-test-echo-trailing-bin"); const grpc::string kTrailingBinValue("\x0a\x0b\x0a\x0b\x0a\x0b"); ; { gpr_log(GPR_DEBUG, "Sending RPC with custom metadata"); ClientContext context; context.AddMetadata(kEchoInitialMetadataKey, kInitialMetadataValue); context.AddMetadata(kEchoTrailingBinMetadataKey, kTrailingBinValue); SimpleRequest request; SimpleResponse response; request.set_response_size(kLargeResponseSize); grpc::string payload(kLargeRequestSize, '\0'); request.mutable_payload()->set_body(payload.c_str(), kLargeRequestSize); Status s = serviceStub_.Get()->UnaryCall(&context, request, &response); if (!AssertStatusOk(s, context.debug_error_string())) { return false; } const auto& server_initial_metadata = context.GetServerInitialMetadata(); auto iter = server_initial_metadata.find(kEchoInitialMetadataKey); GPR_ASSERT(iter != server_initial_metadata.end()); GPR_ASSERT(iter->second == kInitialMetadataValue); const auto& server_trailing_metadata = context.GetServerTrailingMetadata(); iter = server_trailing_metadata.find(kEchoTrailingBinMetadataKey); GPR_ASSERT(iter != server_trailing_metadata.end()); GPR_ASSERT(grpc::string(iter->second.begin(), iter->second.end()) == kTrailingBinValue); gpr_log(GPR_DEBUG, "Done testing RPC with custom metadata"); } { gpr_log(GPR_DEBUG, "Sending stream with custom metadata"); ClientContext context; context.AddMetadata(kEchoInitialMetadataKey, kInitialMetadataValue); context.AddMetadata(kEchoTrailingBinMetadataKey, kTrailingBinValue); std::unique_ptr> stream(serviceStub_.Get()->FullDuplexCall(&context)); StreamingOutputCallRequest request; ResponseParameters* response_parameter = request.add_response_parameters(); response_parameter->set_size(kLargeResponseSize); grpc::string payload(kLargeRequestSize, '\0'); request.mutable_payload()->set_body(payload.c_str(), kLargeRequestSize); StreamingOutputCallResponse response; if (!stream->Write(request)) { gpr_log(GPR_ERROR, "DoCustomMetadata(): stream->Write() failed"); return TransientFailureOrAbort(); } stream->WritesDone(); if (!stream->Read(&response)) { gpr_log(GPR_ERROR, "DoCustomMetadata(): stream->Read() failed"); return TransientFailureOrAbort(); } GPR_ASSERT(response.payload().body() == grpc::string(kLargeResponseSize, '\0')); GPR_ASSERT(!stream->Read(&response)); Status s = stream->Finish(); if (!AssertStatusOk(s, context.debug_error_string())) { return false; } const auto& server_initial_metadata = context.GetServerInitialMetadata(); auto iter = server_initial_metadata.find(kEchoInitialMetadataKey); GPR_ASSERT(iter != server_initial_metadata.end()); GPR_ASSERT(iter->second == kInitialMetadataValue); const auto& server_trailing_metadata = context.GetServerTrailingMetadata(); iter = server_trailing_metadata.find(kEchoTrailingBinMetadataKey); GPR_ASSERT(iter != server_trailing_metadata.end()); GPR_ASSERT(grpc::string(iter->second.begin(), iter->second.end()) == kTrailingBinValue); gpr_log(GPR_DEBUG, "Done testing stream with custom metadata"); } return true; } bool InteropClient::DoRpcSoakTest(int32_t soak_iterations) { gpr_log(GPR_DEBUG, "Sending %d RPCs...", soak_iterations); GPR_ASSERT(soak_iterations > 0); SimpleRequest request; SimpleResponse response; for (int i = 0; i < soak_iterations; ++i) { if (!PerformLargeUnary(&request, &response)) { gpr_log(GPR_ERROR, "rpc_soak test failed on iteration %d", i); return false; } } gpr_log(GPR_DEBUG, "rpc_soak test done."); return true; } bool InteropClient::DoChannelSoakTest(int32_t soak_iterations) { gpr_log(GPR_DEBUG, "Sending %d RPCs, tearing down the channel each time...", soak_iterations); GPR_ASSERT(soak_iterations > 0); SimpleRequest request; SimpleResponse response; for (int i = 0; i < soak_iterations; ++i) { serviceStub_.ResetChannel(); if (!PerformLargeUnary(&request, &response)) { gpr_log(GPR_ERROR, "channel_soak test failed on iteration %d", i); return false; } } gpr_log(GPR_DEBUG, "channel_soak test done."); return true; } bool InteropClient::DoLongLivedChannelTest(int32_t soak_iterations, int32_t iteration_interval) { gpr_log(GPR_DEBUG, "Sending %d RPCs...", soak_iterations); GPR_ASSERT(soak_iterations > 0); GPR_ASSERT(iteration_interval > 0); SimpleRequest request; SimpleResponse response; int num_failures = 0; for (int i = 0; i < soak_iterations; ++i) { gpr_log(GPR_DEBUG, "Sending RPC number %d...", i); if (!PerformLargeUnary(&request, &response)) { gpr_log(GPR_ERROR, "Iteration %d failed.", i); num_failures++; } gpr_sleep_until( gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), gpr_time_from_seconds(iteration_interval, GPR_TIMESPAN))); } if (num_failures == 0) { gpr_log(GPR_DEBUG, "long_lived_channel test done."); return true; } else { gpr_log(GPR_DEBUG, "long_lived_channel test failed with %d rpc failures.", num_failures); return false; } } bool InteropClient::DoUnimplementedService() { gpr_log(GPR_DEBUG, "Sending a request for an unimplemented service..."); Empty request; Empty response; ClientContext context; UnimplementedService::Stub* stub = serviceStub_.GetUnimplementedServiceStub(); Status s = stub->UnimplementedCall(&context, request, &response); if (!AssertStatusCode(s, StatusCode::UNIMPLEMENTED, context.debug_error_string())) { return false; } gpr_log(GPR_DEBUG, "unimplemented service done."); return true; } bool InteropClient::DoUnimplementedMethod() { gpr_log(GPR_DEBUG, "Sending a request for an unimplemented rpc..."); Empty request; Empty response; ClientContext context; Status s = serviceStub_.Get()->UnimplementedCall(&context, request, &response); if (!AssertStatusCode(s, StatusCode::UNIMPLEMENTED, context.debug_error_string())) { return false; } gpr_log(GPR_DEBUG, "unimplemented rpc done."); return true; } } // namespace testing } // namespace grpc