Added streaming C++ tests for sync and sync cases
diff --git a/test/cpp/qps/client_sync.cc b/test/cpp/qps/client_sync.cc
index 7bb7231..1e14aa8 100644
--- a/test/cpp/qps/client_sync.cc
+++ b/test/cpp/qps/client_sync.cc
@@ -48,9 +48,11 @@
#include <grpc/support/host_port.h>
#include <gflags/gflags.h>
#include <grpc++/client_context.h>
-#include <grpc++/status.h>
#include <grpc++/server.h>
#include <grpc++/server_builder.h>
+#include <grpc++/status.h>
+#include <grpc++/stream.h>
+#include <gtest/gtest.h>
#include "test/core/util/grpc_profiler.h"
#include "test/cpp/util/create_test_channel.h"
#include "test/cpp/qps/client.h"
@@ -61,18 +63,28 @@
namespace grpc {
namespace testing {
-class SynchronousClient GRPC_FINAL : public Client {
+class SynchronousClient : public Client {
public:
SynchronousClient(const ClientConfig& config) : Client(config) {
- size_t num_threads =
- config.outstanding_rpcs_per_channel() * config.client_channels();
- responses_.resize(num_threads);
- StartThreads(num_threads);
+ num_threads_ =
+ config.outstanding_rpcs_per_channel() * config.client_channels();
+ responses_.resize(num_threads_);
}
- ~SynchronousClient() { EndThreads(); }
+ virtual ~SynchronousClient() { EndThreads(); }
- void ThreadFunc(Histogram* histogram, size_t thread_idx) {
+ protected:
+ size_t num_threads_;
+ std::vector<SimpleResponse> responses_;
+};
+
+class SynchronousUnaryClient GRPC_FINAL : public SynchronousClient {
+ public:
+ SynchronousUnaryClient(const ClientConfig& config):
+ SynchronousClient(config) {StartThreads(num_threads_);}
+ ~SynchronousUnaryClient() {}
+
+ void ThreadFunc(Histogram* histogram, size_t thread_idx) GRPC_OVERRIDE {
auto* stub = channels_[thread_idx % channels_.size()].get_stub();
double start = Timer::Now();
grpc::ClientContext context;
@@ -80,13 +92,47 @@
stub->UnaryCall(&context, request_, &responses_[thread_idx]);
histogram->Add((Timer::Now() - start) * 1e9);
}
-
- private:
- std::vector<SimpleResponse> responses_;
};
-std::unique_ptr<Client> CreateSynchronousClient(const ClientConfig& config) {
- return std::unique_ptr<Client>(new SynchronousClient(config));
+class SynchronousStreamingClient GRPC_FINAL : public SynchronousClient {
+ public:
+ SynchronousStreamingClient(const ClientConfig& config):
+ SynchronousClient(config) {
+ for (size_t thread_idx=0;thread_idx<num_threads_;thread_idx++){
+ auto* stub = channels_[thread_idx % channels_.size()].get_stub();
+ stream_ = stub->StreamingCall(&context_);
+ }
+ StartThreads(num_threads_);
+ }
+ ~SynchronousStreamingClient() {
+ if (stream_) {
+ SimpleResponse response;
+ stream_->WritesDone();
+ EXPECT_FALSE(stream_->Read(&response));
+
+ Status s = stream_->Finish();
+ EXPECT_TRUE(s.IsOk());
+ }
+ }
+
+ void ThreadFunc(Histogram* histogram, size_t thread_idx) GRPC_OVERRIDE {
+ double start = Timer::Now();
+ EXPECT_TRUE(stream_->Write(request_));
+ EXPECT_TRUE(stream_->Read(&responses_[thread_idx]));
+ histogram->Add((Timer::Now() - start) * 1e9);
+ }
+ private:
+ grpc::ClientContext context_;
+ std::unique_ptr<grpc::ClientReaderWriter<SimpleRequest,SimpleResponse>> stream_;
+};
+
+std::unique_ptr<Client>
+CreateSynchronousUnaryClient(const ClientConfig& config) {
+ return std::unique_ptr<Client>(new SynchronousUnaryClient(config));
+}
+std::unique_ptr<Client>
+CreateSynchronousStreamingClient(const ClientConfig& config) {
+ return std::unique_ptr<Client>(new SynchronousStreamingClient(config));
}
} // namespace testing