Merge branch 'master' of github.com:grpc/grpc into compression-accept-encoding

6 files changed, 117 insertions, 88 deletions

diff --git a/test/cpp/interop/client.cc b/test/cpp/interop/client.cc
index ebc5cfc85a..48143b2e53 100644
--- a/test/cpp/interop/client.cc
+++ b/test/cpp/interop/client.cc
@@ -56,8 +56,12 @@ 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_streaming : request streaming with single response; "
               "server_streaming : single request with response streaming; "
+              "server_compressed_streaming : single request with compressed "
+              "response streaming; "
               "slow_consumer : single request with response; "
               " streaming with slow client consumer; "
               "half_duplex : half-duplex streaming; "
@@ -70,7 +74,7 @@ DEFINE_string(test_case, "large_unary",
               "jwt_token_creds: large_unary with JWT token auth; "
               "oauth2_auth_token: raw oauth2 access token auth; "
               "per_rpc_creds: raw oauth2 access token on a single rpc; "
-	      "status_code_and_message: verify status code & message; "
+              "status_code_and_message: verify status code & message; "
               "all : all of above.");
 DEFINE_string(default_service_account, "",
               "Email of GCE default service account");
@@ -91,10 +95,14 @@ 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 == "client_streaming") {
     client.DoRequestStreaming();
   } else if (FLAGS_test_case == "server_streaming") {
     client.DoResponseStreaming();
+  } else if (FLAGS_test_case == "server_compressed_streaming") {
+    client.DoResponseCompressedStreaming();
   } else if (FLAGS_test_case == "slow_consumer") {
     client.DoResponseStreamingWithSlowConsumer();
   } else if (FLAGS_test_case == "half_duplex") {
@@ -129,6 +137,7 @@ int main(int argc, char** argv) {
     client.DoLargeUnary();
     client.DoRequestStreaming();
     client.DoResponseStreaming();
+    client.DoResponseCompressedStreaming();
     client.DoHalfDuplex();
     client.DoPingPong();
     client.DoCancelAfterBegin();
@@ -148,10 +157,11 @@ int main(int argc, char** argv) {
     gpr_log(
         GPR_ERROR,
         "Unsupported test case %s. Valid options are all|empty_unary|"
-        "large_unary|client_streaming|server_streaming|half_duplex|ping_pong|"
-        "cancel_after_begin|cancel_after_first_response|"
-        "timeout_on_sleeping_server|service_account_creds|compute_engine_creds|"
-        "jwt_token_creds|oauth2_auth_token|per_rpc_creds",
+        "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|"
+        "service_account_creds|compute_engine_creds|jwt_token_creds|"
+        "oauth2_auth_token|per_rpc_creds",
         FLAGS_test_case.c_str());
     ret = 1;
   }
diff --git a/test/cpp/interop/client_helper.cc b/test/cpp/interop/client_helper.cc
index 65fdc63b43..da5627de95 100644
--- a/test/cpp/interop/client_helper.cc
+++ b/test/cpp/interop/client_helper.cc
@@ -52,7 +52,6 @@
 #include "test/core/security/oauth2_utils.h"
 #include "test/cpp/util/create_test_channel.h"
 
-#include "src/core/surface/call.h"
 #include "src/cpp/client/secure_credentials.h"
 
 DECLARE_bool(enable_ssl);
@@ -65,8 +64,6 @@ DECLARE_string(default_service_account);
 DECLARE_string(service_account_key_file);
 DECLARE_string(oauth_scope);
 
-using grpc::testing::CompressionType;
-
 namespace grpc {
 namespace testing {
 
@@ -143,32 +140,5 @@ std::shared_ptr<ChannelInterface> CreateChannelForTestCase(
   }
 }
 
-CompressionType GetInteropCompressionTypeFromCompressionAlgorithm(
-    grpc_compression_algorithm algorithm) {
-  switch (algorithm) {
-    case GRPC_COMPRESS_NONE:
-      return CompressionType::NONE;
-    case GRPC_COMPRESS_GZIP:
-      return CompressionType::GZIP;
-    case GRPC_COMPRESS_DEFLATE:
-      return CompressionType::DEFLATE;
-    default:
-      GPR_ASSERT(false);
-  }
-}
-
-InteropClientContextInspector::InteropClientContextInspector(
-    const ::grpc::ClientContext& context)
-    : context_(context) {}
-
-grpc_compression_algorithm
-InteropClientContextInspector::GetCallCompressionAlgorithm() const {
-  return grpc_call_get_compression_algorithm(context_.call_);
-}
-
-gpr_uint32 InteropClientContextInspector::GetMessageFlags() const {
-  return grpc_call_get_message_flags(context_.call_);
-}
-
 }  // namespace testing
 }  // namespace grpc
diff --git a/test/cpp/interop/client_helper.h b/test/cpp/interop/client_helper.h
index 1c7036d25d..edc69e90ac 100644
--- a/test/cpp/interop/client_helper.h
+++ b/test/cpp/interop/client_helper.h
@@ -39,7 +39,7 @@
 #include <grpc++/config.h>
 #include <grpc++/channel_interface.h>
 
-#include "test/proto/messages.grpc.pb.h"
+#include "src/core/surface/call.h"
 
 namespace grpc {
 namespace testing {
@@ -51,23 +51,24 @@ grpc::string GetOauth2AccessToken();
 std::shared_ptr<ChannelInterface> CreateChannelForTestCase(
     const grpc::string& test_case);
 
-grpc::testing::CompressionType
-GetInteropCompressionTypeFromCompressionAlgorithm(
-    grpc_compression_algorithm algorithm);
-
 class InteropClientContextInspector {
  public:
-  InteropClientContextInspector(const ::grpc::ClientContext& context);
+  InteropClientContextInspector(const ::grpc::ClientContext& context)
+    : context_(context) {}
 
   // Inspector methods, able to peek inside ClientContext, follow.
-  grpc_compression_algorithm GetCallCompressionAlgorithm() const;
-  gpr_uint32 GetMessageFlags() const;
+  grpc_compression_algorithm GetCallCompressionAlgorithm() const {
+    return grpc_call_get_compression_algorithm(context_.call_);
+  }
+
+  gpr_uint32 GetMessageFlags() const {
+    return grpc_call_get_message_flags(context_.call_);
+  }
 
  private:
   const ::grpc::ClientContext& context_;
 };
 
-
 }  // namespace testing
 }  // namespace grpc
 
diff --git a/test/cpp/interop/interop_client.cc b/test/cpp/interop/interop_client.cc
index 8e2d778cff..5ed14d556a 100644
--- a/test/cpp/interop/interop_client.cc
+++ b/test/cpp/interop/interop_client.cc
@@ -41,6 +41,7 @@
 #include <grpc/grpc.h>
 #include <grpc/support/log.h>
 #include <grpc/support/string_util.h>
+#include <grpc/support/useful.h>
 #include <grpc++/channel_interface.h>
 #include <grpc++/client_context.h>
 #include <grpc++/credentials.h>
@@ -67,6 +68,20 @@ const int kResponseMessageSize = 1030;
 const int kReceiveDelayMilliSeconds = 20;
 const int kLargeRequestSize = 271828;
 const int kLargeResponseSize = 314159;
+
+CompressionType GetInteropCompressionTypeFromCompressionAlgorithm(
+    grpc_compression_algorithm algorithm) {
+  switch (algorithm) {
+    case GRPC_COMPRESS_NONE:
+      return CompressionType::NONE;
+    case GRPC_COMPRESS_GZIP:
+      return CompressionType::GZIP;
+    case GRPC_COMPRESS_DEFLATE:
+      return CompressionType::DEFLATE;
+    default:
+      GPR_ASSERT(false);
+  }
+}
 }  // namespace
 
 InteropClient::InteropClient(std::shared_ptr<ChannelInterface> channel)
@@ -102,6 +117,8 @@ void InteropClient::PerformLargeUnary(SimpleRequest* request,
 
   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);
@@ -132,13 +149,12 @@ void InteropClient::PerformLargeUnary(SimpleRequest* request,
                  grpc::string(kLargeResponseSize, '\0'));
       break;
     case PayloadType::UNCOMPRESSABLE: {
-        std::ifstream rnd_file(kRandomFile);
-        GPR_ASSERT(rnd_file.good());
-        for (int i = 0; i < kLargeResponseSize; i++) {
-          GPR_ASSERT(response->payload().body()[i] == (char)rnd_file.get());
-        }
+      std::ifstream rnd_file(kRandomFile);
+      GPR_ASSERT(rnd_file.good());
+      for (int i = 0; i < kLargeResponseSize; i++) {
+        GPR_ASSERT(response->payload().body()[i] == (char)rnd_file.get());
       }
-      break;
+    } break;
     default:
       GPR_ASSERT(false);
   }
@@ -245,22 +261,31 @@ void InteropClient::DoJwtTokenCreds(const grpc::string& username) {
 }
 
 void InteropClient::DoLargeUnary() {
+  gpr_log(GPR_INFO, "Sending a large unary rpc...");
+  SimpleRequest request;
+  SimpleResponse response;
+  request.set_response_type(PayloadType::COMPRESSABLE);
+  PerformLargeUnary(&request, &response);
+  gpr_log(GPR_INFO, "Large unary done.");
+}
+
+void InteropClient::DoLargeCompressedUnary() {
   const CompressionType compression_types[] = {NONE, GZIP, DEFLATE};
   const PayloadType payload_types[] = {COMPRESSABLE, UNCOMPRESSABLE, RANDOM};
-  for (const auto payload_type : payload_types) {
-    for (const auto compression_type : compression_types) {
+  for (size_t i = 0; i < GPR_ARRAY_SIZE(payload_types); i++) {
+    for (size_t j = 0; j < GPR_ARRAY_SIZE(compression_types); j++) {
       char* log_suffix;
       gpr_asprintf(&log_suffix, "(compression=%s; payload=%s)",
-          CompressionType_Name(compression_type).c_str(),
-          PayloadType_Name(payload_type).c_str());
+                   CompressionType_Name(compression_types[j]).c_str(),
+                   PayloadType_Name(payload_types[i]).c_str());
 
-      gpr_log(GPR_INFO, "Sending a large unary rpc %s.", log_suffix);
+      gpr_log(GPR_INFO, "Sending a large compressed unary rpc %s.", log_suffix);
       SimpleRequest request;
       SimpleResponse response;
-      request.set_response_type(payload_type);
-      request.set_response_compression(compression_type);
+      request.set_response_type(payload_types[i]);
+      request.set_response_compression(compression_types[j]);
       PerformLargeUnary(&request, &response);
-      gpr_log(GPR_INFO, "Large unary done %s.", log_suffix);
+      gpr_log(GPR_INFO, "Large compressed unary done %s.", log_suffix);
       gpr_free(log_suffix);
     }
   }
@@ -293,25 +318,51 @@ void InteropClient::DoRequestStreaming() {
 }
 
 void InteropClient::DoResponseStreaming() {
+  gpr_log(GPR_INFO, "Receiving response steaming rpc ...");
+  std::unique_ptr<TestService::Stub> stub(TestService::NewStub(channel_));
+
+  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<ClientReader<StreamingOutputCallResponse>> stream(
+      stub->StreamingOutputCall(&context, request));
+
+  unsigned int i = 0;
+  while (stream->Read(&response)) {
+    GPR_ASSERT(response.payload().body() ==
+               grpc::string(response_stream_sizes[i], '\0'));
+    ++i;
+  }
+  GPR_ASSERT(response_stream_sizes.size() == i);
+  Status s = stream->Finish();
+  AssertOkOrPrintErrorStatus(s);
+  gpr_log(GPR_INFO, "Response streaming done.");
+}
+
+void InteropClient::DoResponseCompressedStreaming() {
   std::unique_ptr<TestService::Stub> stub(TestService::NewStub(channel_));
 
   const CompressionType compression_types[] = {NONE, GZIP, DEFLATE};
   const PayloadType payload_types[] = {COMPRESSABLE, UNCOMPRESSABLE, RANDOM};
-  for (const auto payload_type : payload_types) {
-    for (const auto compression_type : compression_types) {
+  for (size_t i = 0; i < GPR_ARRAY_SIZE(payload_types); i++) {
+    for (size_t j = 0; j < GPR_ARRAY_SIZE(compression_types); j++) {
       ClientContext context;
       InteropClientContextInspector inspector(context);
       StreamingOutputCallRequest request;
 
       char* log_suffix;
       gpr_asprintf(&log_suffix, "(compression=%s; payload=%s)",
-          CompressionType_Name(compression_type).c_str(),
-          PayloadType_Name(payload_type).c_str());
+                   CompressionType_Name(compression_types[j]).c_str(),
+                   PayloadType_Name(payload_types[i]).c_str());
 
       gpr_log(GPR_INFO, "Receiving response steaming rpc %s.", log_suffix);
 
-      request.set_response_type(payload_type);
-      request.set_response_compression(compression_type);
+      request.set_response_type(payload_types[i]);
+      request.set_response_compression(compression_types[j]);
 
       for (unsigned int i = 0; i < response_stream_sizes.size(); ++i) {
         ResponseParameters* response_parameter =
@@ -405,7 +456,6 @@ void InteropClient::DoHalfDuplex() {
   unsigned int i = 0;
   StreamingOutputCallResponse response;
   while (stream->Read(&response)) {
-    GPR_ASSERT(response.payload().has_body());
     GPR_ASSERT(response.payload().body() ==
                grpc::string(response_stream_sizes[i], '\0'));
     ++i;
@@ -435,7 +485,6 @@ void InteropClient::DoPingPong() {
     payload->set_body(grpc::string(request_stream_sizes[i], '\0'));
     GPR_ASSERT(stream->Write(request));
     GPR_ASSERT(stream->Read(&response));
-    GPR_ASSERT(response.payload().has_body());
     GPR_ASSERT(response.payload().body() ==
                grpc::string(response_stream_sizes[i], '\0'));
   }
@@ -482,7 +531,6 @@ void InteropClient::DoCancelAfterFirstResponse() {
   StreamingOutputCallResponse response;
   GPR_ASSERT(stream->Write(request));
   GPR_ASSERT(stream->Read(&response));
-  GPR_ASSERT(response.payload().has_body());
   GPR_ASSERT(response.payload().body() == grpc::string(31415, '\0'));
   gpr_log(GPR_INFO, "Trying to cancel...");
   context.TryCancel();
@@ -519,7 +567,7 @@ void InteropClient::DoStatusWithMessage() {
   ClientContext context;
   SimpleRequest request;
   SimpleResponse response;
-  EchoStatus *requested_status = request.mutable_response_status();
+  EchoStatus* requested_status = request.mutable_response_status();
   requested_status->set_code(grpc::StatusCode::UNKNOWN);
   grpc::string test_msg = "This is a test message";
   requested_status->set_message(test_msg);
diff --git a/test/cpp/interop/interop_client.h b/test/cpp/interop/interop_client.h
index 6e26c49e5d..d6fb9bff39 100644
--- a/test/cpp/interop/interop_client.h
+++ b/test/cpp/interop/interop_client.h
@@ -52,10 +52,12 @@ class InteropClient {
 
   void DoEmpty();
   void DoLargeUnary();
+  void DoLargeCompressedUnary();
   void DoPingPong();
   void DoHalfDuplex();
   void DoRequestStreaming();
   void DoResponseStreaming();
+  void DoResponseCompressedStreaming();
   void DoResponseStreamingWithSlowConsumer();
   void DoCancelAfterBegin();
   void DoCancelAfterFirstResponse();
diff --git a/test/cpp/interop/server.cc b/test/cpp/interop/server.cc
index 0485e447e5..760bb18f73 100644
--- a/test/cpp/interop/server.cc
+++ b/test/cpp/interop/server.cc
@@ -43,6 +43,7 @@
 #include <grpc/grpc.h>
 #include <grpc/support/log.h>
 #include <grpc/support/useful.h>
+
 #include <grpc++/config.h>
 #include <grpc++/server.h>
 #include <grpc++/server_builder.h>
@@ -114,20 +115,18 @@ bool SetPayload(PayloadType type, int size, Payload* payload) {
 template <typename RequestType>
 void SetResponseCompression(ServerContext* context,
                             const RequestType& request) {
-  if (request.has_response_compression()) {
-    switch (request.response_compression()) {
-      case grpc::testing::NONE:
-        context->set_compression_algorithm(GRPC_COMPRESS_NONE);
-        break;
-      case grpc::testing::GZIP:
-        context->set_compression_algorithm(GRPC_COMPRESS_GZIP);
-        break;
-      case grpc::testing::DEFLATE:
-        context->set_compression_algorithm(GRPC_COMPRESS_DEFLATE);
-        break;
-    }
-  } else {
-    context->set_compression_algorithm(GRPC_COMPRESS_NONE);
+  switch (request.response_compression()) {
+    case grpc::testing::NONE:
+      context->set_compression_algorithm(GRPC_COMPRESS_NONE);
+      break;
+    case grpc::testing::GZIP:
+      context->set_compression_algorithm(GRPC_COMPRESS_GZIP);
+      break;
+    case grpc::testing::DEFLATE:
+      context->set_compression_algorithm(GRPC_COMPRESS_DEFLATE);
+      break;
+    default:
+      abort();
   }
 }
 
@@ -140,9 +139,8 @@ class TestServiceImpl : public TestService::Service {
 
   Status UnaryCall(ServerContext* context, const SimpleRequest* request,
                    SimpleResponse* response) {
-    InteropServerContextInspector inspector(*context);
     SetResponseCompression(context, *request);
-    if (request->has_response_size() && request->response_size() > 0) {
+    if (request->response_size() > 0) {
       if (!SetPayload(request->response_type(), request->response_size(),
                       response->mutable_payload())) {
         return Status(grpc::StatusCode::INTERNAL, "Error creating payload.");
@@ -150,9 +148,9 @@ class TestServiceImpl : public TestService::Service {
     }
 
     if (request->has_response_status()) {
-      return Status(static_cast<grpc::StatusCode>
-		    (request->response_status().code()),
-		    request->response_status().message()); 
+      return Status(
+          static_cast<grpc::StatusCode>(request->response_status().code()),
+          request->response_status().message());
     }
 
     return Status::OK;
@@ -184,7 +182,7 @@ class TestServiceImpl : public TestService::Service {
     StreamingInputCallRequest request;
     int aggregated_payload_size = 0;
     while (reader->Read(&request)) {
-      if (request.has_payload() && request.payload().has_body()) {
+      if (request.has_payload()) {
         aggregated_payload_size += request.payload().body().size();
       }
     }