Initial checkin of gRPC to third_party/

third_party/grpc/test/cpp/interop/interop_client.cc

+/*
+ *
+ * Copyright 2015-2016, 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 "test/cpp/interop/interop_client.h"
+
+#include <unistd.h>
+
+#include <fstream>
+#include <memory>
+
+#include <grpc++/channel.h>
+#include <grpc++/client_context.h>
+#include <grpc++/security/credentials.h>
+#include <grpc/grpc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/useful.h>
+
+#include "src/core/transport/byte_stream.h"
+#include "src/proto/grpc/testing/empty.grpc.pb.h"
+#include "src/proto/grpc/testing/test.grpc.pb.h"
+#include "src/proto/grpc/testing/messages.grpc.pb.h"
+#include "test/cpp/interop/client_helper.h"
+
+namespace grpc {
+namespace testing {
+
+static const char* kRandomFile = "test/cpp/interop/rnd.dat";
+
+namespace {
+// The same value is defined by the Java client.
+const std::vector<int> request_stream_sizes = {27182, 8, 1828, 45904};
+const std::vector<int> response_stream_sizes = {31415, 9, 2653, 58979};
+const int kNumResponseMessages = 2000;
+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);
+  }
+}
+
+void NoopChecks(const InteropClientContextInspector& inspector,
+                const SimpleRequest* request, const SimpleResponse* response) {}
+
+void CompressionChecks(const InteropClientContextInspector& inspector,
+                       const SimpleRequest* request,
+                       const SimpleResponse* response) {
+  GPR_ASSERT(request->response_compression() ==
+             GetInteropCompressionTypeFromCompressionAlgorithm(
+                 inspector.GetCallCompressionAlgorithm()));
+  if (request->response_compression() == NONE) {
+    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
+
+InteropClient::ServiceStub::ServiceStub(std::shared_ptr<Channel> channel,
+                                        bool new_stub_every_call)
+    : channel_(channel), new_stub_every_call_(new_stub_every_call) {
+  // If new_stub_every_call is false, then this is our chance to initialize
+  // stub_. (see Get())
+  if (!new_stub_every_call) {
+    stub_ = TestService::NewStub(channel);
+  }
+}
+
+TestService::Stub* InteropClient::ServiceStub::Get() {
+  if (new_stub_every_call_) {
+    stub_ = TestService::NewStub(channel_);
+  }
+
+  return stub_.get();
+}
+
+void InteropClient::ServiceStub::Reset(std::shared_ptr<Channel> channel) {
+  channel_ = channel;
+
+  // Update stub_ as well. Note: If new_stub_every_call_ is true, we can reset
+  // the stub_ since the next call to Get() will create a new stub
+  if (new_stub_every_call_) {
+    stub_.reset();
+  } else {
+    stub_ = TestService::NewStub(channel);
+  }
+}
+
+void InteropClient::Reset(std::shared_ptr<Channel> channel) {
+  serviceStub_.Reset(channel);
+}
+
+InteropClient::InteropClient(std::shared_ptr<Channel> channel)
+    : serviceStub_(channel, true) {}
+
+InteropClient::InteropClient(std::shared_ptr<Channel> channel,
+                             bool new_stub_every_test_case)
+    : serviceStub_(channel, new_stub_every_test_case) {}
+
+void InteropClient::AssertOkOrPrintErrorStatus(const Status& s) {
+  if (s.ok()) {
+    return;
+  }
+  gpr_log(GPR_ERROR, "Error status code: %d, message: %s", s.error_code(),
+          s.error_message().c_str());
+  GPR_ASSERT(0);
+}
+
+void InteropClient::DoEmpty() {
+  gpr_log(GPR_DEBUG, "Sending an empty rpc...");
+
+  Empty request = Empty::default_instance();
+  Empty response = Empty::default_instance();
+  ClientContext context;
+
+  Status s = serviceStub_.Get()->EmptyCall(&context, request, &response);
+  AssertOkOrPrintErrorStatus(s);
+
+  gpr_log(GPR_DEBUG, "Empty rpc done.");
+}
+
+void InteropClient::PerformLargeUnary(SimpleRequest* request,
+                                      SimpleResponse* response) {
+  PerformLargeUnary(request, response, NoopChecks);
+}
+
+void InteropClient::PerformLargeUnary(SimpleRequest* request,
+                                      SimpleResponse* response,
+                                      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);
+
+  Status s = serviceStub_.Get()->UnaryCall(&context, *request, response);
+  AssertOkOrPrintErrorStatus(s);
+
+  custom_checks_fn(inspector, request, response);
+
+  // 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(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());
+      }
+    } break;
+    default:
+      GPR_ASSERT(false);
+  }
+}
+
+void InteropClient::DoComputeEngineCreds(
+    const grpc::string& default_service_account,
+    const grpc::string& oauth_scope) {
+  gpr_log(GPR_DEBUG,
+          "Sending a large unary rpc with compute engine credentials ...");
+  SimpleRequest request;
+  SimpleResponse response;
+  request.set_fill_username(true);
+  request.set_fill_oauth_scope(true);
+  request.set_response_type(PayloadType::COMPRESSABLE);
+  PerformLargeUnary(&request, &response);
+  gpr_log(GPR_DEBUG, "Got username %s", response.username().c_str());
+  gpr_log(GPR_DEBUG, "Got oauth_scope %s", response.oauth_scope().c_str());
+  GPR_ASSERT(!response.username().empty());
+  GPR_ASSERT(response.username().c_str() == default_service_account);
+  GPR_ASSERT(!response.oauth_scope().empty());
+  const char* oauth_scope_str = response.oauth_scope().c_str();
+  GPR_ASSERT(oauth_scope.find(oauth_scope_str) != grpc::string::npos);
+  gpr_log(GPR_DEBUG, "Large unary with compute engine creds done.");
+}
+
+void InteropClient::DoOauth2AuthToken(const grpc::string& username,
+                                      const grpc::string& oauth_scope) {
+  gpr_log(GPR_DEBUG,
+          "Sending a unary rpc with raw oauth2 access token credentials ...");
+  SimpleRequest request;
+  SimpleResponse response;
+  request.set_fill_username(true);
+  request.set_fill_oauth_scope(true);
+
+  ClientContext context;
+
+  Status s = serviceStub_.Get()->UnaryCall(&context, request, &response);
+
+  AssertOkOrPrintErrorStatus(s);
+  GPR_ASSERT(!response.username().empty());
+  GPR_ASSERT(!response.oauth_scope().empty());
+  GPR_ASSERT(username == response.username());
+  const char* oauth_scope_str = response.oauth_scope().c_str();
+  GPR_ASSERT(oauth_scope.find(oauth_scope_str) != grpc::string::npos);
+  gpr_log(GPR_DEBUG, "Unary with oauth2 access token credentials done.");
+}
+
+void InteropClient::DoPerRpcCreds(const grpc::string& json_key) {
+  gpr_log(GPR_DEBUG, "Sending a unary rpc with per-rpc JWT access token ...");
+  SimpleRequest request;
+  SimpleResponse response;
+  request.set_fill_username(true);
+
+  ClientContext context;
+  std::chrono::seconds token_lifetime = std::chrono::hours(1);
+  std::shared_ptr<CallCredentials> creds =
+      ServiceAccountJWTAccessCredentials(json_key, token_lifetime.count());
+
+  context.set_credentials(creds);
+
+  Status s = serviceStub_.Get()->UnaryCall(&context, request, &response);
+
+  AssertOkOrPrintErrorStatus(s);
+  GPR_ASSERT(!response.username().empty());
+  GPR_ASSERT(json_key.find(response.username()) != grpc::string::npos);
+  gpr_log(GPR_DEBUG, "Unary with per-rpc JWT access token done.");
+}
+
+void InteropClient::DoJwtTokenCreds(const grpc::string& username) {
+  gpr_log(GPR_DEBUG,
+          "Sending a large unary rpc with JWT token credentials ...");
+  SimpleRequest request;
+  SimpleResponse response;
+  request.set_fill_username(true);
+  request.set_response_type(PayloadType::COMPRESSABLE);
+  PerformLargeUnary(&request, &response);
+  GPR_ASSERT(!response.username().empty());
+  GPR_ASSERT(username.find(response.username()) != grpc::string::npos);
+  gpr_log(GPR_DEBUG, "Large unary with JWT token creds done.");
+}
+
+void InteropClient::DoLargeUnary() {
+  gpr_log(GPR_DEBUG, "Sending a large unary rpc...");
+  SimpleRequest request;
+  SimpleResponse response;
+  request.set_response_type(PayloadType::COMPRESSABLE);
+  PerformLargeUnary(&request, &response);
+  gpr_log(GPR_DEBUG, "Large unary done.");
+}
+
+void InteropClient::DoLargeCompressedUnary() {
+  const CompressionType compression_types[] = {NONE, GZIP, DEFLATE};
+  const PayloadType payload_types[] = {COMPRESSABLE, UNCOMPRESSABLE, RANDOM};
+  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_types[j]).c_str(),
+                   PayloadType_Name(payload_types[i]).c_str());
+
+      gpr_log(GPR_DEBUG, "Sending a large compressed unary rpc %s.",
+              log_suffix);
+      SimpleRequest request;
+      SimpleResponse response;
+      request.set_response_type(payload_types[i]);
+      request.set_response_compression(compression_types[j]);
+      PerformLargeUnary(&request, &response, CompressionChecks);
+      gpr_log(GPR_DEBUG, "Large compressed unary done %s.", log_suffix);
+      gpr_free(log_suffix);
+    }
+  }
+}
+
+void InteropClient::DoRequestStreaming() {
+  gpr_log(GPR_DEBUG, "Sending request steaming rpc ...");
+
+  ClientContext context;
+  StreamingInputCallRequest request;
+  StreamingInputCallResponse response;
+
+  std::unique_ptr<ClientWriter<StreamingInputCallRequest>> stream(
+      serviceStub_.Get()->StreamingInputCall(&context, &response));
+
+  int aggregated_payload_size = 0;
+  for (unsigned int i = 0; i < request_stream_sizes.size(); ++i) {
+    Payload* payload = request.mutable_payload();
+    payload->set_body(grpc::string(request_stream_sizes[i], '\0'));
+    GPR_ASSERT(stream->Write(request));
+    aggregated_payload_size += request_stream_sizes[i];
+  }
+  stream->WritesDone();
+  Status s = stream->Finish();
+
+  GPR_ASSERT(response.aggregated_payload_size() == aggregated_payload_size);
+  AssertOkOrPrintErrorStatus(s);
+  gpr_log(GPR_DEBUG, "Request streaming done.");
+}
+
+void InteropClient::DoResponseStreaming() {
+  gpr_log(GPR_DEBUG, "Receiving response steaming rpc ...");
+
+  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(
+      serviceStub_.Get()->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_DEBUG, "Response streaming done.");
+}
+
+void InteropClient::DoResponseCompressedStreaming() {
+  const CompressionType compression_types[] = {NONE, GZIP, DEFLATE};
+  const PayloadType payload_types[] = {COMPRESSABLE, UNCOMPRESSABLE, RANDOM};
+  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_types[j]).c_str(),
+                   PayloadType_Name(payload_types[i]).c_str());
+
+      gpr_log(GPR_DEBUG, "Receiving response steaming rpc %s.", log_suffix);
+
+      request.set_response_type(payload_types[i]);
+      request.set_response_compression(compression_types[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.
+        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() ==
+                   GetInteropCompressionTypeFromCompressionAlgorithm(
+                       inspector.GetCallCompressionAlgorithm()));
+        if (request.response_compression() == NONE) {
+          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);
+        }
+
+        ++k;
+      }
+
+      GPR_ASSERT(response_stream_sizes.size() == k);
+      Status s = stream->Finish();
+
+      AssertOkOrPrintErrorStatus(s);
+      gpr_log(GPR_DEBUG, "Response streaming done %s.", log_suffix);
+      gpr_free(log_suffix);
+    }
+  }
+}
+
+void InteropClient::DoResponseStreamingWithSlowConsumer() {
+  gpr_log(GPR_DEBUG, "Receiving response steaming rpc with slow consumer ...");
+
+  ClientContext context;
+  StreamingOutputCallRequest request;
+
+  for (int i = 0; i < kNumResponseMessages; ++i) {
+    ResponseParameters* response_parameter = request.add_response_parameters();
+    response_parameter->set_size(kResponseMessageSize);
+  }
+  StreamingOutputCallResponse response;
+  std::unique_ptr<ClientReader<StreamingOutputCallResponse>> stream(
+      serviceStub_.Get()->StreamingOutputCall(&context, request));
+
+  int i = 0;
+  while (stream->Read(&response)) {
+    GPR_ASSERT(response.payload().body() ==
+               grpc::string(kResponseMessageSize, '\0'));
+    gpr_log(GPR_DEBUG, "received message %d", i);
+    usleep(kReceiveDelayMilliSeconds * 1000);
+    ++i;
+  }
+  GPR_ASSERT(kNumResponseMessages == i);
+  Status s = stream->Finish();
+
+  AssertOkOrPrintErrorStatus(s);
+  gpr_log(GPR_DEBUG, "Response streaming done.");
+}
+
+void InteropClient::DoHalfDuplex() {
+  gpr_log(GPR_DEBUG, "Sending half-duplex streaming rpc ...");
+
+  ClientContext context;
+  std::unique_ptr<ClientReaderWriter<StreamingOutputCallRequest,
+                                     StreamingOutputCallResponse>>
+      stream(serviceStub_.Get()->HalfDuplexCall(&context));
+
+  StreamingOutputCallRequest request;
+  ResponseParameters* response_parameter = request.add_response_parameters();
+  for (unsigned int i = 0; i < response_stream_sizes.size(); ++i) {
+    response_parameter->set_size(response_stream_sizes[i]);
+    GPR_ASSERT(stream->Write(request));
+  }
+  stream->WritesDone();
+
+  unsigned int i = 0;
+  StreamingOutputCallResponse response;
+  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_DEBUG, "Half-duplex streaming rpc done.");
+}
+
+void InteropClient::DoPingPong() {
+  gpr_log(GPR_DEBUG, "Sending Ping Pong streaming rpc ...");
+
+  ClientContext context;
+  std::unique_ptr<ClientReaderWriter<StreamingOutputCallRequest,
+                                     StreamingOutputCallResponse>>
+      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;
+  for (unsigned int i = 0; i < request_stream_sizes.size(); ++i) {
+    response_parameter->set_size(response_stream_sizes[i]);
+    payload->set_body(grpc::string(request_stream_sizes[i], '\0'));
+    GPR_ASSERT(stream->Write(request));
+    GPR_ASSERT(stream->Read(&response));
+    GPR_ASSERT(response.payload().body() ==
+               grpc::string(response_stream_sizes[i], '\0'));
+  }
+
+  stream->WritesDone();
+  GPR_ASSERT(!stream->Read(&response));
+  Status s = stream->Finish();
+  AssertOkOrPrintErrorStatus(s);
+  gpr_log(GPR_DEBUG, "Ping pong streaming done.");
+}
+
+void InteropClient::DoCancelAfterBegin() {
+  gpr_log(GPR_DEBUG, "Sending request steaming rpc ...");
+
+  ClientContext context;
+  StreamingInputCallRequest request;
+  StreamingInputCallResponse response;
+
+  std::unique_ptr<ClientWriter<StreamingInputCallRequest>> stream(
+      serviceStub_.Get()->StreamingInputCall(&context, &response));
+
+  gpr_log(GPR_DEBUG, "Trying to cancel...");
+  context.TryCancel();
+  Status s = stream->Finish();
+  GPR_ASSERT(s.error_code() == StatusCode::CANCELLED);
+  gpr_log(GPR_DEBUG, "Canceling streaming done.");
+}
+
+void InteropClient::DoCancelAfterFirstResponse() {
+  gpr_log(GPR_DEBUG, "Sending Ping Pong streaming rpc ...");
+
+  ClientContext context;
+  std::unique_ptr<ClientReaderWriter<StreamingOutputCallRequest,
+                                     StreamingOutputCallResponse>>
+      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'));
+  StreamingOutputCallResponse response;
+  GPR_ASSERT(stream->Write(request));
+  GPR_ASSERT(stream->Read(&response));
+  GPR_ASSERT(response.payload().body() == grpc::string(31415, '\0'));
+  gpr_log(GPR_DEBUG, "Trying to cancel...");
+  context.TryCancel();
+
+  Status s = stream->Finish();
+  gpr_log(GPR_DEBUG, "Canceling pingpong streaming done.");
+}
+
+void InteropClient::DoTimeoutOnSleepingServer() {
+  gpr_log(GPR_DEBUG,
+          "Sending Ping Pong streaming rpc with a short deadline...");
+
+  ClientContext context;
+  std::chrono::system_clock::time_point deadline =
+      std::chrono::system_clock::now() + std::chrono::milliseconds(1);
+  context.set_deadline(deadline);
+  std::unique_ptr<ClientReaderWriter<StreamingOutputCallRequest,
+                                     StreamingOutputCallResponse>>
+      stream(serviceStub_.Get()->FullDuplexCall(&context));
+
+  StreamingOutputCallRequest request;
+  request.mutable_payload()->set_body(grpc::string(27182, '\0'));
+  stream->Write(request);
+
+  Status s = stream->Finish();
+  GPR_ASSERT(s.error_code() == StatusCode::DEADLINE_EXCEEDED);
+  gpr_log(GPR_DEBUG, "Pingpong streaming timeout done.");
+}
+
+void InteropClient::DoEmptyStream() {
+  gpr_log(GPR_DEBUG, "Starting empty_stream.");
+
+  ClientContext context;
+  std::unique_ptr<ClientReaderWriter<StreamingOutputCallRequest,
+                                     StreamingOutputCallResponse>>
+      stream(serviceStub_.Get()->FullDuplexCall(&context));
+  stream->WritesDone();
+  StreamingOutputCallResponse response;
+  GPR_ASSERT(stream->Read(&response) == false);
+  Status s = stream->Finish();
+  AssertOkOrPrintErrorStatus(s);
+  gpr_log(GPR_DEBUG, "empty_stream done.");
+}
+
+void InteropClient::DoStatusWithMessage() {
+  gpr_log(GPR_DEBUG,
+          "Sending RPC with a request for status code 2 and message");
+
+  ClientContext context;
+  SimpleRequest request;
+  SimpleResponse response;
+  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);
+
+  Status s = serviceStub_.Get()->UnaryCall(&context, request, &response);
+
+  GPR_ASSERT(s.error_code() == grpc::StatusCode::UNKNOWN);
+  GPR_ASSERT(s.error_message() == test_msg);
+  gpr_log(GPR_DEBUG, "Done testing Status and Message");
+}
+
+void InteropClient::DoCustomMetadata() {
+  const grpc::string kEchoInitialMetadataKey("x-grpc-test-echo-initial");
+  const grpc::string kInitialMetadataValue("test_initial_metadata_value");
+  const grpc::string kEchoTrailingBinMetadataKey(
+      "x-grpc-test-echo-trailing-bin");
+  const grpc::string kTrailingBinValue("\x0a\x0b\x0a\x0b\x0a\x0b");
+  ;
+
+  {
+    gpr_log(GPR_DEBUG, "Sending RPC with custom metadata");
+    ClientContext context;
+    context.AddMetadata(kEchoInitialMetadataKey, kInitialMetadataValue);
+    context.AddMetadata(kEchoTrailingBinMetadataKey, kTrailingBinValue);
+    SimpleRequest request;
+    SimpleResponse response;
+    request.set_response_size(kLargeResponseSize);
+    grpc::string payload(kLargeRequestSize, '\0');
+    request.mutable_payload()->set_body(payload.c_str(), kLargeRequestSize);
+
+    Status s = serviceStub_.Get()->UnaryCall(&context, request, &response);
+    AssertOkOrPrintErrorStatus(s);
+    const auto& server_initial_metadata = context.GetServerInitialMetadata();
+    auto iter = server_initial_metadata.find(kEchoInitialMetadataKey);
+    GPR_ASSERT(iter != server_initial_metadata.end());
+    GPR_ASSERT(iter->second.data() == kInitialMetadataValue);
+    const auto& server_trailing_metadata = context.GetServerTrailingMetadata();
+    iter = server_trailing_metadata.find(kEchoTrailingBinMetadataKey);
+    GPR_ASSERT(iter != server_trailing_metadata.end());
+    GPR_ASSERT(grpc::string(iter->second.begin(), iter->second.end()) ==
+               kTrailingBinValue);
+
+    gpr_log(GPR_DEBUG, "Done testing RPC with custom metadata");
+  }
+
+  {
+    gpr_log(GPR_DEBUG, "Sending stream with custom metadata");
+    ClientContext context;
+    context.AddMetadata(kEchoInitialMetadataKey, kInitialMetadataValue);
+    context.AddMetadata(kEchoTrailingBinMetadataKey, kTrailingBinValue);
+    std::unique_ptr<ClientReaderWriter<StreamingOutputCallRequest,
+                                       StreamingOutputCallResponse>>
+        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');
+    request.mutable_payload()->set_body(payload.c_str(), kLargeRequestSize);
+    StreamingOutputCallResponse response;
+    GPR_ASSERT(stream->Write(request));
+    stream->WritesDone();
+    GPR_ASSERT(stream->Read(&response));
+    GPR_ASSERT(response.payload().body() ==
+               grpc::string(kLargeResponseSize, '\0'));
+    GPR_ASSERT(!stream->Read(&response));
+    Status s = stream->Finish();
+    AssertOkOrPrintErrorStatus(s);
+    const auto& server_initial_metadata = context.GetServerInitialMetadata();
+    auto iter = server_initial_metadata.find(kEchoInitialMetadataKey);
+    GPR_ASSERT(iter != server_initial_metadata.end());
+    GPR_ASSERT(iter->second.data() == kInitialMetadataValue);
+    const auto& server_trailing_metadata = context.GetServerTrailingMetadata();
+    iter = server_trailing_metadata.find(kEchoTrailingBinMetadataKey);
+    GPR_ASSERT(iter != server_trailing_metadata.end());
+    GPR_ASSERT(grpc::string(iter->second.begin(), iter->second.end()) ==
+               kTrailingBinValue);
+
+    gpr_log(GPR_DEBUG, "Done testing stream with custom metadata");
+  }
+}
+
+}  // namespace testing
+}  // namespace grpc