From 74686ce7c4354020b4e48ba067ac7635d0e4edb2 Mon Sep 17 00:00:00 2001 From: David Garcia Quintas Date: Thu, 9 Jun 2016 15:33:33 -0700 Subject: Completed interop spec, as well as code for c++ --- test/cpp/interop/client.cc | 63 +++++-- test/cpp/interop/interop_client.cc | 99 ++++++----- test/cpp/interop/interop_client.h | 6 +- test/cpp/interop/interop_server.cc | 328 +++++++++++++++++++++++++++++++++++++ test/cpp/interop/rnd.dat | Bin 524288 -> 0 bytes test/cpp/interop/server_helper.cc | 4 + test/cpp/interop/server_helper.h | 1 + test/cpp/interop/server_main.cc | 291 -------------------------------- 8 files changed, 446 insertions(+), 346 deletions(-) create mode 100644 test/cpp/interop/interop_server.cc delete mode 100644 test/cpp/interop/rnd.dat delete mode 100644 test/cpp/interop/server_main.cc (limited to 'test/cpp/interop') diff --git a/test/cpp/interop/client.cc b/test/cpp/interop/client.cc index 7727824979..c7d081100e 100644 --- a/test/cpp/interop/client.cc +++ b/test/cpp/interop/client.cc @@ -40,7 +40,9 @@ #include #include #include +#include +#include "src/core/lib/support/string.h" #include "test/cpp/interop/client_helper.h" #include "test/cpp/interop/interop_client.h" #include "test/cpp/util/test_config.h" @@ -55,10 +57,14 @@ DEFINE_string(test_case, "large_unary", "Configure different test cases. Valid options are: " "empty_unary : empty (zero bytes) request and response; " "large_unary : single request and (large) response; " - "large_compressed_unary : single request and compressed (large) " - "response; " + + "client_compressed_unary : single compressed request; " + "server_compressed_unary : single compressed response; " + "client_streaming : request streaming with single response; " "server_streaming : single request with response streaming; " + "client_compressed_streaming : compressed request streaming with " + "single response; " "server_compressed_streaming : single request with compressed " "response streaming; " "slow_consumer : single request with response; " @@ -104,14 +110,18 @@ int main(int argc, char** argv) { client.DoEmpty(); } else if (FLAGS_test_case == "large_unary") { client.DoLargeUnary(); - } else if (FLAGS_test_case == "large_compressed_unary") { - client.DoLargeCompressedUnary(); + } else if (FLAGS_test_case == "server_compressed_unary") { + client.DoServerCompressedUnary(); + } else if (FLAGS_test_case == "client_compressed_unary") { + client.DoClientCompressedUnary(); } else if (FLAGS_test_case == "client_streaming") { client.DoRequestStreaming(); } else if (FLAGS_test_case == "server_streaming") { client.DoResponseStreaming(); } else if (FLAGS_test_case == "server_compressed_streaming") { - client.DoResponseCompressedStreaming(); + client.DoServerCompressedStreaming(); + } else if (FLAGS_test_case == "client_compressed_streaming") { + client.DoClientCompressedStreaming(); } else if (FLAGS_test_case == "slow_consumer") { client.DoResponseStreamingWithSlowConsumer(); } else if (FLAGS_test_case == "half_duplex") { @@ -144,9 +154,12 @@ int main(int argc, char** argv) { } else if (FLAGS_test_case == "all") { client.DoEmpty(); client.DoLargeUnary(); + client.DoClientCompressedUnary(); + client.DoServerCompressedUnary(); client.DoRequestStreaming(); client.DoResponseStreaming(); - client.DoResponseCompressedStreaming(); + client.DoClientCompressedStreaming(); + client.DoServerCompressedStreaming(); client.DoHalfDuplex(); client.DoPingPong(); client.DoCancelAfterBegin(); @@ -165,14 +178,36 @@ int main(int argc, char** argv) { } // compute_engine_creds only runs in GCE. } else { - gpr_log( - GPR_ERROR, - "Unsupported test case %s. Valid options are all|empty_unary|" - "large_unary|large_compressed_unary|client_streaming|server_streaming|" - "server_compressed_streaming|half_duplex|ping_pong|cancel_after_begin|" - "cancel_after_first_response|timeout_on_sleeping_server|empty_stream|" - "compute_engine_creds|jwt_token_creds|oauth2_auth_token|per_rpc_creds", - "status_code_and_message|custom_metadata", FLAGS_test_case.c_str()); + const char* testcases[] = + { "all", + "cancel_after_begin", + "cancel_after_first_response", + "client_compressed_streaming", + "client_compressed_unary", + "client_streaming", + "compute_engine_creds", + "custom_metadata", + "empty_stream", + "empty_unary", + "half_duplex", + "jwt_token_creds", + "large_unary", + "oauth2_auth_token", + "oauth2_auth_token", + "per_rpc_creds", + "per_rpc_creds", + "ping_pong", + "server_compressed_streaming", + "server_compressed_unary", + "server_streaming", + "status_code_and_message", + "timeout_on_sleeping_server"}; + char* joined_testcases = + gpr_strjoin_sep(testcases, GPR_ARRAY_SIZE(testcases), "\n", NULL); + + gpr_log(GPR_ERROR, "Unsupported test case %s. Valid options are\n%s", + FLAGS_test_case.c_str(), joined_testcases); + gpr_free(joined_testcases); ret = 1; } diff --git a/test/cpp/interop/interop_client.cc b/test/cpp/interop/interop_client.cc index 7705bb1592..e5d3751402 100644 --- a/test/cpp/interop/interop_client.cc +++ b/test/cpp/interop/interop_client.cc @@ -73,23 +73,22 @@ void CompressionChecks(const InteropClientContextInspector& inspector, const SimpleResponse* response) { const grpc_compression_algorithm received_compression = inspector.GetCallCompressionAlgorithm(); - if (request->request_compressed_response() && - received_compression == GRPC_COMPRESS_NONE) { - if (request->request_compressed_response() && - received_compression == GRPC_COMPRESS_NONE) { + if (request->request_compressed_response()) { + 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(); } - } - if (!request->request_compressed_response()) { + if (request->response_type() == PayloadType::COMPRESSABLE) { + // requested compression and compressable response => results should + // always be compressed. + 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)); - } else if (request->response_type() == PayloadType::COMPRESSABLE) { - // requested compression and compressable response => results should always - // be compressed. - GPR_ASSERT(inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS); } } } // namespace @@ -211,17 +210,6 @@ bool InteropClient::PerformLargeUnary(SimpleRequest* request, GPR_ASSERT(response->payload().body() == grpc::string(kLargeResponseSize, '\0')); break; - case PayloadType::UNCOMPRESSABLE: { - // We don't really check anything: We can't assert that the payload is - // uncompressed because it's the server's prerogative to decide on that, - // and different implementations decide differently (ie, Java always - // compresses when requested to do so, whereas C core throws away the - // compressed payload if the output is larger than the input). - // In addition, we don't compare the actual random bytes received because - // asserting that data is sent/received properly isn't the purpose of this - // test. Moreover, different implementations are also free to use - // different sets of random bytes. - } break; default: GPR_ASSERT(false); } @@ -336,9 +324,39 @@ bool InteropClient::DoLargeUnary() { return true; } -bool InteropClient::DoLargeCompressedUnary() { +bool InteropClient::DoClientCompressedUnary() { + const bool expect_compression[] = {false, true}; + const PayloadType payload_types[] = {COMPRESSABLE}; + for (size_t i = 0; i < GPR_ARRAY_SIZE(payload_types); i++) { + for (size_t j = 0; j < GPR_ARRAY_SIZE(expect_compression); j++) { + char* log_suffix; + gpr_asprintf(&log_suffix, "(compression=%s; payload=%s)", + expect_compression[j] ? "true" : "false", + PayloadType_Name(payload_types[i]).c_str()); + + gpr_log(GPR_DEBUG, "Sending compressed unary request %s.", log_suffix); + SimpleRequest request; + SimpleResponse response; + request.set_response_type(payload_types[i]); + request.set_expect_compressed_request(expect_compression[j]); + + if (!PerformLargeUnary(&request, &response, CompressionChecks)) { + 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 bool request_compression[] = {false, true}; - const PayloadType payload_types[] = {COMPRESSABLE, UNCOMPRESSABLE}; + const PayloadType payload_types[] = {COMPRESSABLE}; for (size_t i = 0; i < GPR_ARRAY_SIZE(payload_types); i++) { for (size_t j = 0; j < GPR_ARRAY_SIZE(request_compression); j++) { char* log_suffix; @@ -346,7 +364,7 @@ bool InteropClient::DoLargeCompressedUnary() { request_compression[j] ? "true" : "false", PayloadType_Name(payload_types[i]).c_str()); - gpr_log(GPR_DEBUG, "Sending a large compressed unary rpc %s.", + gpr_log(GPR_DEBUG, "Sending unary request for compressed response %s.", log_suffix); SimpleRequest request; SimpleResponse response; @@ -354,12 +372,13 @@ bool InteropClient::DoLargeCompressedUnary() { request.set_request_compressed_response(request_compression[j]); if (!PerformLargeUnary(&request, &response, CompressionChecks)) { - gpr_log(GPR_ERROR, "Large compressed unary failed %s", log_suffix); + gpr_log(GPR_ERROR, "Request for compressed unary failed %s", + log_suffix); gpr_free(log_suffix); return false; } - gpr_log(GPR_DEBUG, "Large compressed unary done %s.", log_suffix); + gpr_log(GPR_DEBUG, "Request for compressed unary failed %s", log_suffix); gpr_free(log_suffix); } } @@ -447,9 +466,16 @@ bool InteropClient::DoResponseStreaming() { return true; } -bool InteropClient::DoResponseCompressedStreaming() { +bool InteropClient::DoClientCompressedStreaming() { + // XXX + return false; +} + +bool InteropClient::DoServerCompressedStreaming() { const bool request_compression[] = {false, true}; - const PayloadType payload_types[] = {COMPRESSABLE, UNCOMPRESSABLE}; + const PayloadType payload_types[] = {COMPRESSABLE}; + const std::vector response_stream_sizes = {31415, 58979}; + for (size_t i = 0; i < GPR_ARRAY_SIZE(payload_types); i++) { for (size_t j = 0; j < GPR_ARRAY_SIZE(request_compression); j++) { ClientContext context; @@ -466,11 +492,10 @@ bool InteropClient::DoResponseCompressedStreaming() { request.set_response_type(payload_types[i]); request.set_request_compressed_response(request_compression[j]); - for (size_t k = 0; k < response_stream_sizes.size() / 2; ++k) { + for (size_t k = 0; k < response_stream_sizes.size(); ++k) { ResponseParameters* response_parameter = request.add_response_parameters(); - response_parameter->set_size(response_stream_sizes[k] + - response_stream_sizes[k + 1]); + response_parameter->set_size(response_stream_sizes[k]); } StreamingOutputCallResponse response; @@ -484,11 +509,7 @@ bool InteropClient::DoResponseCompressedStreaming() { switch (response.payload().type()) { case PayloadType::COMPRESSABLE: GPR_ASSERT(response.payload().body() == - grpc::string(response_stream_sizes[k] + - response_stream_sizes[k + 1], - '\0')); - break; - case PayloadType::UNCOMPRESSABLE: + grpc::string(response_stream_sizes[k], '\0')); break; default: GPR_ASSERT(false); @@ -516,14 +537,14 @@ bool InteropClient::DoResponseCompressedStreaming() { gpr_log(GPR_DEBUG, "Response streaming done %s.", log_suffix); gpr_free(log_suffix); - if (k < response_stream_sizes.size() / 2) { + if (k < 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, - "DoResponseCompressedStreaming(): Responses read (k=%d) is " + "DoServerCompressedStreaming(): Responses read (k=%d) is " "less than the expected messages (i.e " "response_stream_sizes.size()/2 (%d)). (i=%d, j=%d)", - k, response_stream_sizes.size() / 2, i, j); + k, response_stream_sizes.size(), i, j); return TransientFailureOrAbort(); } diff --git a/test/cpp/interop/interop_client.h b/test/cpp/interop/interop_client.h index ae75762bb8..ea44986fbc 100644 --- a/test/cpp/interop/interop_client.h +++ b/test/cpp/interop/interop_client.h @@ -64,12 +64,14 @@ class InteropClient { bool DoEmpty(); bool DoLargeUnary(); - bool DoLargeCompressedUnary(); + bool DoServerCompressedUnary(); + bool DoClientCompressedUnary(); bool DoPingPong(); bool DoHalfDuplex(); bool DoRequestStreaming(); bool DoResponseStreaming(); - bool DoResponseCompressedStreaming(); + bool DoServerCompressedStreaming(); + bool DoClientCompressedStreaming(); bool DoResponseStreamingWithSlowConsumer(); bool DoCancelAfterBegin(); bool DoCancelAfterFirstResponse(); diff --git a/test/cpp/interop/interop_server.cc b/test/cpp/interop/interop_server.cc new file mode 100644 index 0000000000..b328f478fa --- /dev/null +++ b/test/cpp/interop/interop_server.cc @@ -0,0 +1,328 @@ +/* + * + * Copyright 2015, 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 +#include + +#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.grpc.pb.h" +#include "src/proto/grpc/testing/messages.grpc.pb.h" +#include "src/proto/grpc/testing/test.grpc.pb.h" +#include "test/cpp/interop/server_helper.h" +#include "test/cpp/util/test_config.h" + +DEFINE_bool(use_tls, false, "Whether to use tls."); +DEFINE_int32(port, 0, "Server port."); + +using grpc::Server; +using grpc::ServerBuilder; +using grpc::ServerContext; +using grpc::ServerCredentials; +using grpc::ServerReader; +using grpc::ServerReaderWriter; +using grpc::ServerWriter; +using grpc::SslServerCredentialsOptions; +using grpc::testing::InteropServerContextInspector; +using grpc::testing::Payload; +using grpc::testing::PayloadType; +using grpc::testing::SimpleRequest; +using grpc::testing::SimpleResponse; +using grpc::testing::StreamingInputCallRequest; +using grpc::testing::StreamingInputCallResponse; +using grpc::testing::StreamingOutputCallRequest; +using grpc::testing::StreamingOutputCallResponse; +using grpc::testing::TestService; +using grpc::Status; + +static bool got_sigint = false; + +const char kEchoInitialMetadataKey[] = "x-grpc-test-echo-initial"; +const char kEchoTrailingBinMetadataKey[] = "x-grpc-test-echo-trailing-bin"; +const char kEchoUserAgentKey[] = "x-grpc-test-echo-useragent"; + +void MaybeEchoMetadata(ServerContext* context) { + const auto& client_metadata = context->client_metadata(); + GPR_ASSERT(client_metadata.count(kEchoInitialMetadataKey) <= 1); + GPR_ASSERT(client_metadata.count(kEchoTrailingBinMetadataKey) <= 1); + + auto iter = client_metadata.find(kEchoInitialMetadataKey); + if (iter != client_metadata.end()) { + context->AddInitialMetadata(kEchoInitialMetadataKey, iter->second.data()); + } + iter = client_metadata.find(kEchoTrailingBinMetadataKey); + if (iter != client_metadata.end()) { + context->AddTrailingMetadata( + kEchoTrailingBinMetadataKey, + grpc::string(iter->second.begin(), iter->second.end())); + } + // Check if client sent a magic key in the header that makes us echo + // back the user-agent (for testing purpose) + iter = client_metadata.find(kEchoUserAgentKey); + if (iter != client_metadata.end()) { + iter = client_metadata.find("user-agent"); + if (iter != client_metadata.end()) { + context->AddInitialMetadata(kEchoUserAgentKey, iter->second.data()); + } + } +} + +bool SetPayload(PayloadType response_type, int size, Payload* payload) { + payload->set_type(response_type); + switch (response_type) { + case PayloadType::COMPRESSABLE: { + std::unique_ptr body(new char[size]()); + payload->set_body(body.get(), size); + } break; + default: + return false; + } + return true; +} + +template +void SetResponseCompression(ServerContext* context, + const RequestType& request) { + if (request.request_compressed_response()) { + // Any level would do, let's go for HIGH because we are overachievers. + context->set_compression_level(GRPC_COMPRESS_LEVEL_HIGH); + } +} + +template +bool CheckExpectedCompression(const ServerContext& context, + const RequestType& request) { + const InteropServerContextInspector inspector(context); + const grpc_compression_algorithm received_compression = + inspector.GetCallCompressionAlgorithm(); + + if (request.expect_compressed_request()) { + if (received_compression == GRPC_COMPRESS_NONE) { + // Expected some compression, got NONE. This is an error. + gpr_log(GPR_ERROR, + "Failure: Expected compression but got uncompressed request " + "from client."); + return false; + } + if (request.payload_type() == PayloadType::COMPRESSABLE) { + if (!(inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS)) { + gpr_log(GPR_ERROR, + "Failure: Requested compression in a compressable request, but " + "compression bit in message flags not set."); + return false; + } + } + } else { + // Didn't expect compression -> make sure the request is uncompressed + if (inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS) { + gpr_log(GPR_ERROR, + "Failure: Didn't requested compression, but compression bit in " + "message flags set."); + return false; + } + } + return true; +} + +class TestServiceImpl : public TestService::Service { + public: + Status EmptyCall(ServerContext* context, const grpc::testing::Empty* request, + grpc::testing::Empty* response) { + MaybeEchoMetadata(context); + return Status::OK; + } + + Status UnaryCall(ServerContext* context, const SimpleRequest* request, + SimpleResponse* response) { + MaybeEchoMetadata(context); + SetResponseCompression(context, *request); + if (!CheckExpectedCompression(*context, *request)) { + return Status(grpc::StatusCode::INVALID_ARGUMENT, + "Compressed request expectation not met."); + } + if (request->response_size() > 0) { + if (!SetPayload(request->response_type(), request->response_size(), + response->mutable_payload())) { + return Status(grpc::StatusCode::INVALID_ARGUMENT, + "Error creating payload."); + } + } + + if (request->has_response_status()) { + return Status( + static_cast(request->response_status().code()), + request->response_status().message()); + } + + return Status::OK; + } + + Status StreamingOutputCall( + ServerContext* context, const StreamingOutputCallRequest* request, + ServerWriter* writer) { + SetResponseCompression(context, *request); + StreamingOutputCallResponse response; + bool write_success = true; + for (int i = 0; write_success && i < request->response_parameters_size(); + i++) { + if (!SetPayload(request->response_type(), + request->response_parameters(i).size(), + response.mutable_payload())) { + return Status(grpc::StatusCode::INVALID_ARGUMENT, + "Error creating payload."); + } + write_success = writer->Write(response); + } + if (write_success) { + return Status::OK; + } else { + return Status(grpc::StatusCode::INTERNAL, "Error writing response."); + } + } + + Status StreamingInputCall(ServerContext* context, + ServerReader* reader, + StreamingInputCallResponse* response) { + StreamingInputCallRequest request; + int aggregated_payload_size = 0; + while (reader->Read(&request)) { + if (!CheckExpectedCompression(*context, request)) { + return Status(grpc::StatusCode::INVALID_ARGUMENT, + "Compressed request expectation not met."); + } + if (request.has_payload()) { + aggregated_payload_size += request.payload().body().size(); + } + } + response->set_aggregated_payload_size(aggregated_payload_size); + return Status::OK; + } + + Status FullDuplexCall( + ServerContext* context, + ServerReaderWriter* stream) { + MaybeEchoMetadata(context); + StreamingOutputCallRequest request; + StreamingOutputCallResponse response; + bool write_success = true; + while (write_success && stream->Read(&request)) { + SetResponseCompression(context, request); + if (request.response_parameters_size() != 0) { + response.mutable_payload()->set_type(request.payload().type()); + response.mutable_payload()->set_body( + grpc::string(request.response_parameters(0).size(), '\0')); + write_success = stream->Write(response); + } + } + if (write_success) { + return Status::OK; + } else { + return Status(grpc::StatusCode::INTERNAL, "Error writing response."); + } + } + + Status HalfDuplexCall( + ServerContext* context, + ServerReaderWriter* stream) { + std::vector requests; + StreamingOutputCallRequest request; + while (stream->Read(&request)) { + requests.push_back(request); + } + + StreamingOutputCallResponse response; + bool write_success = true; + for (unsigned int i = 0; write_success && i < requests.size(); i++) { + response.mutable_payload()->set_type(requests[i].payload().type()); + if (requests[i].response_parameters_size() == 0) { + return Status(grpc::StatusCode::INTERNAL, + "Request does not have response parameters."); + } + response.mutable_payload()->set_body( + grpc::string(requests[i].response_parameters(0).size(), '\0')); + write_success = stream->Write(response); + } + if (write_success) { + return Status::OK; + } else { + return Status(grpc::StatusCode::INTERNAL, "Error writing response."); + } + } +}; + +void RunServer() { + std::ostringstream server_address; + server_address << "0.0.0.0:" << FLAGS_port; + TestServiceImpl service; + + SimpleRequest request; + SimpleResponse response; + + ServerBuilder builder; + builder.RegisterService(&service); + std::shared_ptr creds = + grpc::testing::CreateInteropServerCredentials(); + builder.AddListeningPort(server_address.str(), creds); + std::unique_ptr server(builder.BuildAndStart()); + gpr_log(GPR_INFO, "Server listening on %s", server_address.str().c_str()); + while (!got_sigint) { + sleep(5); + } +} + +static void sigint_handler(int x) { got_sigint = true; } + +int main(int argc, char** argv) { + grpc::testing::InitTest(&argc, &argv, true); + signal(SIGINT, sigint_handler); + + GPR_ASSERT(FLAGS_port != 0); + RunServer(); + + return 0; +} diff --git a/test/cpp/interop/rnd.dat b/test/cpp/interop/rnd.dat deleted file mode 100644 index 8c7f38f9e0..0000000000 Binary files a/test/cpp/interop/rnd.dat and /dev/null differ diff --git a/test/cpp/interop/server_helper.cc b/test/cpp/interop/server_helper.cc index c6d891ad71..8b0b511bcb 100644 --- a/test/cpp/interop/server_helper.cc +++ b/test/cpp/interop/server_helper.cc @@ -72,6 +72,10 @@ uint32_t InteropServerContextInspector::GetEncodingsAcceptedByClient() const { return grpc_call_test_only_get_encodings_accepted_by_peer(context_.call_); } +uint32_t InteropServerContextInspector::GetMessageFlags() const { + return grpc_call_test_only_get_message_flags(context_.call_); +} + std::shared_ptr InteropServerContextInspector::GetAuthContext() const { return context_.auth_context(); diff --git a/test/cpp/interop/server_helper.h b/test/cpp/interop/server_helper.h index 12865e4032..a1da14a4c8 100644 --- a/test/cpp/interop/server_helper.h +++ b/test/cpp/interop/server_helper.h @@ -54,6 +54,7 @@ class InteropServerContextInspector { bool IsCancelled() const; grpc_compression_algorithm GetCallCompressionAlgorithm() const; uint32_t GetEncodingsAcceptedByClient() const; + uint32_t GetMessageFlags() const; private: const ::grpc::ServerContext& context_; diff --git a/test/cpp/interop/server_main.cc b/test/cpp/interop/server_main.cc deleted file mode 100644 index bbedda14d2..0000000000 --- a/test/cpp/interop/server_main.cc +++ /dev/null @@ -1,291 +0,0 @@ -/* - * - * Copyright 2015, 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 -#include - -#include -#include -#include -#include - -#include -#include -#include -#include -#include -#include -#include -#include - -#include "src/proto/grpc/testing/empty.grpc.pb.h" -#include "src/proto/grpc/testing/messages.grpc.pb.h" -#include "src/proto/grpc/testing/test.grpc.pb.h" -#include "test/cpp/interop/server_helper.h" -#include "test/cpp/util/test_config.h" - -DEFINE_bool(use_tls, false, "Whether to use tls."); -DEFINE_int32(port, 0, "Server port."); - -using grpc::Server; -using grpc::ServerBuilder; -using grpc::ServerContext; -using grpc::ServerCredentials; -using grpc::ServerReader; -using grpc::ServerReaderWriter; -using grpc::ServerWriter; -using grpc::SslServerCredentialsOptions; -using grpc::testing::InteropServerContextInspector; -using grpc::testing::Payload; -using grpc::testing::PayloadType; -using grpc::testing::SimpleRequest; -using grpc::testing::SimpleResponse; -using grpc::testing::StreamingInputCallRequest; -using grpc::testing::StreamingInputCallResponse; -using grpc::testing::StreamingOutputCallRequest; -using grpc::testing::StreamingOutputCallResponse; -using grpc::testing::TestService; -using grpc::Status; - -static bool got_sigint = false; -static const char* kRandomFile = "test/cpp/interop/rnd.dat"; - -const char kEchoInitialMetadataKey[] = "x-grpc-test-echo-initial"; -const char kEchoTrailingBinMetadataKey[] = "x-grpc-test-echo-trailing-bin"; -const char kEchoUserAgentKey[] = "x-grpc-test-echo-useragent"; - -void MaybeEchoMetadata(ServerContext* context) { - const auto& client_metadata = context->client_metadata(); - GPR_ASSERT(client_metadata.count(kEchoInitialMetadataKey) <= 1); - GPR_ASSERT(client_metadata.count(kEchoTrailingBinMetadataKey) <= 1); - - auto iter = client_metadata.find(kEchoInitialMetadataKey); - if (iter != client_metadata.end()) { - context->AddInitialMetadata(kEchoInitialMetadataKey, iter->second.data()); - } - iter = client_metadata.find(kEchoTrailingBinMetadataKey); - if (iter != client_metadata.end()) { - context->AddTrailingMetadata( - kEchoTrailingBinMetadataKey, - grpc::string(iter->second.begin(), iter->second.end())); - } - // Check if client sent a magic key in the header that makes us echo - // back the user-agent (for testing purpose) - iter = client_metadata.find(kEchoUserAgentKey); - if (iter != client_metadata.end()) { - iter = client_metadata.find("user-agent"); - if (iter != client_metadata.end()) { - context->AddInitialMetadata(kEchoUserAgentKey, iter->second.data()); - } - } -} - -bool SetPayload(PayloadType response_type, int size, Payload* payload) { - payload->set_type(response_type); - switch (response_type) { - case PayloadType::COMPRESSABLE: { - std::unique_ptr body(new char[size]()); - payload->set_body(body.get(), size); - } break; - case PayloadType::UNCOMPRESSABLE: { - std::unique_ptr body(new char[size]()); - std::ifstream rnd_file(kRandomFile); - GPR_ASSERT(rnd_file.good()); - rnd_file.read(body.get(), size); - GPR_ASSERT(!rnd_file.eof()); // Requested more rnd bytes than available - payload->set_body(body.get(), size); - } break; - default: - GPR_ASSERT(false); - } - return true; -} - -template -void SetResponseCompression(ServerContext* context, - const RequestType& request) { - if (request.request_compressed_response()) { - // Any level would do, let's go for HIGH because we are overachievers. - context->set_compression_level(GRPC_COMPRESS_LEVEL_HIGH); - } -} - -class TestServiceImpl : public TestService::Service { - public: - Status EmptyCall(ServerContext* context, const grpc::testing::Empty* request, - grpc::testing::Empty* response) { - MaybeEchoMetadata(context); - return Status::OK; - } - - Status UnaryCall(ServerContext* context, const SimpleRequest* request, - SimpleResponse* response) { - MaybeEchoMetadata(context); - SetResponseCompression(context, *request); - if (request->response_size() > 0) { - if (!SetPayload(request->response_type(), request->response_size(), - response->mutable_payload())) { - return Status(grpc::StatusCode::INTERNAL, "Error creating payload."); - } - } - - if (request->has_response_status()) { - return Status( - static_cast(request->response_status().code()), - request->response_status().message()); - } - - return Status::OK; - } - - Status StreamingOutputCall( - ServerContext* context, const StreamingOutputCallRequest* request, - ServerWriter* writer) { - SetResponseCompression(context, *request); - StreamingOutputCallResponse response; - bool write_success = true; - for (int i = 0; write_success && i < request->response_parameters_size(); - i++) { - if (!SetPayload(request->response_type(), - request->response_parameters(i).size(), - response.mutable_payload())) { - return Status(grpc::StatusCode::INTERNAL, "Error creating payload."); - } - write_success = writer->Write(response); - } - if (write_success) { - return Status::OK; - } else { - return Status(grpc::StatusCode::INTERNAL, "Error writing response."); - } - } - - Status StreamingInputCall(ServerContext* context, - ServerReader* reader, - StreamingInputCallResponse* response) { - StreamingInputCallRequest request; - int aggregated_payload_size = 0; - while (reader->Read(&request)) { - if (request.has_payload()) { - aggregated_payload_size += request.payload().body().size(); - } - } - response->set_aggregated_payload_size(aggregated_payload_size); - return Status::OK; - } - - Status FullDuplexCall( - ServerContext* context, - ServerReaderWriter* stream) { - MaybeEchoMetadata(context); - StreamingOutputCallRequest request; - StreamingOutputCallResponse response; - bool write_success = true; - while (write_success && stream->Read(&request)) { - SetResponseCompression(context, request); - if (request.response_parameters_size() != 0) { - response.mutable_payload()->set_type(request.payload().type()); - response.mutable_payload()->set_body( - grpc::string(request.response_parameters(0).size(), '\0')); - write_success = stream->Write(response); - } - } - if (write_success) { - return Status::OK; - } else { - return Status(grpc::StatusCode::INTERNAL, "Error writing response."); - } - } - - Status HalfDuplexCall( - ServerContext* context, - ServerReaderWriter* stream) { - std::vector requests; - StreamingOutputCallRequest request; - while (stream->Read(&request)) { - requests.push_back(request); - } - - StreamingOutputCallResponse response; - bool write_success = true; - for (unsigned int i = 0; write_success && i < requests.size(); i++) { - response.mutable_payload()->set_type(requests[i].payload().type()); - if (requests[i].response_parameters_size() == 0) { - return Status(grpc::StatusCode::INTERNAL, - "Request does not have response parameters."); - } - response.mutable_payload()->set_body( - grpc::string(requests[i].response_parameters(0).size(), '\0')); - write_success = stream->Write(response); - } - if (write_success) { - return Status::OK; - } else { - return Status(grpc::StatusCode::INTERNAL, "Error writing response."); - } - } -}; - -void RunServer() { - std::ostringstream server_address; - server_address << "0.0.0.0:" << FLAGS_port; - TestServiceImpl service; - - SimpleRequest request; - SimpleResponse response; - - ServerBuilder builder; - builder.RegisterService(&service); - std::shared_ptr creds = - grpc::testing::CreateInteropServerCredentials(); - builder.AddListeningPort(server_address.str(), creds); - std::unique_ptr server(builder.BuildAndStart()); - gpr_log(GPR_INFO, "Server listening on %s", server_address.str().c_str()); - while (!got_sigint) { - sleep(5); - } -} - -static void sigint_handler(int x) { got_sigint = true; } - -int main(int argc, char** argv) { - grpc::testing::InitTest(&argc, &argv, true); - signal(SIGINT, sigint_handler); - - GPR_ASSERT(FLAGS_port != 0); - RunServer(); - - return 0; -} -- cgit v1.2.3