Merge pull request #3067 from dgquintas/interop_streaming_fix

Added missing payloads to streaming compressed test

test/cpp/interop/interop_client.cc

@@ -362,20 +362,37 @@ void InteropClient::DoResponseCompressedStreaming() {
       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) {
+      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[i]);
+        response_parameter->set_size(response_stream_sizes[k]);
       }
       StreamingOutputCallResponse response;
 
       std::unique_ptr<ClientReader<StreamingOutputCallResponse>> stream(
           stub->StreamingOutputCall(&context, request));
 
-      unsigned int i = 0;
+      size_t k = 0;
       while (stream->Read(&response)) {
-        GPR_ASSERT(response.payload().body() ==
-                   grpc::string(response_stream_sizes[i], '\0'));
+        // Payload related checks.
+        if (request.response_type() != PayloadType::RANDOM) {
+          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;
+          case PayloadType::UNCOMPRESSABLE: {
+            std::ifstream rnd_file(kRandomFile);
+            GPR_ASSERT(rnd_file.good());
+            for (int n = 0; n < response_stream_sizes[k]; n++) {
+              GPR_ASSERT(response.payload().body()[n] == (char)rnd_file.get());
+            }
+          } break;
+          default:
+            GPR_ASSERT(false);
+        }
 
         // Compression related checks.
         GPR_ASSERT(request.response_compression() ==
@@ -391,10 +408,10 @@ void InteropClient::DoResponseCompressedStreaming() {
                      GRPC_WRITE_INTERNAL_COMPRESS);
         }
 
-        ++i;
+        ++k;
       }
 
-      GPR_ASSERT(response_stream_sizes.size() == i);
+      GPR_ASSERT(response_stream_sizes.size() == k);
       Status s = stream->Finish();
 
       AssertOkOrPrintErrorStatus(s);

test/cpp/interop/server.cc

@@ -158,11 +158,13 @@ class TestServiceImpl : public TestService::Service {
     SetResponseCompression(context, *request);
     StreamingOutputCallResponse response;
     bool write_success = true;
-    response.mutable_payload()->set_type(request->response_type());
     for (int i = 0; write_success && i < request->response_parameters_size();
          i++) {
-      response.mutable_payload()->set_body(
-          grpc::string(request->response_parameters(i).size(), '\0'));
+      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) {