Imported Upstream version 1.41.0
[platform/upstream/grpc.git] / test / cpp / interop / interop_client.cc
1 /*
2  *
3  * Copyright 2015-2016 gRPC authors.
4  *
5  * Licensed under the Apache License, Version 2.0 (the "License");
6  * you may not use this file except in compliance with the License.
7  * You may obtain a copy of the License at
8  *
9  *     http://www.apache.org/licenses/LICENSE-2.0
10  *
11  * Unless required by applicable law or agreed to in writing, software
12  * distributed under the License is distributed on an "AS IS" BASIS,
13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14  * See the License for the specific language governing permissions and
15  * limitations under the License.
16  *
17  */
18
19 #include "test/cpp/interop/interop_client.h"
20
21 #include <cinttypes>
22 #include <fstream>
23 #include <memory>
24 #include <string>
25 #include <type_traits>
26 #include <utility>
27
28 #include "absl/strings/match.h"
29 #include "absl/strings/str_format.h"
30
31 #include <grpc/grpc.h>
32 #include <grpc/support/alloc.h>
33 #include <grpc/support/log.h>
34 #include <grpc/support/string_util.h>
35 #include <grpc/support/time.h>
36 #include <grpcpp/channel.h>
37 #include <grpcpp/client_context.h>
38 #include <grpcpp/security/credentials.h>
39
40 #include "src/proto/grpc/testing/empty.pb.h"
41 #include "src/proto/grpc/testing/messages.pb.h"
42 #include "src/proto/grpc/testing/test.grpc.pb.h"
43 #include "test/core/util/histogram.h"
44 #include "test/cpp/interop/client_helper.h"
45
46 namespace grpc {
47 namespace testing {
48
49 namespace {
50 // The same value is defined by the Java client.
51 const std::vector<int> request_stream_sizes = {27182, 8, 1828, 45904};
52 const std::vector<int> response_stream_sizes = {31415, 9, 2653, 58979};
53 const int kNumResponseMessages = 2000;
54 const int kResponseMessageSize = 1030;
55 const int kReceiveDelayMilliSeconds = 20;
56 const int kLargeRequestSize = 271828;
57 const int kLargeResponseSize = 314159;
58
59 void NoopChecks(const InteropClientContextInspector& /*inspector*/,
60                 const SimpleRequest* /*request*/,
61                 const SimpleResponse* /*response*/) {}
62
63 void UnaryCompressionChecks(const InteropClientContextInspector& inspector,
64                             const SimpleRequest* request,
65                             const SimpleResponse* /*response*/) {
66   const grpc_compression_algorithm received_compression =
67       inspector.GetCallCompressionAlgorithm();
68   if (request->response_compressed().value()) {
69     if (received_compression == GRPC_COMPRESS_NONE) {
70       // Requested some compression, got NONE. This is an error.
71       gpr_log(GPR_ERROR,
72               "Failure: Requested compression but got uncompressed response "
73               "from server.");
74       abort();
75     }
76     GPR_ASSERT(inspector.WasCompressed());
77   } else {
78     // Didn't request compression -> make sure the response is uncompressed
79     GPR_ASSERT(!(inspector.WasCompressed()));
80   }
81 }
82 }  // namespace
83
84 InteropClient::ServiceStub::ServiceStub(
85     ChannelCreationFunc channel_creation_func, bool new_stub_every_call)
86     : channel_creation_func_(std::move(channel_creation_func)),
87       channel_(channel_creation_func_()),
88       new_stub_every_call_(new_stub_every_call) {
89   // If new_stub_every_call is false, then this is our chance to initialize
90   // stub_. (see Get())
91   if (!new_stub_every_call) {
92     stub_ = TestService::NewStub(channel_);
93   }
94 }
95
96 TestService::Stub* InteropClient::ServiceStub::Get() {
97   if (new_stub_every_call_) {
98     stub_ = TestService::NewStub(channel_);
99   }
100
101   return stub_.get();
102 }
103
104 UnimplementedService::Stub*
105 InteropClient::ServiceStub::GetUnimplementedServiceStub() {
106   if (unimplemented_service_stub_ == nullptr) {
107     unimplemented_service_stub_ = UnimplementedService::NewStub(channel_);
108   }
109   return unimplemented_service_stub_.get();
110 }
111
112 void InteropClient::ServiceStub::ResetChannel() {
113   channel_ = channel_creation_func_();
114   if (!new_stub_every_call_) {
115     stub_ = TestService::NewStub(channel_);
116   }
117 }
118
119 InteropClient::InteropClient(ChannelCreationFunc channel_creation_func,
120                              bool new_stub_every_test_case,
121                              bool do_not_abort_on_transient_failures)
122     : serviceStub_(std::move(channel_creation_func), new_stub_every_test_case),
123       do_not_abort_on_transient_failures_(do_not_abort_on_transient_failures) {}
124
125 bool InteropClient::AssertStatusOk(const Status& s,
126                                    const std::string& optional_debug_string) {
127   if (s.ok()) {
128     return true;
129   }
130
131   // Note: At this point, s.error_code is definitely not StatusCode::OK (we
132   // already checked for s.ok() above). So, the following will call abort()
133   // (unless s.error_code() corresponds to a transient failure and
134   // 'do_not_abort_on_transient_failures' is true)
135   return AssertStatusCode(s, StatusCode::OK, optional_debug_string);
136 }
137
138 bool InteropClient::AssertStatusCode(const Status& s, StatusCode expected_code,
139                                      const std::string& optional_debug_string) {
140   if (s.error_code() == expected_code) {
141     return true;
142   }
143
144   gpr_log(GPR_ERROR,
145           "Error status code: %d (expected: %d), message: %s,"
146           " debug string: %s",
147           s.error_code(), expected_code, s.error_message().c_str(),
148           optional_debug_string.c_str());
149
150   // In case of transient transient/retryable failures (like a broken
151   // connection) we may or may not abort (see TransientFailureOrAbort())
152   if (s.error_code() == grpc::StatusCode::UNAVAILABLE) {
153     return TransientFailureOrAbort();
154   }
155
156   abort();
157 }
158
159 bool InteropClient::DoEmpty() {
160   gpr_log(GPR_DEBUG, "Sending an empty rpc...");
161
162   Empty request;
163   Empty response;
164   ClientContext context;
165
166   Status s = serviceStub_.Get()->EmptyCall(&context, request, &response);
167
168   if (!AssertStatusOk(s, context.debug_error_string())) {
169     return false;
170   }
171
172   gpr_log(GPR_DEBUG, "Empty rpc done.");
173   return true;
174 }
175
176 bool InteropClient::PerformLargeUnary(SimpleRequest* request,
177                                       SimpleResponse* response) {
178   return PerformLargeUnary(request, response, NoopChecks);
179 }
180
181 bool InteropClient::PerformLargeUnary(SimpleRequest* request,
182                                       SimpleResponse* response,
183                                       const CheckerFn& custom_checks_fn) {
184   ClientContext context;
185   InteropClientContextInspector inspector(context);
186   request->set_response_size(kLargeResponseSize);
187   std::string payload(kLargeRequestSize, '\0');
188   request->mutable_payload()->set_body(payload.c_str(), kLargeRequestSize);
189   if (request->has_expect_compressed()) {
190     if (request->expect_compressed().value()) {
191       context.set_compression_algorithm(GRPC_COMPRESS_GZIP);
192     } else {
193       context.set_compression_algorithm(GRPC_COMPRESS_NONE);
194     }
195   }
196
197   Status s = serviceStub_.Get()->UnaryCall(&context, *request, response);
198   if (!AssertStatusOk(s, context.debug_error_string())) {
199     return false;
200   }
201
202   custom_checks_fn(inspector, request, response);
203
204   // Payload related checks.
205   GPR_ASSERT(response->payload().body() ==
206              std::string(kLargeResponseSize, '\0'));
207   return true;
208 }
209
210 bool InteropClient::DoComputeEngineCreds(
211     const std::string& default_service_account,
212     const std::string& oauth_scope) {
213   gpr_log(GPR_DEBUG,
214           "Sending a large unary rpc with compute engine credentials ...");
215   SimpleRequest request;
216   SimpleResponse response;
217   request.set_fill_username(true);
218   request.set_fill_oauth_scope(true);
219
220   if (!PerformLargeUnary(&request, &response)) {
221     return false;
222   }
223
224   gpr_log(GPR_DEBUG, "Got username %s", response.username().c_str());
225   gpr_log(GPR_DEBUG, "Got oauth_scope %s", response.oauth_scope().c_str());
226   GPR_ASSERT(!response.username().empty());
227   GPR_ASSERT(response.username().c_str() == default_service_account);
228   GPR_ASSERT(!response.oauth_scope().empty());
229   const char* oauth_scope_str = response.oauth_scope().c_str();
230   GPR_ASSERT(absl::StrContains(oauth_scope, oauth_scope_str));
231   gpr_log(GPR_DEBUG, "Large unary with compute engine creds done.");
232   return true;
233 }
234
235 bool InteropClient::DoOauth2AuthToken(const std::string& username,
236                                       const std::string& oauth_scope) {
237   gpr_log(GPR_DEBUG,
238           "Sending a unary rpc with raw oauth2 access token credentials ...");
239   SimpleRequest request;
240   SimpleResponse response;
241   request.set_fill_username(true);
242   request.set_fill_oauth_scope(true);
243
244   ClientContext context;
245
246   Status s = serviceStub_.Get()->UnaryCall(&context, request, &response);
247
248   if (!AssertStatusOk(s, context.debug_error_string())) {
249     return false;
250   }
251
252   GPR_ASSERT(!response.username().empty());
253   GPR_ASSERT(!response.oauth_scope().empty());
254   GPR_ASSERT(username == response.username());
255   const char* oauth_scope_str = response.oauth_scope().c_str();
256   GPR_ASSERT(absl::StrContains(oauth_scope, oauth_scope_str));
257   gpr_log(GPR_DEBUG, "Unary with oauth2 access token credentials done.");
258   return true;
259 }
260
261 bool InteropClient::DoPerRpcCreds(const std::string& json_key) {
262   gpr_log(GPR_DEBUG, "Sending a unary rpc with per-rpc JWT access token ...");
263   SimpleRequest request;
264   SimpleResponse response;
265   request.set_fill_username(true);
266
267   ClientContext context;
268   std::chrono::seconds token_lifetime = std::chrono::hours(1);
269   std::shared_ptr<CallCredentials> creds =
270       ServiceAccountJWTAccessCredentials(json_key, token_lifetime.count());
271
272   context.set_credentials(creds);
273
274   Status s = serviceStub_.Get()->UnaryCall(&context, request, &response);
275
276   if (!AssertStatusOk(s, context.debug_error_string())) {
277     return false;
278   }
279
280   GPR_ASSERT(!response.username().empty());
281   GPR_ASSERT(json_key.find(response.username()) != std::string::npos);
282   gpr_log(GPR_DEBUG, "Unary with per-rpc JWT access token done.");
283   return true;
284 }
285
286 bool InteropClient::DoJwtTokenCreds(const std::string& username) {
287   gpr_log(GPR_DEBUG,
288           "Sending a large unary rpc with JWT token credentials ...");
289   SimpleRequest request;
290   SimpleResponse response;
291   request.set_fill_username(true);
292
293   if (!PerformLargeUnary(&request, &response)) {
294     return false;
295   }
296
297   GPR_ASSERT(!response.username().empty());
298   GPR_ASSERT(username.find(response.username()) != std::string::npos);
299   gpr_log(GPR_DEBUG, "Large unary with JWT token creds done.");
300   return true;
301 }
302
303 bool InteropClient::DoGoogleDefaultCredentials(
304     const std::string& default_service_account) {
305   gpr_log(GPR_DEBUG,
306           "Sending a large unary rpc with GoogleDefaultCredentials...");
307   SimpleRequest request;
308   SimpleResponse response;
309   request.set_fill_username(true);
310
311   if (!PerformLargeUnary(&request, &response)) {
312     return false;
313   }
314
315   gpr_log(GPR_DEBUG, "Got username %s", response.username().c_str());
316   GPR_ASSERT(!response.username().empty());
317   GPR_ASSERT(response.username().c_str() == default_service_account);
318   gpr_log(GPR_DEBUG, "Large unary rpc with GoogleDefaultCredentials done.");
319   return true;
320 }
321
322 bool InteropClient::DoLargeUnary() {
323   gpr_log(GPR_DEBUG, "Sending a large unary rpc...");
324   SimpleRequest request;
325   SimpleResponse response;
326   if (!PerformLargeUnary(&request, &response)) {
327     return false;
328   }
329   gpr_log(GPR_DEBUG, "Large unary done.");
330   return true;
331 }
332
333 bool InteropClient::DoClientCompressedUnary() {
334   // Probing for compression-checks support.
335   ClientContext probe_context;
336   SimpleRequest probe_req;
337   SimpleResponse probe_res;
338
339   probe_context.set_compression_algorithm(GRPC_COMPRESS_NONE);
340   probe_req.mutable_expect_compressed()->set_value(true);  // lies!
341
342   probe_req.set_response_size(kLargeResponseSize);
343   probe_req.mutable_payload()->set_body(std::string(kLargeRequestSize, '\0'));
344
345   gpr_log(GPR_DEBUG, "Sending probe for compressed unary request.");
346   const Status s =
347       serviceStub_.Get()->UnaryCall(&probe_context, probe_req, &probe_res);
348   if (s.error_code() != grpc::StatusCode::INVALID_ARGUMENT) {
349     // The server isn't able to evaluate incoming compression, making the rest
350     // of this test moot.
351     gpr_log(GPR_DEBUG, "Compressed unary request probe failed");
352     return false;
353   }
354   gpr_log(GPR_DEBUG, "Compressed unary request probe succeeded. Proceeding.");
355
356   const std::vector<bool> compressions = {true, false};
357   for (size_t i = 0; i < compressions.size(); i++) {
358     std::string log_suffix =
359         absl::StrFormat("(compression=%s)", compressions[i] ? "true" : "false");
360
361     gpr_log(GPR_DEBUG, "Sending compressed unary request %s.",
362             log_suffix.c_str());
363     SimpleRequest request;
364     SimpleResponse response;
365     request.mutable_expect_compressed()->set_value(compressions[i]);
366     if (!PerformLargeUnary(&request, &response, UnaryCompressionChecks)) {
367       gpr_log(GPR_ERROR, "Compressed unary request failed %s",
368               log_suffix.c_str());
369       return false;
370     }
371
372     gpr_log(GPR_DEBUG, "Compressed unary request failed %s",
373             log_suffix.c_str());
374   }
375
376   return true;
377 }
378
379 bool InteropClient::DoServerCompressedUnary() {
380   const std::vector<bool> compressions = {true, false};
381   for (size_t i = 0; i < compressions.size(); i++) {
382     std::string log_suffix =
383         absl::StrFormat("(compression=%s)", compressions[i] ? "true" : "false");
384
385     gpr_log(GPR_DEBUG, "Sending unary request for compressed response %s.",
386             log_suffix.c_str());
387     SimpleRequest request;
388     SimpleResponse response;
389     request.mutable_response_compressed()->set_value(compressions[i]);
390
391     if (!PerformLargeUnary(&request, &response, UnaryCompressionChecks)) {
392       gpr_log(GPR_ERROR, "Request for compressed unary failed %s",
393               log_suffix.c_str());
394       return false;
395     }
396
397     gpr_log(GPR_DEBUG, "Request for compressed unary failed %s",
398             log_suffix.c_str());
399   }
400
401   return true;
402 }
403
404 // Either abort() (unless do_not_abort_on_transient_failures_ is true) or return
405 // false
406 bool InteropClient::TransientFailureOrAbort() {
407   if (do_not_abort_on_transient_failures_) {
408     return false;
409   }
410
411   abort();
412 }
413
414 bool InteropClient::DoRequestStreaming() {
415   gpr_log(GPR_DEBUG, "Sending request steaming rpc ...");
416
417   ClientContext context;
418   StreamingInputCallRequest request;
419   StreamingInputCallResponse response;
420
421   std::unique_ptr<ClientWriter<StreamingInputCallRequest>> stream(
422       serviceStub_.Get()->StreamingInputCall(&context, &response));
423
424   int aggregated_payload_size = 0;
425   for (size_t i = 0; i < request_stream_sizes.size(); ++i) {
426     Payload* payload = request.mutable_payload();
427     payload->set_body(std::string(request_stream_sizes[i], '\0'));
428     if (!stream->Write(request)) {
429       gpr_log(GPR_ERROR, "DoRequestStreaming(): stream->Write() failed");
430       return TransientFailureOrAbort();
431     }
432     aggregated_payload_size += request_stream_sizes[i];
433   }
434   GPR_ASSERT(stream->WritesDone());
435
436   Status s = stream->Finish();
437   if (!AssertStatusOk(s, context.debug_error_string())) {
438     return false;
439   }
440
441   GPR_ASSERT(response.aggregated_payload_size() == aggregated_payload_size);
442   return true;
443 }
444
445 bool InteropClient::DoResponseStreaming() {
446   gpr_log(GPR_DEBUG, "Receiving response streaming rpc ...");
447
448   ClientContext context;
449   StreamingOutputCallRequest request;
450   for (unsigned int i = 0; i < response_stream_sizes.size(); ++i) {
451     ResponseParameters* response_parameter = request.add_response_parameters();
452     response_parameter->set_size(response_stream_sizes[i]);
453   }
454   StreamingOutputCallResponse response;
455   std::unique_ptr<ClientReader<StreamingOutputCallResponse>> stream(
456       serviceStub_.Get()->StreamingOutputCall(&context, request));
457
458   unsigned int i = 0;
459   while (stream->Read(&response)) {
460     GPR_ASSERT(response.payload().body() ==
461                std::string(response_stream_sizes[i], '\0'));
462     ++i;
463   }
464
465   if (i < response_stream_sizes.size()) {
466     // stream->Read() failed before reading all the expected messages. This is
467     // most likely due to connection failure.
468     gpr_log(GPR_ERROR,
469             "DoResponseStreaming(): Read fewer streams (%d) than "
470             "response_stream_sizes.size() (%" PRIuPTR ")",
471             i, response_stream_sizes.size());
472     return TransientFailureOrAbort();
473   }
474
475   Status s = stream->Finish();
476   if (!AssertStatusOk(s, context.debug_error_string())) {
477     return false;
478   }
479
480   gpr_log(GPR_DEBUG, "Response streaming done.");
481   return true;
482 }
483
484 bool InteropClient::DoClientCompressedStreaming() {
485   // Probing for compression-checks support.
486   ClientContext probe_context;
487   StreamingInputCallRequest probe_req;
488   StreamingInputCallResponse probe_res;
489
490   probe_context.set_compression_algorithm(GRPC_COMPRESS_NONE);
491   probe_req.mutable_expect_compressed()->set_value(true);  // lies!
492   probe_req.mutable_payload()->set_body(std::string(27182, '\0'));
493
494   gpr_log(GPR_DEBUG, "Sending probe for compressed streaming request.");
495
496   std::unique_ptr<ClientWriter<StreamingInputCallRequest>> probe_stream(
497       serviceStub_.Get()->StreamingInputCall(&probe_context, &probe_res));
498
499   if (!probe_stream->Write(probe_req)) {
500     gpr_log(GPR_ERROR, "%s(): stream->Write() failed", __func__);
501     return TransientFailureOrAbort();
502   }
503   Status s = probe_stream->Finish();
504   if (s.error_code() != grpc::StatusCode::INVALID_ARGUMENT) {
505     // The server isn't able to evaluate incoming compression, making the rest
506     // of this test moot.
507     gpr_log(GPR_DEBUG, "Compressed streaming request probe failed");
508     return false;
509   }
510   gpr_log(GPR_DEBUG,
511           "Compressed streaming request probe succeeded. Proceeding.");
512
513   ClientContext context;
514   StreamingInputCallRequest request;
515   StreamingInputCallResponse response;
516
517   context.set_compression_algorithm(GRPC_COMPRESS_GZIP);
518   std::unique_ptr<ClientWriter<StreamingInputCallRequest>> stream(
519       serviceStub_.Get()->StreamingInputCall(&context, &response));
520
521   request.mutable_payload()->set_body(std::string(27182, '\0'));
522   request.mutable_expect_compressed()->set_value(true);
523   gpr_log(GPR_DEBUG, "Sending streaming request with compression enabled");
524   if (!stream->Write(request)) {
525     gpr_log(GPR_ERROR, "%s(): stream->Write() failed", __func__);
526     return TransientFailureOrAbort();
527   }
528
529   WriteOptions wopts;
530   wopts.set_no_compression();
531   request.mutable_payload()->set_body(std::string(45904, '\0'));
532   request.mutable_expect_compressed()->set_value(false);
533   gpr_log(GPR_DEBUG, "Sending streaming request with compression disabled");
534   if (!stream->Write(request, wopts)) {
535     gpr_log(GPR_ERROR, "%s(): stream->Write() failed", __func__);
536     return TransientFailureOrAbort();
537   }
538   GPR_ASSERT(stream->WritesDone());
539
540   s = stream->Finish();
541   return AssertStatusOk(s, context.debug_error_string());
542 }
543
544 bool InteropClient::DoServerCompressedStreaming() {
545   const std::vector<bool> compressions = {true, false};
546   const std::vector<int> sizes = {31415, 92653};
547
548   ClientContext context;
549   InteropClientContextInspector inspector(context);
550   StreamingOutputCallRequest request;
551
552   GPR_ASSERT(compressions.size() == sizes.size());
553   for (size_t i = 0; i < sizes.size(); i++) {
554     std::string log_suffix =
555         absl::StrFormat("(compression=%s; size=%d)",
556                         compressions[i] ? "true" : "false", sizes[i]);
557
558     gpr_log(GPR_DEBUG, "Sending request streaming rpc %s.", log_suffix.c_str());
559
560     ResponseParameters* const response_parameter =
561         request.add_response_parameters();
562     response_parameter->mutable_compressed()->set_value(compressions[i]);
563     response_parameter->set_size(sizes[i]);
564   }
565   std::unique_ptr<ClientReader<StreamingOutputCallResponse>> stream(
566       serviceStub_.Get()->StreamingOutputCall(&context, request));
567
568   size_t k = 0;
569   StreamingOutputCallResponse response;
570   while (stream->Read(&response)) {
571     // Payload size checks.
572     GPR_ASSERT(response.payload().body() ==
573                std::string(request.response_parameters(k).size(), '\0'));
574
575     // Compression checks.
576     GPR_ASSERT(request.response_parameters(k).has_compressed());
577     if (request.response_parameters(k).compressed().value()) {
578       GPR_ASSERT(inspector.GetCallCompressionAlgorithm() > GRPC_COMPRESS_NONE);
579       GPR_ASSERT(inspector.WasCompressed());
580     } else {
581       // requested *no* compression.
582       GPR_ASSERT(!(inspector.WasCompressed()));
583     }
584     ++k;
585   }
586
587   if (k < sizes.size()) {
588     // stream->Read() failed before reading all the expected messages. This
589     // is most likely due to a connection failure.
590     gpr_log(GPR_ERROR,
591             "%s(): Responses read (k=%" PRIuPTR
592             ") is less than the expected number of  messages (%" PRIuPTR ").",
593             __func__, k, sizes.size());
594     return TransientFailureOrAbort();
595   }
596
597   Status s = stream->Finish();
598   return AssertStatusOk(s, context.debug_error_string());
599 }
600
601 bool InteropClient::DoResponseStreamingWithSlowConsumer() {
602   gpr_log(GPR_DEBUG, "Receiving response streaming rpc with slow consumer ...");
603
604   ClientContext context;
605   StreamingOutputCallRequest request;
606
607   for (int i = 0; i < kNumResponseMessages; ++i) {
608     ResponseParameters* response_parameter = request.add_response_parameters();
609     response_parameter->set_size(kResponseMessageSize);
610   }
611   StreamingOutputCallResponse response;
612   std::unique_ptr<ClientReader<StreamingOutputCallResponse>> stream(
613       serviceStub_.Get()->StreamingOutputCall(&context, request));
614
615   int i = 0;
616   while (stream->Read(&response)) {
617     GPR_ASSERT(response.payload().body() ==
618                std::string(kResponseMessageSize, '\0'));
619     gpr_log(GPR_DEBUG, "received message %d", i);
620     gpr_sleep_until(gpr_time_add(
621         gpr_now(GPR_CLOCK_REALTIME),
622         gpr_time_from_millis(kReceiveDelayMilliSeconds, GPR_TIMESPAN)));
623     ++i;
624   }
625
626   if (i < kNumResponseMessages) {
627     gpr_log(GPR_ERROR,
628             "DoResponseStreamingWithSlowConsumer(): Responses read (i=%d) is "
629             "less than the expected messages (i.e kNumResponseMessages = %d)",
630             i, kNumResponseMessages);
631
632     return TransientFailureOrAbort();
633   }
634
635   Status s = stream->Finish();
636   if (!AssertStatusOk(s, context.debug_error_string())) {
637     return false;
638   }
639
640   gpr_log(GPR_DEBUG, "Response streaming done.");
641   return true;
642 }
643
644 bool InteropClient::DoHalfDuplex() {
645   gpr_log(GPR_DEBUG, "Sending half-duplex streaming rpc ...");
646
647   ClientContext context;
648   std::unique_ptr<ClientReaderWriter<StreamingOutputCallRequest,
649                                      StreamingOutputCallResponse>>
650       stream(serviceStub_.Get()->HalfDuplexCall(&context));
651
652   StreamingOutputCallRequest request;
653   ResponseParameters* response_parameter = request.add_response_parameters();
654   for (unsigned int i = 0; i < response_stream_sizes.size(); ++i) {
655     response_parameter->set_size(response_stream_sizes[i]);
656
657     if (!stream->Write(request)) {
658       gpr_log(GPR_ERROR, "DoHalfDuplex(): stream->Write() failed. i=%d", i);
659       return TransientFailureOrAbort();
660     }
661   }
662   stream->WritesDone();
663
664   unsigned int i = 0;
665   StreamingOutputCallResponse response;
666   while (stream->Read(&response)) {
667     GPR_ASSERT(response.payload().body() ==
668                std::string(response_stream_sizes[i], '\0'));
669     ++i;
670   }
671
672   if (i < response_stream_sizes.size()) {
673     // stream->Read() failed before reading all the expected messages. This is
674     // most likely due to a connection failure
675     gpr_log(GPR_ERROR,
676             "DoHalfDuplex(): Responses read (i=%d) are less than the expected "
677             "number of messages response_stream_sizes.size() (%" PRIuPTR ")",
678             i, response_stream_sizes.size());
679     return TransientFailureOrAbort();
680   }
681
682   Status s = stream->Finish();
683   if (!AssertStatusOk(s, context.debug_error_string())) {
684     return false;
685   }
686
687   gpr_log(GPR_DEBUG, "Half-duplex streaming rpc done.");
688   return true;
689 }
690
691 bool InteropClient::DoPingPong() {
692   gpr_log(GPR_DEBUG, "Sending Ping Pong streaming rpc ...");
693
694   ClientContext context;
695   std::unique_ptr<ClientReaderWriter<StreamingOutputCallRequest,
696                                      StreamingOutputCallResponse>>
697       stream(serviceStub_.Get()->FullDuplexCall(&context));
698
699   StreamingOutputCallRequest request;
700   ResponseParameters* response_parameter = request.add_response_parameters();
701   Payload* payload = request.mutable_payload();
702   StreamingOutputCallResponse response;
703
704   for (unsigned int i = 0; i < request_stream_sizes.size(); ++i) {
705     response_parameter->set_size(response_stream_sizes[i]);
706     payload->set_body(std::string(request_stream_sizes[i], '\0'));
707
708     if (!stream->Write(request)) {
709       gpr_log(GPR_ERROR, "DoPingPong(): stream->Write() failed. i: %d", i);
710       return TransientFailureOrAbort();
711     }
712
713     if (!stream->Read(&response)) {
714       gpr_log(GPR_ERROR, "DoPingPong(): stream->Read() failed. i:%d", i);
715       return TransientFailureOrAbort();
716     }
717
718     GPR_ASSERT(response.payload().body() ==
719                std::string(response_stream_sizes[i], '\0'));
720   }
721
722   stream->WritesDone();
723
724   GPR_ASSERT(!stream->Read(&response));
725
726   Status s = stream->Finish();
727   if (!AssertStatusOk(s, context.debug_error_string())) {
728     return false;
729   }
730
731   gpr_log(GPR_DEBUG, "Ping pong streaming done.");
732   return true;
733 }
734
735 bool InteropClient::DoCancelAfterBegin() {
736   gpr_log(GPR_DEBUG, "Sending request streaming rpc ...");
737
738   ClientContext context;
739   StreamingInputCallRequest request;
740   StreamingInputCallResponse response;
741
742   std::unique_ptr<ClientWriter<StreamingInputCallRequest>> stream(
743       serviceStub_.Get()->StreamingInputCall(&context, &response));
744
745   gpr_log(GPR_DEBUG, "Trying to cancel...");
746   context.TryCancel();
747   Status s = stream->Finish();
748
749   if (!AssertStatusCode(s, StatusCode::CANCELLED,
750                         context.debug_error_string())) {
751     return false;
752   }
753
754   gpr_log(GPR_DEBUG, "Canceling streaming done.");
755   return true;
756 }
757
758 bool InteropClient::DoCancelAfterFirstResponse() {
759   gpr_log(GPR_DEBUG, "Sending Ping Pong streaming rpc ...");
760
761   ClientContext context;
762   std::unique_ptr<ClientReaderWriter<StreamingOutputCallRequest,
763                                      StreamingOutputCallResponse>>
764       stream(serviceStub_.Get()->FullDuplexCall(&context));
765
766   StreamingOutputCallRequest request;
767   ResponseParameters* response_parameter = request.add_response_parameters();
768   response_parameter->set_size(31415);
769   request.mutable_payload()->set_body(std::string(27182, '\0'));
770   StreamingOutputCallResponse response;
771
772   if (!stream->Write(request)) {
773     gpr_log(GPR_ERROR, "DoCancelAfterFirstResponse(): stream->Write() failed");
774     return TransientFailureOrAbort();
775   }
776
777   if (!stream->Read(&response)) {
778     gpr_log(GPR_ERROR, "DoCancelAfterFirstResponse(): stream->Read failed");
779     return TransientFailureOrAbort();
780   }
781   GPR_ASSERT(response.payload().body() == std::string(31415, '\0'));
782
783   gpr_log(GPR_DEBUG, "Trying to cancel...");
784   context.TryCancel();
785
786   Status s = stream->Finish();
787   gpr_log(GPR_DEBUG, "Canceling pingpong streaming done.");
788   return true;
789 }
790
791 bool InteropClient::DoTimeoutOnSleepingServer() {
792   gpr_log(GPR_DEBUG,
793           "Sending Ping Pong streaming rpc with a short deadline...");
794
795   ClientContext context;
796   std::chrono::system_clock::time_point deadline =
797       std::chrono::system_clock::now() + std::chrono::milliseconds(1);
798   context.set_deadline(deadline);
799   std::unique_ptr<ClientReaderWriter<StreamingOutputCallRequest,
800                                      StreamingOutputCallResponse>>
801       stream(serviceStub_.Get()->FullDuplexCall(&context));
802
803   StreamingOutputCallRequest request;
804   request.mutable_payload()->set_body(std::string(27182, '\0'));
805   stream->Write(request);
806
807   Status s = stream->Finish();
808   if (!AssertStatusCode(s, StatusCode::DEADLINE_EXCEEDED,
809                         context.debug_error_string())) {
810     return false;
811   }
812
813   gpr_log(GPR_DEBUG, "Pingpong streaming timeout done.");
814   return true;
815 }
816
817 bool InteropClient::DoEmptyStream() {
818   gpr_log(GPR_DEBUG, "Starting empty_stream.");
819
820   ClientContext context;
821   std::unique_ptr<ClientReaderWriter<StreamingOutputCallRequest,
822                                      StreamingOutputCallResponse>>
823       stream(serviceStub_.Get()->FullDuplexCall(&context));
824   stream->WritesDone();
825   StreamingOutputCallResponse response;
826   GPR_ASSERT(stream->Read(&response) == false);
827
828   Status s = stream->Finish();
829   if (!AssertStatusOk(s, context.debug_error_string())) {
830     return false;
831   }
832
833   gpr_log(GPR_DEBUG, "empty_stream done.");
834   return true;
835 }
836
837 bool InteropClient::DoStatusWithMessage() {
838   gpr_log(GPR_DEBUG,
839           "Sending RPC with a request for status code 2 and message");
840
841   const grpc::StatusCode test_code = grpc::StatusCode::UNKNOWN;
842   const std::string test_msg = "This is a test message";
843
844   // Test UnaryCall.
845   ClientContext context;
846   SimpleRequest request;
847   SimpleResponse response;
848   EchoStatus* requested_status = request.mutable_response_status();
849   requested_status->set_code(test_code);
850   requested_status->set_message(test_msg);
851   Status s = serviceStub_.Get()->UnaryCall(&context, request, &response);
852   if (!AssertStatusCode(s, grpc::StatusCode::UNKNOWN,
853                         context.debug_error_string())) {
854     return false;
855   }
856   GPR_ASSERT(s.error_message() == test_msg);
857
858   // Test FullDuplexCall.
859   ClientContext stream_context;
860   std::shared_ptr<ClientReaderWriter<StreamingOutputCallRequest,
861                                      StreamingOutputCallResponse>>
862       stream(serviceStub_.Get()->FullDuplexCall(&stream_context));
863   StreamingOutputCallRequest streaming_request;
864   requested_status = streaming_request.mutable_response_status();
865   requested_status->set_code(test_code);
866   requested_status->set_message(test_msg);
867   stream->Write(streaming_request);
868   stream->WritesDone();
869   StreamingOutputCallResponse streaming_response;
870   while (stream->Read(&streaming_response)) {
871   }
872   s = stream->Finish();
873   if (!AssertStatusCode(s, grpc::StatusCode::UNKNOWN,
874                         context.debug_error_string())) {
875     return false;
876   }
877   GPR_ASSERT(s.error_message() == test_msg);
878
879   gpr_log(GPR_DEBUG, "Done testing Status and Message");
880   return true;
881 }
882
883 bool InteropClient::DoSpecialStatusMessage() {
884   gpr_log(
885       GPR_DEBUG,
886       "Sending RPC with a request for status code 2 and message - \\t\\ntest "
887       "with whitespace\\r\\nand Unicode BMP â˜º and non-BMP ðŸ˜ˆ\\t\\n");
888   const grpc::StatusCode test_code = grpc::StatusCode::UNKNOWN;
889   const std::string test_msg =
890       "\t\ntest with whitespace\r\nand Unicode BMP â˜º and non-BMP ðŸ˜ˆ\t\n";
891   ClientContext context;
892   SimpleRequest request;
893   SimpleResponse response;
894   EchoStatus* requested_status = request.mutable_response_status();
895   requested_status->set_code(test_code);
896   requested_status->set_message(test_msg);
897   Status s = serviceStub_.Get()->UnaryCall(&context, request, &response);
898   if (!AssertStatusCode(s, grpc::StatusCode::UNKNOWN,
899                         context.debug_error_string())) {
900     return false;
901   }
902   GPR_ASSERT(s.error_message() == test_msg);
903   gpr_log(GPR_DEBUG, "Done testing Special Status Message");
904   return true;
905 }
906
907 bool InteropClient::DoCacheableUnary() {
908   gpr_log(GPR_DEBUG, "Sending RPC with cacheable response");
909
910   // Create request with current timestamp
911   gpr_timespec ts = gpr_now(GPR_CLOCK_PRECISE);
912   std::string timestamp =
913       std::to_string(static_cast<long long unsigned>(ts.tv_nsec));
914   SimpleRequest request;
915   request.mutable_payload()->set_body(timestamp.c_str(), timestamp.size());
916
917   // Request 1
918   ClientContext context1;
919   SimpleResponse response1;
920   context1.set_cacheable(true);
921   // Add fake user IP since some proxy's (GFE) won't cache requests from
922   // localhost.
923   context1.AddMetadata("x-user-ip", "1.2.3.4");
924   Status s1 =
925       serviceStub_.Get()->CacheableUnaryCall(&context1, request, &response1);
926   if (!AssertStatusOk(s1, context1.debug_error_string())) {
927     return false;
928   }
929   gpr_log(GPR_DEBUG, "response 1 payload: %s",
930           response1.payload().body().c_str());
931
932   // Request 2
933   ClientContext context2;
934   SimpleResponse response2;
935   context2.set_cacheable(true);
936   context2.AddMetadata("x-user-ip", "1.2.3.4");
937   Status s2 =
938       serviceStub_.Get()->CacheableUnaryCall(&context2, request, &response2);
939   if (!AssertStatusOk(s2, context2.debug_error_string())) {
940     return false;
941   }
942   gpr_log(GPR_DEBUG, "response 2 payload: %s",
943           response2.payload().body().c_str());
944
945   // Check that the body is same for both requests. It will be the same if the
946   // second response is a cached copy of the first response
947   GPR_ASSERT(response2.payload().body() == response1.payload().body());
948
949   // Request 3
950   // Modify the request body so it will not get a cache hit
951   ts = gpr_now(GPR_CLOCK_PRECISE);
952   timestamp = std::to_string(static_cast<long long unsigned>(ts.tv_nsec));
953   SimpleRequest request1;
954   request1.mutable_payload()->set_body(timestamp.c_str(), timestamp.size());
955   ClientContext context3;
956   SimpleResponse response3;
957   context3.set_cacheable(true);
958   context3.AddMetadata("x-user-ip", "1.2.3.4");
959   Status s3 =
960       serviceStub_.Get()->CacheableUnaryCall(&context3, request1, &response3);
961   if (!AssertStatusOk(s3, context3.debug_error_string())) {
962     return false;
963   }
964   gpr_log(GPR_DEBUG, "response 3 payload: %s",
965           response3.payload().body().c_str());
966
967   // Check that the response is different from the previous response.
968   GPR_ASSERT(response3.payload().body() != response1.payload().body());
969   return true;
970 }
971
972 bool InteropClient::DoPickFirstUnary() {
973   const int rpcCount = 100;
974   SimpleRequest request;
975   SimpleResponse response;
976   std::string server_id;
977   request.set_fill_server_id(true);
978   for (int i = 0; i < rpcCount; i++) {
979     ClientContext context;
980     Status s = serviceStub_.Get()->UnaryCall(&context, request, &response);
981     if (!AssertStatusOk(s, context.debug_error_string())) {
982       return false;
983     }
984     if (i == 0) {
985       server_id = response.server_id();
986       continue;
987     }
988     if (response.server_id() != server_id) {
989       gpr_log(GPR_ERROR, "#%d rpc hits server_id %s, expect server_id %s", i,
990               response.server_id().c_str(), server_id.c_str());
991       return false;
992     }
993   }
994   gpr_log(GPR_DEBUG, "pick first unary successfully finished");
995   return true;
996 }
997
998 bool InteropClient::DoCustomMetadata() {
999   const std::string kEchoInitialMetadataKey("x-grpc-test-echo-initial");
1000   const std::string kInitialMetadataValue("test_initial_metadata_value");
1001   const std::string kEchoTrailingBinMetadataKey(
1002       "x-grpc-test-echo-trailing-bin");
1003   const std::string kTrailingBinValue("\x0a\x0b\x0a\x0b\x0a\x0b");
1004
1005   {
1006     gpr_log(GPR_DEBUG, "Sending RPC with custom metadata");
1007     ClientContext context;
1008     context.AddMetadata(kEchoInitialMetadataKey, kInitialMetadataValue);
1009     context.AddMetadata(kEchoTrailingBinMetadataKey, kTrailingBinValue);
1010     SimpleRequest request;
1011     SimpleResponse response;
1012     request.set_response_size(kLargeResponseSize);
1013     std::string payload(kLargeRequestSize, '\0');
1014     request.mutable_payload()->set_body(payload.c_str(), kLargeRequestSize);
1015
1016     Status s = serviceStub_.Get()->UnaryCall(&context, request, &response);
1017     if (!AssertStatusOk(s, context.debug_error_string())) {
1018       return false;
1019     }
1020
1021     const auto& server_initial_metadata = context.GetServerInitialMetadata();
1022     auto iter = server_initial_metadata.find(kEchoInitialMetadataKey);
1023     GPR_ASSERT(iter != server_initial_metadata.end());
1024     GPR_ASSERT(iter->second == kInitialMetadataValue);
1025     const auto& server_trailing_metadata = context.GetServerTrailingMetadata();
1026     iter = server_trailing_metadata.find(kEchoTrailingBinMetadataKey);
1027     GPR_ASSERT(iter != server_trailing_metadata.end());
1028     GPR_ASSERT(std::string(iter->second.begin(), iter->second.end()) ==
1029                kTrailingBinValue);
1030
1031     gpr_log(GPR_DEBUG, "Done testing RPC with custom metadata");
1032   }
1033
1034   {
1035     gpr_log(GPR_DEBUG, "Sending stream with custom metadata");
1036     ClientContext context;
1037     context.AddMetadata(kEchoInitialMetadataKey, kInitialMetadataValue);
1038     context.AddMetadata(kEchoTrailingBinMetadataKey, kTrailingBinValue);
1039     std::unique_ptr<ClientReaderWriter<StreamingOutputCallRequest,
1040                                        StreamingOutputCallResponse>>
1041         stream(serviceStub_.Get()->FullDuplexCall(&context));
1042
1043     StreamingOutputCallRequest request;
1044     ResponseParameters* response_parameter = request.add_response_parameters();
1045     response_parameter->set_size(kLargeResponseSize);
1046     std::string payload(kLargeRequestSize, '\0');
1047     request.mutable_payload()->set_body(payload.c_str(), kLargeRequestSize);
1048     StreamingOutputCallResponse response;
1049
1050     if (!stream->Write(request)) {
1051       gpr_log(GPR_ERROR, "DoCustomMetadata(): stream->Write() failed");
1052       return TransientFailureOrAbort();
1053     }
1054
1055     stream->WritesDone();
1056
1057     if (!stream->Read(&response)) {
1058       gpr_log(GPR_ERROR, "DoCustomMetadata(): stream->Read() failed");
1059       return TransientFailureOrAbort();
1060     }
1061
1062     GPR_ASSERT(response.payload().body() ==
1063                std::string(kLargeResponseSize, '\0'));
1064
1065     GPR_ASSERT(!stream->Read(&response));
1066
1067     Status s = stream->Finish();
1068     if (!AssertStatusOk(s, context.debug_error_string())) {
1069       return false;
1070     }
1071
1072     const auto& server_initial_metadata = context.GetServerInitialMetadata();
1073     auto iter = server_initial_metadata.find(kEchoInitialMetadataKey);
1074     GPR_ASSERT(iter != server_initial_metadata.end());
1075     GPR_ASSERT(iter->second == kInitialMetadataValue);
1076     const auto& server_trailing_metadata = context.GetServerTrailingMetadata();
1077     iter = server_trailing_metadata.find(kEchoTrailingBinMetadataKey);
1078     GPR_ASSERT(iter != server_trailing_metadata.end());
1079     GPR_ASSERT(std::string(iter->second.begin(), iter->second.end()) ==
1080                kTrailingBinValue);
1081
1082     gpr_log(GPR_DEBUG, "Done testing stream with custom metadata");
1083   }
1084
1085   return true;
1086 }
1087
1088 std::tuple<bool, int32_t, std::string>
1089 InteropClient::PerformOneSoakTestIteration(
1090     const bool reset_channel,
1091     const int32_t max_acceptable_per_iteration_latency_ms) {
1092   gpr_timespec start = gpr_now(GPR_CLOCK_MONOTONIC);
1093   SimpleRequest request;
1094   SimpleResponse response;
1095   // Don't set the deadline on the RPC, and instead just
1096   // record how long the RPC took and compare. This makes
1097   // debugging easier when looking at failure results.
1098   ClientContext context;
1099   InteropClientContextInspector inspector(context);
1100   request.set_response_size(kLargeResponseSize);
1101   std::string payload(kLargeRequestSize, '\0');
1102   request.mutable_payload()->set_body(payload.c_str(), kLargeRequestSize);
1103   if (reset_channel) {
1104     serviceStub_.ResetChannel();
1105   }
1106   Status s = serviceStub_.Get()->UnaryCall(&context, request, &response);
1107   gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
1108   int32_t elapsed_ms = gpr_time_to_millis(gpr_time_sub(now, start));
1109   if (!s.ok()) {
1110     return std::make_tuple(false, elapsed_ms, context.debug_error_string());
1111   } else if (elapsed_ms > max_acceptable_per_iteration_latency_ms) {
1112     std::string debug_string = absl::StrFormat(
1113         "%d ms exceeds max acceptable latency: %d ms, peer: %s", elapsed_ms,
1114         max_acceptable_per_iteration_latency_ms, context.peer());
1115     return std::make_tuple(false, elapsed_ms, std::move(debug_string));
1116   } else {
1117     return std::make_tuple(true, elapsed_ms, "");
1118   }
1119 }
1120
1121 void InteropClient::PerformSoakTest(
1122     const bool reset_channel_per_iteration, const int32_t soak_iterations,
1123     const int32_t max_failures,
1124     const int32_t max_acceptable_per_iteration_latency_ms,
1125     const int32_t overall_timeout_seconds) {
1126   std::vector<std::tuple<bool, int32_t, std::string>> results;
1127   grpc_histogram* latencies_ms_histogram = grpc_histogram_create(
1128       1 /* resolution */,
1129       500 * 1e3 /* largest bucket; 500 seconds is unlikely */);
1130   gpr_timespec overall_deadline = gpr_time_add(
1131       gpr_now(GPR_CLOCK_MONOTONIC),
1132       gpr_time_from_seconds(overall_timeout_seconds, GPR_TIMESPAN));
1133   int32_t iterations_ran = 0;
1134   for (int i = 0;
1135        i < soak_iterations &&
1136        gpr_time_cmp(gpr_now(GPR_CLOCK_MONOTONIC), overall_deadline) < 0;
1137        ++i) {
1138     auto result = PerformOneSoakTestIteration(
1139         reset_channel_per_iteration, max_acceptable_per_iteration_latency_ms);
1140     results.push_back(result);
1141     grpc_histogram_add(latencies_ms_histogram, std::get<1>(result));
1142     iterations_ran++;
1143   }
1144   int total_failures = 0;
1145   for (size_t i = 0; i < results.size(); i++) {
1146     bool success = std::get<0>(results[i]);
1147     int32_t elapsed_ms = std::get<1>(results[i]);
1148     std::string debug_string = std::get<2>(results[i]);
1149     if (!success) {
1150       gpr_log(GPR_DEBUG, "soak iteration: %ld elapsed_ms: %d failed: %s", i,
1151               elapsed_ms, debug_string.c_str());
1152       total_failures++;
1153     } else {
1154       gpr_log(GPR_DEBUG, "soak iteration: %ld elapsed_ms: %d succeeded", i,
1155               elapsed_ms);
1156     }
1157   }
1158   double latency_ms_median =
1159       grpc_histogram_percentile(latencies_ms_histogram, 50);
1160   double latency_ms_90th =
1161       grpc_histogram_percentile(latencies_ms_histogram, 90);
1162   double latency_ms_worst = grpc_histogram_maximum(latencies_ms_histogram);
1163   grpc_histogram_destroy(latencies_ms_histogram);
1164   if (iterations_ran < soak_iterations) {
1165     gpr_log(
1166         GPR_ERROR,
1167         "soak test consumed all %d seconds of time and quit early, only "
1168         "having ran %d out of desired %d iterations. "
1169         "total_failures: %d. "
1170         "max_failures_threshold: %d. "
1171         "median_soak_iteration_latency: %lf ms. "
1172         "90th_soak_iteration_latency: %lf ms. "
1173         "worst_soak_iteration_latency: %lf ms. "
1174         "Some or all of the iterations that did run were unexpectedly slow. "
1175         "See breakdown above for which iterations succeeded, failed, and "
1176         "why for more info.",
1177         overall_timeout_seconds, iterations_ran, soak_iterations,
1178         total_failures, max_failures, latency_ms_median, latency_ms_90th,
1179         latency_ms_worst);
1180     GPR_ASSERT(0);
1181   } else if (total_failures > max_failures) {
1182     gpr_log(GPR_ERROR,
1183             "soak test ran: %d iterations. total_failures: %d exceeds "
1184             "max_failures_threshold: %d. "
1185             "median_soak_iteration_latency: %lf ms. "
1186             "90th_soak_iteration_latency: %lf ms. "
1187             "worst_soak_iteration_latency: %lf ms. "
1188             "See breakdown above for which iterations succeeded, failed, and "
1189             "why for more info.",
1190             soak_iterations, total_failures, max_failures, latency_ms_median,
1191             latency_ms_90th, latency_ms_worst);
1192     GPR_ASSERT(0);
1193   } else {
1194     gpr_log(GPR_INFO,
1195             "soak test ran: %d iterations. total_failures: %d is within "
1196             "max_failures_threshold: %d. "
1197             "median_soak_iteration_latency: %lf ms. "
1198             "90th_soak_iteration_latency: %lf ms. "
1199             "worst_soak_iteration_latency: %lf ms. "
1200             "See breakdown above for which iterations succeeded, failed, and "
1201             "why for more info.",
1202             soak_iterations, total_failures, max_failures, latency_ms_median,
1203             latency_ms_90th, latency_ms_worst);
1204   }
1205 }
1206
1207 bool InteropClient::DoRpcSoakTest(
1208     int32_t soak_iterations, int32_t max_failures,
1209     int64_t max_acceptable_per_iteration_latency_ms,
1210     int32_t overall_timeout_seconds) {
1211   gpr_log(GPR_DEBUG, "Sending %d RPCs...", soak_iterations);
1212   GPR_ASSERT(soak_iterations > 0);
1213   PerformSoakTest(false /* reset channel per iteration */, soak_iterations,
1214                   max_failures, max_acceptable_per_iteration_latency_ms,
1215                   overall_timeout_seconds);
1216   gpr_log(GPR_DEBUG, "rpc_soak test done.");
1217   return true;
1218 }
1219
1220 bool InteropClient::DoChannelSoakTest(
1221     int32_t soak_iterations, int32_t max_failures,
1222     int64_t max_acceptable_per_iteration_latency_ms,
1223     int32_t overall_timeout_seconds) {
1224   gpr_log(GPR_DEBUG, "Sending %d RPCs, tearing down the channel each time...",
1225           soak_iterations);
1226   GPR_ASSERT(soak_iterations > 0);
1227   PerformSoakTest(true /* reset channel per iteration */, soak_iterations,
1228                   max_failures, max_acceptable_per_iteration_latency_ms,
1229                   overall_timeout_seconds);
1230   gpr_log(GPR_DEBUG, "channel_soak test done.");
1231   return true;
1232 }
1233
1234 bool InteropClient::DoLongLivedChannelTest(int32_t soak_iterations,
1235                                            int32_t iteration_interval) {
1236   gpr_log(GPR_DEBUG, "Sending %d RPCs...", soak_iterations);
1237   GPR_ASSERT(soak_iterations > 0);
1238   GPR_ASSERT(iteration_interval > 0);
1239   SimpleRequest request;
1240   SimpleResponse response;
1241   int num_failures = 0;
1242   for (int i = 0; i < soak_iterations; ++i) {
1243     gpr_log(GPR_DEBUG, "Sending RPC number %d...", i);
1244     if (!PerformLargeUnary(&request, &response)) {
1245       gpr_log(GPR_ERROR, "Iteration %d failed.", i);
1246       num_failures++;
1247     }
1248     gpr_sleep_until(
1249         gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
1250                      gpr_time_from_seconds(iteration_interval, GPR_TIMESPAN)));
1251   }
1252   if (num_failures == 0) {
1253     gpr_log(GPR_DEBUG, "long_lived_channel test done.");
1254     return true;
1255   } else {
1256     gpr_log(GPR_DEBUG, "long_lived_channel test failed with %d rpc failures.",
1257             num_failures);
1258     return false;
1259   }
1260 }
1261
1262 bool InteropClient::DoUnimplementedService() {
1263   gpr_log(GPR_DEBUG, "Sending a request for an unimplemented service...");
1264
1265   Empty request;
1266   Empty response;
1267   ClientContext context;
1268
1269   UnimplementedService::Stub* stub = serviceStub_.GetUnimplementedServiceStub();
1270
1271   Status s = stub->UnimplementedCall(&context, request, &response);
1272
1273   if (!AssertStatusCode(s, StatusCode::UNIMPLEMENTED,
1274                         context.debug_error_string())) {
1275     return false;
1276   }
1277
1278   gpr_log(GPR_DEBUG, "unimplemented service done.");
1279   return true;
1280 }
1281
1282 bool InteropClient::DoUnimplementedMethod() {
1283   gpr_log(GPR_DEBUG, "Sending a request for an unimplemented rpc...");
1284
1285   Empty request;
1286   Empty response;
1287   ClientContext context;
1288
1289   Status s =
1290       serviceStub_.Get()->UnimplementedCall(&context, request, &response);
1291
1292   if (!AssertStatusCode(s, StatusCode::UNIMPLEMENTED,
1293                         context.debug_error_string())) {
1294     return false;
1295   }
1296
1297   gpr_log(GPR_DEBUG, "unimplemented rpc done.");
1298   return true;
1299 }
1300
1301 }  // namespace testing
1302 }  // namespace grpc