aboutsummaryrefslogtreecommitdiffhomepage
diff options
context:
space:
mode:
Diffstat
-rw-r--r--test/cpp/interop/interop_client.cc337
-rw-r--r--test/cpp/interop/interop_server.cc86
2 files changed, 227 insertions, 196 deletions
diff --git a/test/cpp/interop/interop_client.cc b/test/cpp/interop/interop_client.cc
index e5d3751402..509e8b97ab 100644
--- a/test/cpp/interop/interop_client.cc
+++ b/test/cpp/interop/interop_client.cc
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -68,12 +68,12 @@ const int kLargeResponseSize = 314159;
void NoopChecks(const InteropClientContextInspector& inspector,
const SimpleRequest* request, const SimpleResponse* response) {}
-void CompressionChecks(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->request_compressed_response()) {
+ if (request->response_compressed().value()) {
if (received_compression == GRPC_COMPRESS_NONE) {
// Requested some compression, got NONE. This is an error.
gpr_log(GPR_ERROR,
@@ -81,11 +81,7 @@ void CompressionChecks(const InteropClientContextInspector& inspector,
"from server.");
abort();
}
- if (request->response_type() == PayloadType::COMPRESSABLE) {
- // requested compression and compressable response => results should
- // always be compressed.
- GPR_ASSERT(inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS);
- }
+ 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));
@@ -190,11 +186,16 @@ bool InteropClient::PerformLargeUnary(SimpleRequest* request,
CheckerFn custom_checks_fn) {
ClientContext context;
InteropClientContextInspector inspector(context);
- // If the request doesn't already specify the response type, default to
- // COMPRESSABLE.
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)) {
@@ -204,16 +205,8 @@ bool InteropClient::PerformLargeUnary(SimpleRequest* request,
custom_checks_fn(inspector, request, response);
// Payload related checks.
- GPR_ASSERT(response->payload().type() == request->response_type());
- switch (response->payload().type()) {
- case PayloadType::COMPRESSABLE:
- GPR_ASSERT(response->payload().body() ==
- grpc::string(kLargeResponseSize, '\0'));
- break;
- default:
- GPR_ASSERT(false);
- }
-
+ GPR_ASSERT(response->payload().body() ==
+ grpc::string(kLargeResponseSize, '\0'));
return true;
}
@@ -226,7 +219,6 @@ bool InteropClient::DoComputeEngineCreds(
SimpleResponse response;
request.set_fill_username(true);
request.set_fill_oauth_scope(true);
- request.set_response_type(PayloadType::COMPRESSABLE);
if (!PerformLargeUnary(&request, &response)) {
return false;
@@ -300,7 +292,6 @@ bool InteropClient::DoJwtTokenCreds(const grpc::string& username) {
SimpleRequest request;
SimpleResponse response;
request.set_fill_username(true);
- request.set_response_type(PayloadType::COMPRESSABLE);
if (!PerformLargeUnary(&request, &response)) {
return false;
@@ -316,7 +307,6 @@ bool InteropClient::DoLargeUnary() {
gpr_log(GPR_DEBUG, "Sending a large unary rpc...");
SimpleRequest request;
SimpleResponse response;
- request.set_response_type(PayloadType::COMPRESSABLE);
if (!PerformLargeUnary(&request, &response)) {
return false;
}
@@ -325,62 +315,72 @@ bool InteropClient::DoLargeUnary() {
}
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);
+ // 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 %s");
+ return false;
+ }
+ gpr_log(GPR_DEBUG, "Compressed unary request probe succeeded. Proceeding.");
+
+ const std::vector<bool> 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 bool request_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(request_compression); j++) {
- char* log_suffix;
- gpr_asprintf(&log_suffix, "(compression=%s; payload=%s)",
- request_compression[j] ? "true" : "false",
- PayloadType_Name(payload_types[i]).c_str());
-
- gpr_log(GPR_DEBUG, "Sending unary request for compressed response %s.",
- log_suffix);
- SimpleRequest request;
- SimpleResponse response;
- request.set_response_type(payload_types[i]);
- request.set_request_compressed_response(request_compression[j]);
-
- if (!PerformLargeUnary(&request, &response, CompressionChecks)) {
- 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);
+ const std::vector<bool> 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;
@@ -407,7 +407,7 @@ bool InteropClient::DoRequestStreaming() {
serviceStub_.Get()->StreamingInputCall(&context, &response));
int aggregated_payload_size = 0;
- for (unsigned int i = 0; i < request_stream_sizes.size(); ++i) {
+ 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)) {
@@ -416,7 +416,7 @@ bool InteropClient::DoRequestStreaming() {
}
aggregated_payload_size += request_stream_sizes[i];
}
- stream->WritesDone();
+ GPR_ASSERT(stream->WritesDone());
Status s = stream->Finish();
if (!AssertStatusOk(s)) {
@@ -467,94 +467,128 @@ bool InteropClient::DoResponseStreaming() {
}
bool InteropClient::DoClientCompressedStreaming() {
- // XXX
- return false;
+ // 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<ClientWriter<StreamingInputCallRequest>> 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 %s");
+ 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<ClientWriter<StreamingInputCallRequest>> 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)) {
+ return false;
+ }
+
+ return true;
}
bool InteropClient::DoServerCompressedStreaming() {
- const bool request_compression[] = {false, true};
- const PayloadType payload_types[] = {COMPRESSABLE};
- const std::vector<int> response_stream_sizes = {31415, 58979};
+ const std::vector<bool> compressions = {true, false};
+ const std::vector<int> sizes = {31415, 92653};
- 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;
- InteropClientContextInspector inspector(context);
- StreamingOutputCallRequest request;
+ ClientContext context;
+ InteropClientContextInspector inspector(context);
+ StreamingOutputCallRequest request;
+ for (size_t i = 0; i < compressions.size(); i++) {
+ for (size_t j = 0; j < sizes.size(); j++) {
char* log_suffix;
- gpr_asprintf(&log_suffix, "(compression=%s; payload=%s)",
- request_compression[j] ? "true" : "false",
- PayloadType_Name(payload_types[i]).c_str());
-
- gpr_log(GPR_DEBUG, "Receiving response streaming rpc %s.", log_suffix);
-
- 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(); ++k) {
- ResponseParameters* response_parameter =
- request.add_response_parameters();
- response_parameter->set_size(response_stream_sizes[k]);
- }
- StreamingOutputCallResponse response;
-
- std::unique_ptr<ClientReader<StreamingOutputCallResponse>> stream(
- serviceStub_.Get()->StreamingOutputCall(&context, request));
-
- size_t k = 0;
- while (stream->Read(&response)) {
- // Payload related checks.
- GPR_ASSERT(response.payload().type() == request.response_type());
- switch (response.payload().type()) {
- case PayloadType::COMPRESSABLE:
- GPR_ASSERT(response.payload().body() ==
- grpc::string(response_stream_sizes[k], '\0'));
- break;
- default:
- GPR_ASSERT(false);
- }
-
- // Compression related checks.
- if (request.request_compressed_response()) {
- GPR_ASSERT(inspector.GetCallCompressionAlgorithm() >
- GRPC_COMPRESS_NONE);
- 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 {
- // requested *no* compression.
- GPR_ASSERT(
- !(inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS));
- }
-
- ++k;
- }
-
- gpr_log(GPR_DEBUG, "Response streaming done %s.", log_suffix);
+ gpr_asprintf(&log_suffix, "(compression=%s; size=%d)",
+ compressions[i] ? "true" : "false", sizes[j]);
+
+ gpr_log(GPR_DEBUG, "Sending request streaming rpc %s.", log_suffix);
gpr_free(log_suffix);
- 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,
- "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(), i, j);
- return TransientFailureOrAbort();
- }
-
- Status s = stream->Finish();
- if (!AssertStatusOk(s)) {
- return false;
- }
+ ResponseParameters* const response_parameter =
+ request.add_response_parameters();
+ response_parameter->mutable_compressed()->set_value(compressions[i]);
+ response_parameter->set_size(sizes[j]);
}
}
+ std::unique_ptr<ClientReader<StreamingOutputCallResponse>> 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 < 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,
+ "%s(): Responses read (k=%d) is "
+ "less than the expected number of messages (%d)).",
+ __func__, k, response_stream_sizes.size());
+ return TransientFailureOrAbort();
+ }
+ Status s = stream->Finish();
+ if (!AssertStatusOk(s)) {
+ return false;
+ }
return true;
}
@@ -655,7 +689,6 @@ bool InteropClient::DoPingPong() {
stream(serviceStub_.Get()->FullDuplexCall(&context));
StreamingOutputCallRequest request;
- request.set_response_type(PayloadType::COMPRESSABLE);
ResponseParameters* response_parameter = request.add_response_parameters();
Payload* payload = request.mutable_payload();
StreamingOutputCallResponse response;
@@ -722,7 +755,6 @@ bool InteropClient::DoCancelAfterFirstResponse() {
stream(serviceStub_.Get()->FullDuplexCall(&context));
StreamingOutputCallRequest request;
- request.set_response_type(PayloadType::COMPRESSABLE);
ResponseParameters* response_parameter = request.add_response_parameters();
response_parameter->set_size(31415);
request.mutable_payload()->set_body(grpc::string(27182, '\0'));
@@ -862,7 +894,6 @@ bool InteropClient::DoCustomMetadata() {
stream(serviceStub_.Get()->FullDuplexCall(&context));
StreamingOutputCallRequest request;
- request.set_response_type(PayloadType::COMPRESSABLE);
ResponseParameters* response_parameter = request.add_response_parameters();
response_parameter->set_size(kLargeResponseSize);
grpc::string payload(kLargeRequestSize, '\0');
diff --git a/test/cpp/interop/interop_server.cc b/test/cpp/interop/interop_server.cc
index b328f478fa..8c5c0e24e1 100644
--- a/test/cpp/interop/interop_server.cc
+++ b/test/cpp/interop/interop_server.cc
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -65,10 +65,10 @@ using grpc::ServerCredentials;
using grpc::ServerReader;
using grpc::ServerReaderWriter;
using grpc::ServerWriter;
+using grpc::WriteOptions;
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;
@@ -110,50 +110,30 @@ void MaybeEchoMetadata(ServerContext* context) {
}
}
-bool SetPayload(PayloadType response_type, int size, Payload* payload) {
- payload->set_type(response_type);
- switch (response_type) {
- case PayloadType::COMPRESSABLE: {
- std::unique_ptr<char[]> body(new char[size]());
- payload->set_body(body.get(), size);
- } break;
- default:
- return false;
- }
+bool SetPayload(int size, Payload* payload) {
+ std::unique_ptr<char[]> body(new char[size]());
+ payload->set_body(body.get(), size);
return true;
}
-template <typename RequestType>
-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 <typename RequestType>
bool CheckExpectedCompression(const ServerContext& context,
- const RequestType& request) {
+ const bool compression_expected) {
const InteropServerContextInspector inspector(context);
const grpc_compression_algorithm received_compression =
inspector.GetCallCompressionAlgorithm();
- if (request.expect_compressed_request()) {
+ if (compression_expected) {
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.");
+ "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;
- }
+ 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
@@ -178,14 +158,24 @@ class TestServiceImpl : public TestService::Service {
Status UnaryCall(ServerContext* context, const SimpleRequest* request,
SimpleResponse* response) {
MaybeEchoMetadata(context);
- SetResponseCompression(context, *request);
- if (!CheckExpectedCompression(*context, *request)) {
+ if (request->has_response_compressed()) {
+ const bool compression_requested = request->response_compressed().value();
+ gpr_log(GPR_DEBUG, "Request for compression (%s) present for %s",
+ compression_requested ? "enabled" : "disabled", __func__);
+ if (compression_requested) {
+ // Any level would do, let's go for HIGH because we are overachievers.
+ context->set_compression_level(GRPC_COMPRESS_LEVEL_HIGH);
+ } else {
+ context->set_compression_level(GRPC_COMPRESS_LEVEL_NONE);
+ }
+ }
+ if (!CheckExpectedCompression(*context,
+ request->expect_compressed().value())) {
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())) {
+ if (!SetPayload(request->response_size(), response->mutable_payload())) {
return Status(grpc::StatusCode::INVALID_ARGUMENT,
"Error creating payload.");
}
@@ -203,18 +193,28 @@ class TestServiceImpl : public TestService::Service {
Status StreamingOutputCall(
ServerContext* context, const StreamingOutputCallRequest* request,
ServerWriter<StreamingOutputCallResponse>* writer) {
- SetResponseCompression(context, *request);
StreamingOutputCallResponse response;
+ // Compress by default. Disabled on a per-message basis.
+ context->set_compression_level(GRPC_COMPRESS_LEVEL_HIGH);
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(),
+ if (!SetPayload(request->response_parameters(i).size(),
response.mutable_payload())) {
return Status(grpc::StatusCode::INVALID_ARGUMENT,
"Error creating payload.");
}
- write_success = writer->Write(response);
+ WriteOptions wopts;
+ if (request->response_parameters(i).has_compressed()) {
+ const bool compression_requested =
+ request->response_parameters(i).compressed().value();
+ gpr_log(GPR_DEBUG, "Request for compression (%s) present for %s",
+ compression_requested ? "enabled" : "disabled", __func__);
+ if (!compression_requested) {
+ wopts.set_no_compression();
+ } // else, compression is already enabled via the context.
+ }
+ write_success = writer->Write(response, wopts);
}
if (write_success) {
return Status::OK;
@@ -229,7 +229,8 @@ class TestServiceImpl : public TestService::Service {
StreamingInputCallRequest request;
int aggregated_payload_size = 0;
while (reader->Read(&request)) {
- if (!CheckExpectedCompression(*context, request)) {
+ if (!CheckExpectedCompression(*context,
+ request.expect_compressed().value())) {
return Status(grpc::StatusCode::INVALID_ARGUMENT,
"Compressed request expectation not met.");
}
@@ -250,7 +251,6 @@ class TestServiceImpl : public TestService::Service {
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(
generated by cgit v1.2.3 (git 2.39.1) at 2024-12-22 13:59:12 +0000