Support bidirectional stream RPC

Add bidistream test

Add bad request tests

Replace deprecated DynamicMessageFactory constructor

1 files changed, 82 insertions, 0 deletions

diff --git a/test/cpp/util/cli_call.cc b/test/cpp/util/cli_call.cc
index 1101abe3c9..4d045da098 100644
--- a/test/cpp/util/cli_call.cc
+++ b/test/cpp/util/cli_call.cc
@@ -67,6 +67,8 @@ CliCall::CliCall(std::shared_ptr<grpc::Channel> channel,
                  const grpc::string& method,
                  const OutgoingMetadataContainer& metadata)
     : stub_(new grpc::GenericStub(channel)) {
+  gpr_mu_init(&write_mu_);
+  gpr_cv_init(&write_cv_);
   if (!metadata.empty()) {
     for (OutgoingMetadataContainer::const_iterator iter = metadata.begin();
          iter != metadata.end(); ++iter) {
@@ -80,6 +82,11 @@ CliCall::CliCall(std::shared_ptr<grpc::Channel> channel,
   GPR_ASSERT(ok);
 }
 
+CliCall::~CliCall() {
+  gpr_cv_destroy(&write_cv_);
+  gpr_mu_destroy(&write_mu_);
+}
+
 void CliCall::Write(const grpc::string& request) {
   void* got_tag;
   bool ok;
@@ -126,6 +133,81 @@ void CliCall::WritesDone() {
   GPR_ASSERT(ok);
 }
 
+void CliCall::WriteAndWait(const grpc::string& request) {
+  grpc_slice s = grpc_slice_from_copied_string(request.c_str());
+  grpc::Slice req_slice(s, grpc::Slice::STEAL_REF);
+  grpc::ByteBuffer send_buffer(&req_slice, 1);
+
+  gpr_mu_lock(&write_mu_);
+  call_->Write(send_buffer, tag(2));
+  write_done_ = false;
+  while (!write_done_) {
+    gpr_cv_wait(&write_cv_, &write_mu_, gpr_inf_future(GPR_CLOCK_REALTIME));
+  }
+  gpr_mu_unlock(&write_mu_);
+}
+
+void CliCall::WritesDoneAndWait() {
+  gpr_mu_lock(&write_mu_);
+  call_->WritesDone(tag(4));
+  write_done_ = false;
+  while (!write_done_) {
+    gpr_cv_wait(&write_cv_, &write_mu_, gpr_inf_future(GPR_CLOCK_REALTIME));
+  }
+  gpr_mu_unlock(&write_mu_);
+}
+
+bool CliCall::ReadAndMaybeNotifyWrite(
+    grpc::string* response,
+    IncomingMetadataContainer* server_initial_metadata) {
+  void* got_tag;
+  bool ok;
+  grpc::ByteBuffer recv_buffer;
+
+  call_->Read(&recv_buffer, tag(3));
+  bool cq_result = cq_.Next(&got_tag, &ok);
+
+  while (got_tag != tag(3)) {
+    gpr_mu_lock(&write_mu_);
+    write_done_ = true;
+    gpr_cv_signal(&write_cv_);
+    gpr_mu_unlock(&write_mu_);
+
+    cq_result = cq_.Next(&got_tag, &ok);
+    if (got_tag == tag(2)) {
+      GPR_ASSERT(ok);
+    }
+  }
+
+  if (!cq_result || !ok) {
+    // If the RPC is ended on the server side, we should still wait for the
+    // pending write on the client side to be done.
+    if (!ok) {
+      gpr_mu_lock(&write_mu_);
+      if (!write_done_) {
+        cq_.Next(&got_tag, &ok);
+        GPR_ASSERT(got_tag != tag(2));
+        write_done_ = true;
+        gpr_cv_signal(&write_cv_);
+      }
+      gpr_mu_unlock(&write_mu_);
+    }
+    return false;
+  }
+
+  std::vector<grpc::Slice> slices;
+  recv_buffer.Dump(&slices);
+  response->clear();
+  for (size_t i = 0; i < slices.size(); i++) {
+    response->append(reinterpret_cast<const char*>(slices[i].begin()),
+                     slices[i].size());
+  }
+  if (server_initial_metadata) {
+    *server_initial_metadata = ctx_.GetServerInitialMetadata();
+  }
+  return true;
+}
+
 Status CliCall::Finish(IncomingMetadataContainer* server_trailing_metadata) {
   void* got_tag;
   bool ok;