Support bidirectional stream RPC

Add bidistream test

Add bad request tests

Replace deprecated DynamicMessageFactory constructor

6 files changed, 308 insertions, 78 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;
diff --git a/test/cpp/util/cli_call.h b/test/cpp/util/cli_call.h
index 34fa88433f..91f0dbc9ed 100644
--- a/test/cpp/util/cli_call.h
+++ b/test/cpp/util/cli_call.h
@@ -48,6 +48,9 @@ class ClientContext;
 
 namespace testing {
 
+// CliCall handles the sending and receiving of generic messages given the name
+// of the remote method. This class is only used by GrpcTool. Its thread-safe
+// and thread-unsafe methods should not be used together.
 class CliCall final {
  public:
   typedef std::multimap<grpc::string, grpc::string> OutgoingMetadataContainer;
@@ -56,7 +59,9 @@ class CliCall final {
 
   CliCall(std::shared_ptr<grpc::Channel> channel, const grpc::string& method,
           const OutgoingMetadataContainer& metadata);
+  ~CliCall();
 
+  // Perform an unary generic RPC.
   static Status Call(std::shared_ptr<grpc::Channel> channel,
                      const grpc::string& method, const grpc::string& request,
                      grpc::string* response,
@@ -64,13 +69,32 @@ class CliCall final {
                      IncomingMetadataContainer* server_initial_metadata,
                      IncomingMetadataContainer* server_trailing_metadata);
 
+  // Send a generic request message in a synchronous manner. NOT thread-safe.
   void Write(const grpc::string& request);
 
+  // Send a generic request message in a synchronous manner. NOT thread-safe.
   void WritesDone();
 
+  // Receive a generic response message in a synchronous manner.NOT thread-safe.
   bool Read(grpc::string* response,
             IncomingMetadataContainer* server_initial_metadata);
 
+  // Thread-safe write. Must be used with ReadAndMaybeNotifyWrite. Send out a
+  // generic request message and wait for ReadAndMaybeNotifyWrite to finish it.
+  void WriteAndWait(const grpc::string& request);
+
+  // Thread-safe WritesDone. Must be used with ReadAndMaybeNotifyWrite. Send out
+  // WritesDone for gereneric request messages and wait for
+  // ReadAndMaybeNotifyWrite to finish it.
+  void WritesDoneAndWait();
+
+  // Thread-safe Read. Blockingly receive a generic response message. Notify
+  // writes if they are finished when this read is waiting for a resposne.
+  bool ReadAndMaybeNotifyWrite(
+      grpc::string* response,
+      IncomingMetadataContainer* server_initial_metadata);
+
+  // Finish the RPC.
   Status Finish(IncomingMetadataContainer* server_trailing_metadata);
 
  private:
@@ -78,6 +102,9 @@ class CliCall final {
   grpc::ClientContext ctx_;
   std::unique_ptr<grpc::GenericClientAsyncReaderWriter> call_;
   grpc::CompletionQueue cq_;
+  gpr_mu write_mu_;
+  gpr_cv write_cv_;  // Protected by write_mu_;
+  bool write_done_;  // Portected by write_mu_;
 };
 
 }  // namespace testing
diff --git a/test/cpp/util/grpc_cli.cc b/test/cpp/util/grpc_cli.cc
index fe68ccb619..a78bed4b90 100644
--- a/test/cpp/util/grpc_cli.cc
+++ b/test/cpp/util/grpc_cli.cc
@@ -85,7 +85,7 @@ static bool SimplePrint(const grpc::string& outfile,
   if (outfile.empty()) {
     std::cout << output << std::endl;
   } else {
-    std::ofstream output_file(outfile, std::ios::trunc | std::ios::binary);
+    std::ofstream output_file(outfile, std::ios::app | std::ios::binary);
     output_file << output << std::endl;
     output_file.close();
   }
diff --git a/test/cpp/util/grpc_tool.cc b/test/cpp/util/grpc_tool.cc
index 762f8e8c23..39acd8eb4b 100644
--- a/test/cpp/util/grpc_tool.cc
+++ b/test/cpp/util/grpc_tool.cc
@@ -39,6 +39,7 @@
 #include <memory>
 #include <sstream>
 #include <string>
+#include <thread>
 
 #include <gflags/gflags.h>
 #include <grpc++/channel.h>
@@ -159,6 +160,36 @@ void PrintMetadata(const T& m, const grpc::string& message) {
   }
 }
 
+void ReadResponse(CliCall* call, const grpc::string& method_name,
+                  GrpcToolOutputCallback callback, ProtoFileParser* parser,
+                  gpr_mu* parser_mu, bool print_mode) {
+  grpc::string serialized_response_proto;
+  std::multimap<grpc::string_ref, grpc::string_ref> server_initial_metadata;
+
+  for (bool receive_initial_metadata = true; call->ReadAndMaybeNotifyWrite(
+           &serialized_response_proto,
+           receive_initial_metadata ? &server_initial_metadata : nullptr);
+       receive_initial_metadata = false) {
+    fprintf(stderr, "got response.\n");
+    if (!FLAGS_binary_output) {
+      gpr_mu_lock(parser_mu);
+      serialized_response_proto = parser->GetTextFormatFromMethod(
+          method_name, serialized_response_proto, false /* is_request */);
+      if (parser->HasError() && print_mode) {
+        fprintf(stderr, "Failed to parse response.\n");
+      }
+      gpr_mu_unlock(parser_mu);
+    }
+    if (receive_initial_metadata) {
+      PrintMetadata(server_initial_metadata,
+                    "Received initial metadata from server:");
+    }
+    if (!callback(serialized_response_proto) && print_mode) {
+      fprintf(stderr, "Failed to output response.\n");
+    }
+  }
+}
+
 struct Command {
   const char* command;
   std::function<bool(GrpcTool*, int, const char**, const CliCredentials&,
@@ -416,7 +447,7 @@ bool GrpcTool::CallMethod(int argc, const char** argv,
   grpc::string server_address(argv[0]);
   grpc::string method_name(argv[1]);
   grpc::string formatted_method_name;
-  std::unique_ptr<grpc::testing::ProtoFileParser> parser;
+  std::unique_ptr<ProtoFileParser> parser;
   grpc::string serialized_request_proto;
   bool print_mode = false;
 
@@ -428,21 +459,17 @@ bool GrpcTool::CallMethod(int argc, const char** argv,
   parser.reset(new grpc::testing::ProtoFileParser(channel, FLAGS_proto_path,
                                                   FLAGS_protofiles));
 
-  grpc::string formated_method_name =
-      parser->GetFormatedMethodName(method_name);
+  if (FLAGS_binary_input) {
+    formatted_method_name = method_name;
+  } else {
+    formatted_method_name = parser->GetFormattedMethodName(method_name);
+  }
 
   if (parser->HasError()) {
     return false;
   }
 
   if (parser->IsStreaming(method_name, true /* is_request */)) {
-    // TODO(zyc): Support BidiStream
-    if (parser->IsStreaming(method_name, false /* is_request */)) {
-      fprintf(stderr,
-              "Bidirectional-streaming method is not supported.");
-      return false;
-    }
-
     std::istream* input_stream;
     std::ifstream input_file;
 
@@ -454,7 +481,7 @@ bool GrpcTool::CallMethod(int argc, const char** argv,
     ParseMetadataFlag(&client_metadata);
     PrintMetadata(client_metadata, "Sending client initial metadata:");
 
-    CliCall call(channel, formated_method_name, client_metadata);
+    CliCall call(channel, formatted_method_name, client_metadata);
 
     if (FLAGS_infile.empty()) {
       if (isatty(STDIN_FILENO)) {
@@ -467,6 +494,11 @@ bool GrpcTool::CallMethod(int argc, const char** argv,
       input_stream = &input_file;
     }
 
+    gpr_mu parser_mu;
+    gpr_mu_init(&parser_mu);
+    std::thread read_thread(ReadResponse, &call, method_name, callback,
+                            parser.get(), &parser_mu, print_mode);
+
     std::stringstream request_ss;
     grpc::string line;
     while (!request_text.empty() ||
@@ -476,6 +508,7 @@ bool GrpcTool::CallMethod(int argc, const char** argv,
           serialized_request_proto = request_text;
           request_text.clear();
         } else {
+          gpr_mu_lock(&parser_mu);
           serialized_request_proto = parser->GetSerializedProtoFromMethod(
               method_name, request_text, true /* is_request */);
           request_text.clear();
@@ -483,11 +516,13 @@ bool GrpcTool::CallMethod(int argc, const char** argv,
             if (print_mode) {
               fprintf(stderr, "Failed to parse request.\n");
             }
+            gpr_mu_unlock(&parser_mu);
             continue;
           }
+          gpr_mu_unlock(&parser_mu);
         }
 
-        call.Write(serialized_request_proto);
+        call.WriteAndWait(serialized_request_proto);
         if (print_mode) {
           fprintf(stderr, "Request sent.\n");
         }
@@ -505,35 +540,21 @@ bool GrpcTool::CallMethod(int argc, const char** argv,
       input_file.close();
     }
 
-    call.WritesDone();
+    call.WritesDoneAndWait();
+    read_thread.join();
 
-    grpc::string serialized_response_proto;
-    std::multimap<grpc::string_ref, grpc::string_ref> server_initial_metadata,
-        server_trailing_metadata;
-    if (!call.Read(&serialized_response_proto, &server_trailing_metadata)) {
-      fprintf(stderr, "Failed to read response.\n");
-    }
+    std::multimap<grpc::string_ref, grpc::string_ref> server_trailing_metadata;
     Status status = call.Finish(&server_trailing_metadata);
-
-    PrintMetadata(server_initial_metadata,
-                  "Received initial metadata from server:");
     PrintMetadata(server_trailing_metadata,
                   "Received trailing metadata from server:");
+
     if (status.ok()) {
       fprintf(stderr, "Stream RPC succeeded with OK status\n");
-      if (FLAGS_binary_output) {
-        output_ss << serialized_response_proto;
-      } else {
-        grpc::string response_text = parser->GetTextFormatFromMethod(
-            method_name, serialized_response_proto, false /* is_request */);
-        if (parser->HasError()) {
-          return false;
-        }
-        output_ss << response_text;
-      }
+      return true;
     } else {
       fprintf(stderr, "Rpc failed with status code %d, error message: %s\n",
               status.error_code(), status.error_message().c_str());
+      return false;
     }
 
   } else {  // parser->IsStreaming(method_name, true /* is_request */)
@@ -559,7 +580,9 @@ bool GrpcTool::CallMethod(int argc, const char** argv,
 
     if (FLAGS_binary_input) {
       serialized_request_proto = request_text;
+      // formatted_method_name = method_name;
     } else {
+      // formatted_method_name = parser->GetFormattedMethodName(method_name);
       serialized_request_proto = parser->GetSerializedProtoFromMethod(
           method_name, request_text, true /* is_request */);
       if (parser->HasError()) {
@@ -575,7 +598,7 @@ bool GrpcTool::CallMethod(int argc, const char** argv,
     ParseMetadataFlag(&client_metadata);
     PrintMetadata(client_metadata, "Sending client initial metadata:");
 
-    CliCall call(channel, formated_method_name, client_metadata);
+    CliCall call(channel, formatted_method_name, client_metadata);
     call.Write(serialized_request_proto);
     call.WritesDone();
 
@@ -608,7 +631,7 @@ bool GrpcTool::CallMethod(int argc, const char** argv,
       return false;
     }
   }
-  return callback(output_ss.str());
+  GPR_UNREACHABLE_CODE(return false);
 }
 
 bool GrpcTool::ParseMessage(int argc, const char** argv,
diff --git a/test/cpp/util/grpc_tool_test.cc b/test/cpp/util/grpc_tool_test.cc
index e2eebd4089..26e2b1f502 100644
--- a/test/cpp/util/grpc_tool_test.cc
+++ b/test/cpp/util/grpc_tool_test.cc
@@ -142,7 +142,7 @@ class TestServiceImpl : public ::grpc::testing::EchoTestService::Service {
 
   Status RequestStream(ServerContext* context,
                        ServerReader<EchoRequest>* reader,
-                       EchoResponse* response) GRPC_OVERRIDE {
+                       EchoResponse* response) override {
     EchoRequest request;
     response->set_message("");
     if (!context->client_metadata().empty()) {
@@ -162,7 +162,7 @@ class TestServiceImpl : public ::grpc::testing::EchoTestService::Service {
   }
 
   Status ResponseStream(ServerContext* context, const EchoRequest* request,
-                        ServerWriter<EchoResponse>* writer) GRPC_OVERRIDE {
+                        ServerWriter<EchoResponse>* writer) override {
     if (!context->client_metadata().empty()) {
       for (std::multimap<grpc::string_ref, grpc::string_ref>::const_iterator
                iter = context->client_metadata().begin();
@@ -181,6 +181,29 @@ class TestServiceImpl : public ::grpc::testing::EchoTestService::Service {
 
     return Status::OK;
   }
+
+  Status BidiStream(
+      ServerContext* context,
+      ServerReaderWriter<EchoResponse, EchoRequest>* stream) override {
+    EchoRequest request;
+    EchoResponse response;
+    if (!context->client_metadata().empty()) {
+      for (std::multimap<grpc::string_ref, grpc::string_ref>::const_iterator
+               iter = context->client_metadata().begin();
+           iter != context->client_metadata().end(); ++iter) {
+        context->AddInitialMetadata(ToString(iter->first),
+                                    ToString(iter->second));
+      }
+    }
+    context->AddTrailingMetadata("trailing_key", "trailing_value");
+
+    while (stream->Read(&request)) {
+      response.set_message(request.message());
+      stream->Write(response);
+    }
+
+    return Status::OK;
+  }
 };
 
 }  // namespace
@@ -391,48 +414,32 @@ TEST_F(GrpcToolTest, CallCommand) {
   ShutdownServer();
 }
 
-TEST_F(GrpcToolTest, ParseCommand) {
-  // Test input "grpc_cli parse localhost:<port> grpc.testing.EchoResponse
-  // ECHO_RESPONSE_MESSAGE"
+TEST_F(GrpcToolTest, CallCommandRequestStream) {
+  // Test input: grpc_cli call localhost:<port> RequestStream "message:
+  // 'Hello0'"
   std::stringstream output_stream;
-  std::stringstream binary_output_stream;
 
   const grpc::string server_address = SetUpServer();
-  const char* argv[] = {"grpc_cli", "parse", server_address.c_str(),
-                        "grpc.testing.EchoResponse", ECHO_RESPONSE_MESSAGE};
+  const char* argv[] = {"grpc_cli", "call", server_address.c_str(),
+                        "RequestStream", "message: 'Hello0'"};
 
-  FLAGS_binary_input = false;
-  FLAGS_binary_output = false;
-  EXPECT_TRUE(0 == GrpcToolMainLib(ArraySize(argv), argv, TestCliCredentials(),
-                                   std::bind(PrintStream, &output_stream,
-                                             std::placeholders::_1)));
-  // Expected output: ECHO_RESPONSE_MESSAGE
-  EXPECT_TRUE(0 == strcmp(output_stream.str().c_str(), ECHO_RESPONSE_MESSAGE));
+  // Mock std::cin input "message: 'Hello1'\n\n message: 'Hello2'\n\n"
+  std::streambuf* orig = std::cin.rdbuf();
+  std::istringstream ss("message: 'Hello1'\n\n message: 'Hello2'\n\n");
+  std::cin.rdbuf(ss.rdbuf());
 
-  // Parse text message to binary message and then parse it back to text message
-  output_stream.str(grpc::string());
-  output_stream.clear();
-  FLAGS_binary_output = true;
   EXPECT_TRUE(0 == GrpcToolMainLib(ArraySize(argv), argv, TestCliCredentials(),
                                    std::bind(PrintStream, &output_stream,
                                              std::placeholders::_1)));
-  grpc::string binary_data = output_stream.str();
-  output_stream.str(grpc::string());
-  output_stream.clear();
-  argv[4] = binary_data.c_str();
-  FLAGS_binary_input = true;
-  FLAGS_binary_output = false;
-  EXPECT_TRUE(0 == GrpcToolMainLib(5, argv, TestCliCredentials(),
-                                   std::bind(PrintStream, &output_stream,
-                                             std::placeholders::_1)));
-
-  // Expected output: ECHO_RESPONSE_MESSAGE
-  EXPECT_TRUE(0 == strcmp(output_stream.str().c_str(), ECHO_RESPONSE_MESSAGE));
 
+  // Expected output: "message: \"Hello0Hello1Hello2\""
+  EXPECT_TRUE(NULL != strstr(output_stream.str().c_str(),
+                             "message: \"Hello0Hello1Hello2\""));
+  std::cin.rdbuf(orig);
   ShutdownServer();
 }
 
-TEST_F(GrpcToolTest, CallCommandRequestStream) {
+TEST_F(GrpcToolTest, CallCommandRequestStreamWithBadRequest) {
   // Test input: grpc_cli call localhost:<port> RequestStream "message:
   // 'Hello0'"
   std::stringstream output_stream;
@@ -441,18 +448,18 @@ TEST_F(GrpcToolTest, CallCommandRequestStream) {
   const char* argv[] = {"grpc_cli", "call", server_address.c_str(),
                         "RequestStream", "message: 'Hello0'"};
 
-  // Mock std::cin input "message: 'Hello1'\n\n message: 'Hello2'\n\n"
+  // Mock std::cin input "bad_field: 'Hello1'\n\n message: 'Hello2'\n\n"
   std::streambuf* orig = std::cin.rdbuf();
-  std::istringstream ss("message: 'Hello1'\n\n message: 'Hello2'\n\n");
+  std::istringstream ss("bad_field: 'Hello1'\n\n message: 'Hello2'\n\n");
   std::cin.rdbuf(ss.rdbuf());
 
   EXPECT_TRUE(0 == GrpcToolMainLib(ArraySize(argv), argv, TestCliCredentials(),
                                    std::bind(PrintStream, &output_stream,
                                              std::placeholders::_1)));
 
-  // Expected output: "message: \"Hello0Hello1Hello2\""
-  EXPECT_TRUE(NULL != strstr(output_stream.str().c_str(),
-                             "message: \"Hello0Hello1Hello2\""));
+  // Expected output: "message: \"Hello0Hello2\""
+  EXPECT_TRUE(NULL !=
+              strstr(output_stream.str().c_str(), "message: \"Hello0Hello2\""));
   std::cin.rdbuf(orig);
   ShutdownServer();
 }
@@ -470,12 +477,10 @@ TEST_F(GrpcToolTest, CallCommandResponseStream) {
                                    std::bind(PrintStream, &output_stream,
                                              std::placeholders::_1)));
 
-  fprintf(stderr, "%s\n", output_stream.str().c_str());
   // Expected output: "message: \"Hello{n}\""
-
   for (int i = 0; i < kNumResponseStreamsMsgs; i++) {
     grpc::string expected_response_text =
-        "message: \"Hello" + grpc::to_string(i) + "\"\n\n";
+        "message: \"Hello" + grpc::to_string(i) + "\"\n";
     EXPECT_TRUE(NULL != strstr(output_stream.str().c_str(),
                                expected_response_text.c_str()));
   }
@@ -483,6 +488,99 @@ TEST_F(GrpcToolTest, CallCommandResponseStream) {
   ShutdownServer();
 }
 
+TEST_F(GrpcToolTest, CallCommandBidiStream) {
+  // Test input: grpc_cli call localhost:<port> BidiStream "message: 'Hello0'"
+  std::stringstream output_stream;
+
+  const grpc::string server_address = SetUpServer();
+  const char* argv[] = {"grpc_cli", "call", server_address.c_str(),
+                        "BidiStream", "message: 'Hello0'"};
+
+  // Mock std::cin input "message: 'Hello1'\n\n message: 'Hello2'\n\n"
+  std::streambuf* orig = std::cin.rdbuf();
+  std::istringstream ss("message: 'Hello1'\n\n message: 'Hello2'\n\n");
+  std::cin.rdbuf(ss.rdbuf());
+
+  EXPECT_TRUE(0 == GrpcToolMainLib(ArraySize(argv), argv, TestCliCredentials(),
+                                   std::bind(PrintStream, &output_stream,
+                                             std::placeholders::_1)));
+
+  // Expected output: "message: \"Hello0\"\nmessage: \"Hello1\"\nmessage:
+  // \"Hello2\"\n\n"
+  EXPECT_TRUE(NULL != strstr(output_stream.str().c_str(),
+                             "message: \"Hello0\"\nmessage: "
+                             "\"Hello1\"\nmessage: \"Hello2\"\n"));
+  std::cin.rdbuf(orig);
+  ShutdownServer();
+}
+
+TEST_F(GrpcToolTest, CallCommandBidiStreamWithBadRequest) {
+  // Test input: grpc_cli call localhost:<port> BidiStream "message: 'Hello0'"
+  std::stringstream output_stream;
+
+  const grpc::string server_address = SetUpServer();
+  const char* argv[] = {"grpc_cli", "call", server_address.c_str(),
+                        "BidiStream", "message: 'Hello0'"};
+
+  // Mock std::cin input "message: 'Hello1'\n\n message: 'Hello2'\n\n"
+  std::streambuf* orig = std::cin.rdbuf();
+  std::istringstream ss("message: 1.0\n\n message: 'Hello2'\n\n");
+  std::cin.rdbuf(ss.rdbuf());
+
+  EXPECT_TRUE(0 == GrpcToolMainLib(ArraySize(argv), argv, TestCliCredentials(),
+                                   std::bind(PrintStream, &output_stream,
+                                             std::placeholders::_1)));
+
+  // Expected output: "message: \"Hello0\"\nmessage: \"Hello1\"\nmessage:
+  // \"Hello2\"\n\n"
+  EXPECT_TRUE(NULL != strstr(output_stream.str().c_str(),
+                             "message: \"Hello0\"\nmessage: \"Hello2\"\n"));
+  std::cin.rdbuf(orig);
+
+  ShutdownServer();
+}
+
+TEST_F(GrpcToolTest, ParseCommand) {
+  // Test input "grpc_cli parse localhost:<port> grpc.testing.EchoResponse
+  // ECHO_RESPONSE_MESSAGE"
+  std::stringstream output_stream;
+  std::stringstream binary_output_stream;
+
+  const grpc::string server_address = SetUpServer();
+  const char* argv[] = {"grpc_cli", "parse", server_address.c_str(),
+                        "grpc.testing.EchoResponse", ECHO_RESPONSE_MESSAGE};
+
+  FLAGS_binary_input = false;
+  FLAGS_binary_output = false;
+  EXPECT_TRUE(0 == GrpcToolMainLib(ArraySize(argv), argv, TestCliCredentials(),
+                                   std::bind(PrintStream, &output_stream,
+                                             std::placeholders::_1)));
+  // Expected output: ECHO_RESPONSE_MESSAGE
+  EXPECT_TRUE(0 == strcmp(output_stream.str().c_str(), ECHO_RESPONSE_MESSAGE));
+
+  // Parse text message to binary message and then parse it back to text message
+  output_stream.str(grpc::string());
+  output_stream.clear();
+  FLAGS_binary_output = true;
+  EXPECT_TRUE(0 == GrpcToolMainLib(ArraySize(argv), argv, TestCliCredentials(),
+                                   std::bind(PrintStream, &output_stream,
+                                             std::placeholders::_1)));
+  grpc::string binary_data = output_stream.str();
+  output_stream.str(grpc::string());
+  output_stream.clear();
+  argv[4] = binary_data.c_str();
+  FLAGS_binary_input = true;
+  FLAGS_binary_output = false;
+  EXPECT_TRUE(0 == GrpcToolMainLib(5, argv, TestCliCredentials(),
+                                   std::bind(PrintStream, &output_stream,
+                                             std::placeholders::_1)));
+
+  // Expected output: ECHO_RESPONSE_MESSAGE
+  EXPECT_TRUE(0 == strcmp(output_stream.str().c_str(), ECHO_RESPONSE_MESSAGE));
+
+  ShutdownServer();
+}
+
 TEST_F(GrpcToolTest, TooFewArguments) {
   // Test input "grpc_cli call Echo"
   std::stringstream output_stream;
diff --git a/test/cpp/util/proto_file_parser.cc b/test/cpp/util/proto_file_parser.cc
index 9f1f05595e..d501c3697b 100644
--- a/test/cpp/util/proto_file_parser.cc
+++ b/test/cpp/util/proto_file_parser.cc
@@ -81,8 +81,9 @@ class ErrorPrinter : public protobuf::compiler::MultiFileErrorCollector {
 ProtoFileParser::ProtoFileParser(std::shared_ptr<grpc::Channel> channel,
                                  const grpc::string& proto_path,
                                  const grpc::string& protofiles)
-    : has_error_(false) {
-  std::vector<grpc::string> service_list;
+    : has_error_(false),
+      dynamic_factory_(new protobuf::DynamicMessageFactory()) {
+  std::vector<std::string> service_list;
   if (channel) {
     reflection_db_.reset(new grpc::ProtoReflectionDescriptorDatabase(channel));
     reflection_db_->GetServices(&service_list);
@@ -127,7 +128,6 @@ ProtoFileParser::ProtoFileParser(std::shared_ptr<grpc::Channel> channel,
   }
 
   desc_pool_.reset(new protobuf::DescriptorPool(desc_db_.get()));
-  dynamic_factory_.reset(new protobuf::DynamicMessageFactory(desc_pool_.get()));
 
   for (auto it = service_list.begin(); it != service_list.end(); it++) {
     if (known_services.find(*it) == known_services.end()) {