c++ client & server changes to bring code up to spec

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(