test/cpp/end2end/client_callback_end2end_test.cc

   1 /*
   2  *
   3  * Copyright 2018 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 <algorithm>
  20 #include <functional>
  21 #include <mutex>
  22 #include <sstream>
  23 #include <thread>
  24
  25 #include <grpcpp/channel.h>
  26 #include <grpcpp/client_context.h>
  27 #include <grpcpp/create_channel.h>
  28 #include <grpcpp/generic/generic_stub.h>
  29 #include <grpcpp/impl/codegen/proto_utils.h>
  30 #include <grpcpp/server.h>
  31 #include <grpcpp/server_builder.h>
  32 #include <grpcpp/server_context.h>
  33 #include <grpcpp/support/client_callback.h>
  34
  35 #include "src/core/lib/gpr/env.h"
  36 #include "src/core/lib/iomgr/iomgr.h"
  37 #include "src/proto/grpc/testing/echo.grpc.pb.h"
  38 #include "test/core/util/port.h"
  39 #include "test/core/util/test_config.h"
  40 #include "test/cpp/end2end/interceptors_util.h"
  41 #include "test/cpp/end2end/test_service_impl.h"
  42 #include "test/cpp/util/byte_buffer_proto_helper.h"
  43 #include "test/cpp/util/string_ref_helper.h"
  44 #include "test/cpp/util/test_credentials_provider.h"
  45
  46 #include <gtest/gtest.h>
  47
  48 // MAYBE_SKIP_TEST is a macro to determine if this particular test configuration
  49 // should be skipped based on a decision made at SetUp time. In particular, any
  50 // callback tests can only be run if the iomgr can run in the background or if
  51 // the transport is in-process.
  52 #define MAYBE_SKIP_TEST \
  53   do {                  \
  54     if (do_not_test_) { \
  55       return;           \
  56     }                   \
  57   } while (0)
  58
  59 namespace grpc {
  60 namespace testing {
  61 namespace {
  62
  63 enum class Protocol { INPROC, TCP };
  64
  65 class TestScenario {
  66  public:
  67   TestScenario(bool serve_callback, Protocol protocol, bool intercept,
  68                const grpc::string& creds_type)
  69       : callback_server(serve_callback),
  70         protocol(protocol),
  71         use_interceptors(intercept),
  72         credentials_type(creds_type) {}
  73   void Log() const;
  74   bool callback_server;
  75   Protocol protocol;
  76   bool use_interceptors;
  77   const grpc::string credentials_type;
  78 };
  79
  80 static std::ostream& operator<<(std::ostream& out,
  81                                 const TestScenario& scenario) {
  82   return out << "TestScenario{callback_server="
  83              << (scenario.callback_server ? "true" : "false") << ",protocol="
  84              << (scenario.protocol == Protocol::INPROC ? "INPROC" : "TCP")
  85              << ",intercept=" << (scenario.use_interceptors ? "true" : "false")
  86              << ",creds=" << scenario.credentials_type << "}";
  87 }
  88
  89 void TestScenario::Log() const {
  90   std::ostringstream out;
  91   out << *this;
  92   gpr_log(GPR_DEBUG, "%s", out.str().c_str());
  93 }
  94
  95 class ClientCallbackEnd2endTest
  96     : public ::testing::TestWithParam<TestScenario> {
  97  protected:
  98   ClientCallbackEnd2endTest() { GetParam().Log(); }
  99
 100   void SetUp() override {
 101     ServerBuilder builder;
 102
 103     auto server_creds = GetCredentialsProvider()->GetServerCredentials(
 104         GetParam().credentials_type);
 105     // TODO(vjpai): Support testing of AuthMetadataProcessor
 106
 107     if (GetParam().protocol == Protocol::TCP) {
 108       picked_port_ = grpc_pick_unused_port_or_die();
 109       server_address_ << "localhost:" << picked_port_;
 110       builder.AddListeningPort(server_address_.str(), server_creds);
 111     }
 112     if (!GetParam().callback_server) {
 113       builder.RegisterService(&service_);
 114     } else {
 115       builder.RegisterService(&callback_service_);
 116     }
 117
 118     if (GetParam().use_interceptors) {
 119       std::vector<
 120           std::unique_ptr<experimental::ServerInterceptorFactoryInterface>>
 121           creators;
 122       // Add 20 dummy server interceptors
 123       creators.reserve(20);
 124       for (auto i = 0; i < 20; i++) {
 125         creators.push_back(std::unique_ptr<DummyInterceptorFactory>(
 126             new DummyInterceptorFactory()));
 127       }
 128       builder.experimental().SetInterceptorCreators(std::move(creators));
 129     }
 130
 131     server_ = builder.BuildAndStart();
 132     is_server_started_ = true;
 133     if (GetParam().protocol == Protocol::TCP &&
 134         !grpc_iomgr_run_in_background()) {
 135       do_not_test_ = true;
 136     }
 137   }
 138
 139   void ResetStub() {
 140     ChannelArguments args;
 141     auto channel_creds = GetCredentialsProvider()->GetChannelCredentials(
 142         GetParam().credentials_type, &args);
 143     switch (GetParam().protocol) {
 144       case Protocol::TCP:
 145         if (!GetParam().use_interceptors) {
 146           channel_ = ::grpc::CreateCustomChannel(server_address_.str(),
 147                                                  channel_creds, args);
 148         } else {
 149           channel_ = CreateCustomChannelWithInterceptors(
 150               server_address_.str(), channel_creds, args,
 151               CreateDummyClientInterceptors());
 152         }
 153         break;
 154       case Protocol::INPROC:
 155         if (!GetParam().use_interceptors) {
 156           channel_ = server_->InProcessChannel(args);
 157         } else {
 158           channel_ = server_->experimental().InProcessChannelWithInterceptors(
 159               args, CreateDummyClientInterceptors());
 160         }
 161         break;
 162       default:
 163         assert(false);
 164     }
 165     stub_ = grpc::testing::EchoTestService::NewStub(channel_);
 166     generic_stub_.reset(new GenericStub(channel_));
 167     DummyInterceptor::Reset();
 168   }
 169
 170   void TearDown() override {
 171     if (is_server_started_) {
 172       // Although we would normally do an explicit shutdown, the server
 173       // should also work correctly with just a destructor call. The regular
 174       // end2end test uses explicit shutdown, so let this one just do reset.
 175       server_.reset();
 176     }
 177     if (picked_port_ > 0) {
 178       grpc_recycle_unused_port(picked_port_);
 179     }
 180   }
 181
 182   void SendRpcs(int num_rpcs, bool with_binary_metadata) {
 183     grpc::string test_string("");
 184     for (int i = 0; i < num_rpcs; i++) {
 185       EchoRequest request;
 186       EchoResponse response;
 187       ClientContext cli_ctx;
 188
 189       test_string += "Hello world. ";
 190       request.set_message(test_string);
 191       grpc::string val;
 192       if (with_binary_metadata) {
 193         request.mutable_param()->set_echo_metadata(true);
 194         char bytes[8] = {'\0', '\1', '\2', '\3',
 195                          '\4', '\5', '\6', static_cast<char>(i)};
 196         val = grpc::string(bytes, 8);
 197         cli_ctx.AddMetadata("custom-bin", val);
 198       }
 199
 200       cli_ctx.set_compression_algorithm(GRPC_COMPRESS_GZIP);
 201
 202       std::mutex mu;
 203       std::condition_variable cv;
 204       bool done = false;
 205       stub_->experimental_async()->Echo(
 206           &cli_ctx, &request, &response,
 207           [&cli_ctx, &request, &response, &done, &mu, &cv, val,
 208            with_binary_metadata](Status s) {
 209             GPR_ASSERT(s.ok());
 210
 211             EXPECT_EQ(request.message(), response.message());
 212             if (with_binary_metadata) {
 213               EXPECT_EQ(
 214                   1u, cli_ctx.GetServerTrailingMetadata().count("custom-bin"));
 215               EXPECT_EQ(val, ToString(cli_ctx.GetServerTrailingMetadata()
 216                                           .find("custom-bin")
 217                                           ->second));
 218             }
 219             std::lock_guard<std::mutex> l(mu);
 220             done = true;
 221             cv.notify_one();
 222           });
 223       std::unique_lock<std::mutex> l(mu);
 224       while (!done) {
 225         cv.wait(l);
 226       }
 227     }
 228   }
 229
 230   void SendRpcsRawReq(int num_rpcs) {
 231     grpc::string test_string("Hello raw world.");
 232     EchoRequest request;
 233     request.set_message(test_string);
 234     std::unique_ptr<ByteBuffer> send_buf = SerializeToByteBuffer(&request);
 235
 236     for (int i = 0; i < num_rpcs; i++) {
 237       EchoResponse response;
 238       ClientContext cli_ctx;
 239
 240       std::mutex mu;
 241       std::condition_variable cv;
 242       bool done = false;
 243       stub_->experimental_async()->Echo(
 244           &cli_ctx, send_buf.get(), &response,
 245           [&request, &response, &done, &mu, &cv](Status s) {
 246             GPR_ASSERT(s.ok());
 247
 248             EXPECT_EQ(request.message(), response.message());
 249             std::lock_guard<std::mutex> l(mu);
 250             done = true;
 251             cv.notify_one();
 252           });
 253       std::unique_lock<std::mutex> l(mu);
 254       while (!done) {
 255         cv.wait(l);
 256       }
 257     }
 258   }
 259
 260   void SendRpcsGeneric(int num_rpcs, bool maybe_except) {
 261     const grpc::string kMethodName("/grpc.testing.EchoTestService/Echo");
 262     grpc::string test_string("");
 263     for (int i = 0; i < num_rpcs; i++) {
 264       EchoRequest request;
 265       std::unique_ptr<ByteBuffer> send_buf;
 266       ByteBuffer recv_buf;
 267       ClientContext cli_ctx;
 268
 269       test_string += "Hello world. ";
 270       request.set_message(test_string);
 271       send_buf = SerializeToByteBuffer(&request);
 272
 273       std::mutex mu;
 274       std::condition_variable cv;
 275       bool done = false;
 276       generic_stub_->experimental().UnaryCall(
 277           &cli_ctx, kMethodName, send_buf.get(), &recv_buf,
 278           [&request, &recv_buf, &done, &mu, &cv, maybe_except](Status s) {
 279             GPR_ASSERT(s.ok());
 280
 281             EchoResponse response;
 282             EXPECT_TRUE(ParseFromByteBuffer(&recv_buf, &response));
 283             EXPECT_EQ(request.message(), response.message());
 284             std::lock_guard<std::mutex> l(mu);
 285             done = true;
 286             cv.notify_one();
 287 #if GRPC_ALLOW_EXCEPTIONS
 288             if (maybe_except) {
 289               throw - 1;
 290             }
 291 #else
 292             GPR_ASSERT(!maybe_except);
 293 #endif
 294           });
 295       std::unique_lock<std::mutex> l(mu);
 296       while (!done) {
 297         cv.wait(l);
 298       }
 299     }
 300   }
 301
 302   void SendGenericEchoAsBidi(int num_rpcs, int reuses) {
 303     const grpc::string kMethodName("/grpc.testing.EchoTestService/Echo");
 304     grpc::string test_string("");
 305     for (int i = 0; i < num_rpcs; i++) {
 306       test_string += "Hello world. ";
 307       class Client : public grpc::experimental::ClientBidiReactor<ByteBuffer,
 308                                                                   ByteBuffer> {
 309        public:
 310         Client(ClientCallbackEnd2endTest* test, const grpc::string& method_name,
 311                const grpc::string& test_str, int reuses)
 312             : reuses_remaining_(reuses) {
 313           activate_ = [this, test, method_name, test_str] {
 314             if (reuses_remaining_ > 0) {
 315               cli_ctx_.reset(new ClientContext);
 316               reuses_remaining_--;
 317               test->generic_stub_->experimental().PrepareBidiStreamingCall(
 318                   cli_ctx_.get(), method_name, this);
 319               request_.set_message(test_str);
 320               send_buf_ = SerializeToByteBuffer(&request_);
 321               StartWrite(send_buf_.get());
 322               StartRead(&recv_buf_);
 323               StartCall();
 324             } else {
 325               std::unique_lock<std::mutex> l(mu_);
 326               done_ = true;
 327               cv_.notify_one();
 328             }
 329           };
 330           activate_();
 331         }
 332         void OnWriteDone(bool /*ok*/) override { StartWritesDone(); }
 333         void OnReadDone(bool /*ok*/) override {
 334           EchoResponse response;
 335           EXPECT_TRUE(ParseFromByteBuffer(&recv_buf_, &response));
 336           EXPECT_EQ(request_.message(), response.message());
 337         };
 338         void OnDone(const Status& s) override {
 339           EXPECT_TRUE(s.ok());
 340           activate_();
 341         }
 342         void Await() {
 343           std::unique_lock<std::mutex> l(mu_);
 344           while (!done_) {
 345             cv_.wait(l);
 346           }
 347         }
 348
 349         EchoRequest request_;
 350         std::unique_ptr<ByteBuffer> send_buf_;
 351         ByteBuffer recv_buf_;
 352         std::unique_ptr<ClientContext> cli_ctx_;
 353         int reuses_remaining_;
 354         std::function<void()> activate_;
 355         std::mutex mu_;
 356         std::condition_variable cv_;
 357         bool done_ = false;
 358       } rpc{this, kMethodName, test_string, reuses};
 359
 360       rpc.Await();
 361     }
 362   }
 363   bool do_not_test_{false};
 364   bool is_server_started_{false};
 365   int picked_port_{0};
 366   std::shared_ptr<Channel> channel_;
 367   std::unique_ptr<grpc::testing::EchoTestService::Stub> stub_;
 368   std::unique_ptr<grpc::GenericStub> generic_stub_;
 369   TestServiceImpl service_;
 370   CallbackTestServiceImpl callback_service_;
 371   std::unique_ptr<Server> server_;
 372   std::ostringstream server_address_;
 373 };
 374
 375 TEST_P(ClientCallbackEnd2endTest, SimpleRpc) {
 376   MAYBE_SKIP_TEST;
 377   ResetStub();
 378   SendRpcs(1, false);
 379 }
 380
 381 TEST_P(ClientCallbackEnd2endTest, SimpleRpcUnderLockNested) {
 382   MAYBE_SKIP_TEST;
 383   ResetStub();
 384   std::mutex mu1, mu2, mu3;
 385   std::condition_variable cv;
 386   bool done = false;
 387   EchoRequest request1, request2, request3;
 388   request1.set_message("Hello locked world1.");
 389   request2.set_message("Hello locked world2.");
 390   request3.set_message("Hello locked world3.");
 391   EchoResponse response1, response2, response3;
 392   ClientContext cli_ctx1, cli_ctx2, cli_ctx3;
 393   {
 394     std::lock_guard<std::mutex> l(mu1);
 395     stub_->experimental_async()->Echo(
 396         &cli_ctx1, &request1, &response1,
 397         [this, &mu1, &mu2, &mu3, &cv, &done, &request1, &request2, &request3,
 398          &response1, &response2, &response3, &cli_ctx2, &cli_ctx3](Status s1) {
 399           std::lock_guard<std::mutex> l1(mu1);
 400           EXPECT_TRUE(s1.ok());
 401           EXPECT_EQ(request1.message(), response1.message());
 402           // start the second level of nesting
 403           std::unique_lock<std::mutex> l2(mu2);
 404           this->stub_->experimental_async()->Echo(
 405               &cli_ctx2, &request2, &response2,
 406               [this, &mu2, &mu3, &cv, &done, &request2, &request3, &response2,
 407                &response3, &cli_ctx3](Status s2) {
 408                 std::lock_guard<std::mutex> l2(mu2);
 409                 EXPECT_TRUE(s2.ok());
 410                 EXPECT_EQ(request2.message(), response2.message());
 411                 // start the third level of nesting
 412                 std::lock_guard<std::mutex> l3(mu3);
 413                 stub_->experimental_async()->Echo(
 414                     &cli_ctx3, &request3, &response3,
 415                     [&mu3, &cv, &done, &request3, &response3](Status s3) {
 416                       std::lock_guard<std::mutex> l(mu3);
 417                       EXPECT_TRUE(s3.ok());
 418                       EXPECT_EQ(request3.message(), response3.message());
 419                       done = true;
 420                       cv.notify_all();
 421                     });
 422               });
 423         });
 424   }
 425
 426   std::unique_lock<std::mutex> l(mu3);
 427   while (!done) {
 428     cv.wait(l);
 429   }
 430 }
 431
 432 TEST_P(ClientCallbackEnd2endTest, SimpleRpcUnderLock) {
 433   MAYBE_SKIP_TEST;
 434   ResetStub();
 435   std::mutex mu;
 436   std::condition_variable cv;
 437   bool done = false;
 438   EchoRequest request;
 439   request.set_message("Hello locked world.");
 440   EchoResponse response;
 441   ClientContext cli_ctx;
 442   {
 443     std::lock_guard<std::mutex> l(mu);
 444     stub_->experimental_async()->Echo(
 445         &cli_ctx, &request, &response,
 446         [&mu, &cv, &done, &request, &response](Status s) {
 447           std::lock_guard<std::mutex> l(mu);
 448           EXPECT_TRUE(s.ok());
 449           EXPECT_EQ(request.message(), response.message());
 450           done = true;
 451           cv.notify_one();
 452         });
 453   }
 454   std::unique_lock<std::mutex> l(mu);
 455   while (!done) {
 456     cv.wait(l);
 457   }
 458 }
 459
 460 TEST_P(ClientCallbackEnd2endTest, SequentialRpcs) {
 461   MAYBE_SKIP_TEST;
 462   ResetStub();
 463   SendRpcs(10, false);
 464 }
 465
 466 TEST_P(ClientCallbackEnd2endTest, SequentialRpcsRawReq) {
 467   MAYBE_SKIP_TEST;
 468   ResetStub();
 469   SendRpcsRawReq(10);
 470 }
 471
 472 TEST_P(ClientCallbackEnd2endTest, SendClientInitialMetadata) {
 473   MAYBE_SKIP_TEST;
 474   ResetStub();
 475   SimpleRequest request;
 476   SimpleResponse response;
 477   ClientContext cli_ctx;
 478
 479   cli_ctx.AddMetadata(kCheckClientInitialMetadataKey,
 480                       kCheckClientInitialMetadataVal);
 481
 482   std::mutex mu;
 483   std::condition_variable cv;
 484   bool done = false;
 485   stub_->experimental_async()->CheckClientInitialMetadata(
 486       &cli_ctx, &request, &response, [&done, &mu, &cv](Status s) {
 487         GPR_ASSERT(s.ok());
 488
 489         std::lock_guard<std::mutex> l(mu);
 490         done = true;
 491         cv.notify_one();
 492       });
 493   std::unique_lock<std::mutex> l(mu);
 494   while (!done) {
 495     cv.wait(l);
 496   }
 497 }
 498
 499 TEST_P(ClientCallbackEnd2endTest, SimpleRpcWithBinaryMetadata) {
 500   MAYBE_SKIP_TEST;
 501   ResetStub();
 502   SendRpcs(1, true);
 503 }
 504
 505 TEST_P(ClientCallbackEnd2endTest, SequentialRpcsWithVariedBinaryMetadataValue) {
 506   MAYBE_SKIP_TEST;
 507   ResetStub();
 508   SendRpcs(10, true);
 509 }
 510
 511 TEST_P(ClientCallbackEnd2endTest, SequentialGenericRpcs) {
 512   MAYBE_SKIP_TEST;
 513   ResetStub();
 514   SendRpcsGeneric(10, false);
 515 }
 516
 517 TEST_P(ClientCallbackEnd2endTest, SequentialGenericRpcsAsBidi) {
 518   MAYBE_SKIP_TEST;
 519   ResetStub();
 520   SendGenericEchoAsBidi(10, 1);
 521 }
 522
 523 TEST_P(ClientCallbackEnd2endTest, SequentialGenericRpcsAsBidiWithReactorReuse) {
 524   MAYBE_SKIP_TEST;
 525   ResetStub();
 526   SendGenericEchoAsBidi(10, 10);
 527 }
 528
 529 #if GRPC_ALLOW_EXCEPTIONS
 530 TEST_P(ClientCallbackEnd2endTest, ExceptingRpc) {
 531   MAYBE_SKIP_TEST;
 532   ResetStub();
 533   SendRpcsGeneric(10, true);
 534 }
 535 #endif
 536
 537 TEST_P(ClientCallbackEnd2endTest, MultipleRpcsWithVariedBinaryMetadataValue) {
 538   MAYBE_SKIP_TEST;
 539   ResetStub();
 540   std::vector<std::thread> threads;
 541   threads.reserve(10);
 542   for (int i = 0; i < 10; ++i) {
 543     threads.emplace_back([this] { SendRpcs(10, true); });
 544   }
 545   for (int i = 0; i < 10; ++i) {
 546     threads[i].join();
 547   }
 548 }
 549
 550 TEST_P(ClientCallbackEnd2endTest, MultipleRpcs) {
 551   MAYBE_SKIP_TEST;
 552   ResetStub();
 553   std::vector<std::thread> threads;
 554   threads.reserve(10);
 555   for (int i = 0; i < 10; ++i) {
 556     threads.emplace_back([this] { SendRpcs(10, false); });
 557   }
 558   for (int i = 0; i < 10; ++i) {
 559     threads[i].join();
 560   }
 561 }
 562
 563 TEST_P(ClientCallbackEnd2endTest, CancelRpcBeforeStart) {
 564   MAYBE_SKIP_TEST;
 565   ResetStub();
 566   EchoRequest request;
 567   EchoResponse response;
 568   ClientContext context;
 569   request.set_message("hello");
 570   context.TryCancel();
 571
 572   std::mutex mu;
 573   std::condition_variable cv;
 574   bool done = false;
 575   stub_->experimental_async()->Echo(
 576       &context, &request, &response, [&response, &done, &mu, &cv](Status s) {
 577         EXPECT_EQ("", response.message());
 578         EXPECT_EQ(grpc::StatusCode::CANCELLED, s.error_code());
 579         std::lock_guard<std::mutex> l(mu);
 580         done = true;
 581         cv.notify_one();
 582       });
 583   std::unique_lock<std::mutex> l(mu);
 584   while (!done) {
 585     cv.wait(l);
 586   }
 587   if (GetParam().use_interceptors) {
 588     EXPECT_EQ(20, DummyInterceptor::GetNumTimesCancel());
 589   }
 590 }
 591
 592 TEST_P(ClientCallbackEnd2endTest, RequestEchoServerCancel) {
 593   MAYBE_SKIP_TEST;
 594   ResetStub();
 595   EchoRequest request;
 596   EchoResponse response;
 597   ClientContext context;
 598   request.set_message("hello");
 599   context.AddMetadata(kServerTryCancelRequest,
 600                       grpc::to_string(CANCEL_BEFORE_PROCESSING));
 601
 602   std::mutex mu;
 603   std::condition_variable cv;
 604   bool done = false;
 605   stub_->experimental_async()->Echo(
 606       &context, &request, &response, [&done, &mu, &cv](Status s) {
 607         EXPECT_FALSE(s.ok());
 608         EXPECT_EQ(grpc::StatusCode::CANCELLED, s.error_code());
 609         std::lock_guard<std::mutex> l(mu);
 610         done = true;
 611         cv.notify_one();
 612       });
 613   std::unique_lock<std::mutex> l(mu);
 614   while (!done) {
 615     cv.wait(l);
 616   }
 617 }
 618
 619 struct ClientCancelInfo {
 620   bool cancel{false};
 621   int ops_before_cancel;
 622
 623   ClientCancelInfo() : cancel{false} {}
 624   explicit ClientCancelInfo(int ops) : cancel{true}, ops_before_cancel{ops} {}
 625 };
 626
 627 class WriteClient : public grpc::experimental::ClientWriteReactor<EchoRequest> {
 628  public:
 629   WriteClient(grpc::testing::EchoTestService::Stub* stub,
 630               ServerTryCancelRequestPhase server_try_cancel,
 631               int num_msgs_to_send, ClientCancelInfo client_cancel = {})
 632       : server_try_cancel_(server_try_cancel),
 633         num_msgs_to_send_(num_msgs_to_send),
 634         client_cancel_{client_cancel} {
 635     grpc::string msg{"Hello server."};
 636     for (int i = 0; i < num_msgs_to_send; i++) {
 637       desired_ += msg;
 638     }
 639     if (server_try_cancel != DO_NOT_CANCEL) {
 640       // Send server_try_cancel value in the client metadata
 641       context_.AddMetadata(kServerTryCancelRequest,
 642                            grpc::to_string(server_try_cancel));
 643     }
 644     context_.set_initial_metadata_corked(true);
 645     stub->experimental_async()->RequestStream(&context_, &response_, this);
 646     StartCall();
 647     request_.set_message(msg);
 648     MaybeWrite();
 649   }
 650   void OnWriteDone(bool ok) override {
 651     if (ok) {
 652       num_msgs_sent_++;
 653       MaybeWrite();
 654     }
 655   }
 656   void OnDone(const Status& s) override {
 657     gpr_log(GPR_INFO, "Sent %d messages", num_msgs_sent_);
 658     int num_to_send =
 659         (client_cancel_.cancel)
 660             ? std::min(num_msgs_to_send_, client_cancel_.ops_before_cancel)
 661             : num_msgs_to_send_;
 662     switch (server_try_cancel_) {
 663       case CANCEL_BEFORE_PROCESSING:
 664       case CANCEL_DURING_PROCESSING:
 665         // If the RPC is canceled by server before / during messages from the
 666         // client, it means that the client most likely did not get a chance to
 667         // send all the messages it wanted to send. i.e num_msgs_sent <=
 668         // num_msgs_to_send
 669         EXPECT_LE(num_msgs_sent_, num_to_send);
 670         break;
 671       case DO_NOT_CANCEL:
 672       case CANCEL_AFTER_PROCESSING:
 673         // If the RPC was not canceled or canceled after all messages were read
 674         // by the server, the client did get a chance to send all its messages
 675         EXPECT_EQ(num_msgs_sent_, num_to_send);
 676         break;
 677       default:
 678         assert(false);
 679         break;
 680     }
 681     if ((server_try_cancel_ == DO_NOT_CANCEL) && !client_cancel_.cancel) {
 682       EXPECT_TRUE(s.ok());
 683       EXPECT_EQ(response_.message(), desired_);
 684     } else {
 685       EXPECT_FALSE(s.ok());
 686       EXPECT_EQ(grpc::StatusCode::CANCELLED, s.error_code());
 687     }
 688     std::unique_lock<std::mutex> l(mu_);
 689     done_ = true;
 690     cv_.notify_one();
 691   }
 692   void Await() {
 693     std::unique_lock<std::mutex> l(mu_);
 694     while (!done_) {
 695       cv_.wait(l);
 696     }
 697   }
 698
 699  private:
 700   void MaybeWrite() {
 701     if (client_cancel_.cancel &&
 702         num_msgs_sent_ == client_cancel_.ops_before_cancel) {
 703       context_.TryCancel();
 704     } else if (num_msgs_to_send_ > num_msgs_sent_ + 1) {
 705       StartWrite(&request_);
 706     } else if (num_msgs_to_send_ == num_msgs_sent_ + 1) {
 707       StartWriteLast(&request_, WriteOptions());
 708     }
 709   }
 710   EchoRequest request_;
 711   EchoResponse response_;
 712   ClientContext context_;
 713   const ServerTryCancelRequestPhase server_try_cancel_;
 714   int num_msgs_sent_{0};
 715   const int num_msgs_to_send_;
 716   grpc::string desired_;
 717   const ClientCancelInfo client_cancel_;
 718   std::mutex mu_;
 719   std::condition_variable cv_;
 720   bool done_ = false;
 721 };
 722
 723 TEST_P(ClientCallbackEnd2endTest, RequestStream) {
 724   MAYBE_SKIP_TEST;
 725   ResetStub();
 726   WriteClient test{stub_.get(), DO_NOT_CANCEL, 3};
 727   test.Await();
 728   // Make sure that the server interceptors were not notified to cancel
 729   if (GetParam().use_interceptors) {
 730     EXPECT_EQ(0, DummyInterceptor::GetNumTimesCancel());
 731   }
 732 }
 733
 734 TEST_P(ClientCallbackEnd2endTest, ClientCancelsRequestStream) {
 735   MAYBE_SKIP_TEST;
 736   ResetStub();
 737   WriteClient test{stub_.get(), DO_NOT_CANCEL, 3, ClientCancelInfo{2}};
 738   test.Await();
 739   // Make sure that the server interceptors got the cancel
 740   if (GetParam().use_interceptors) {
 741     EXPECT_EQ(20, DummyInterceptor::GetNumTimesCancel());
 742   }
 743 }
 744
 745 // Server to cancel before doing reading the request
 746 TEST_P(ClientCallbackEnd2endTest, RequestStreamServerCancelBeforeReads) {
 747   MAYBE_SKIP_TEST;
 748   ResetStub();
 749   WriteClient test{stub_.get(), CANCEL_BEFORE_PROCESSING, 1};
 750   test.Await();
 751   // Make sure that the server interceptors were notified
 752   if (GetParam().use_interceptors) {
 753     EXPECT_EQ(20, DummyInterceptor::GetNumTimesCancel());
 754   }
 755 }
 756
 757 // Server to cancel while reading a request from the stream in parallel
 758 TEST_P(ClientCallbackEnd2endTest, RequestStreamServerCancelDuringRead) {
 759   MAYBE_SKIP_TEST;
 760   ResetStub();
 761   WriteClient test{stub_.get(), CANCEL_DURING_PROCESSING, 10};
 762   test.Await();
 763   // Make sure that the server interceptors were notified
 764   if (GetParam().use_interceptors) {
 765     EXPECT_EQ(20, DummyInterceptor::GetNumTimesCancel());
 766   }
 767 }
 768
 769 // Server to cancel after reading all the requests but before returning to the
 770 // client
 771 TEST_P(ClientCallbackEnd2endTest, RequestStreamServerCancelAfterReads) {
 772   MAYBE_SKIP_TEST;
 773   ResetStub();
 774   WriteClient test{stub_.get(), CANCEL_AFTER_PROCESSING, 4};
 775   test.Await();
 776   // Make sure that the server interceptors were notified
 777   if (GetParam().use_interceptors) {
 778     EXPECT_EQ(20, DummyInterceptor::GetNumTimesCancel());
 779   }
 780 }
 781
 782 TEST_P(ClientCallbackEnd2endTest, UnaryReactor) {
 783   MAYBE_SKIP_TEST;
 784   ResetStub();
 785   class UnaryClient : public grpc::experimental::ClientUnaryReactor {
 786    public:
 787     UnaryClient(grpc::testing::EchoTestService::Stub* stub) {
 788       cli_ctx_.AddMetadata("key1", "val1");
 789       cli_ctx_.AddMetadata("key2", "val2");
 790       request_.mutable_param()->set_echo_metadata_initially(true);
 791       request_.set_message("Hello metadata");
 792       stub->experimental_async()->Echo(&cli_ctx_, &request_, &response_, this);
 793       StartCall();
 794     }
 795     void OnReadInitialMetadataDone(bool ok) override {
 796       EXPECT_TRUE(ok);
 797       EXPECT_EQ(1u, cli_ctx_.GetServerInitialMetadata().count("key1"));
 798       EXPECT_EQ(
 799           "val1",
 800           ToString(cli_ctx_.GetServerInitialMetadata().find("key1")->second));
 801       EXPECT_EQ(1u, cli_ctx_.GetServerInitialMetadata().count("key2"));
 802       EXPECT_EQ(
 803           "val2",
 804           ToString(cli_ctx_.GetServerInitialMetadata().find("key2")->second));
 805       initial_metadata_done_ = true;
 806     }
 807     void OnDone(const Status& s) override {
 808       EXPECT_TRUE(initial_metadata_done_);
 809       EXPECT_EQ(0u, cli_ctx_.GetServerTrailingMetadata().size());
 810       EXPECT_TRUE(s.ok());
 811       EXPECT_EQ(request_.message(), response_.message());
 812       std::unique_lock<std::mutex> l(mu_);
 813       done_ = true;
 814       cv_.notify_one();
 815     }
 816     void Await() {
 817       std::unique_lock<std::mutex> l(mu_);
 818       while (!done_) {
 819         cv_.wait(l);
 820       }
 821     }
 822
 823    private:
 824     EchoRequest request_;
 825     EchoResponse response_;
 826     ClientContext cli_ctx_;
 827     std::mutex mu_;
 828     std::condition_variable cv_;
 829     bool done_{false};
 830     bool initial_metadata_done_{false};
 831   };
 832
 833   UnaryClient test{stub_.get()};
 834   test.Await();
 835   // Make sure that the server interceptors were not notified of a cancel
 836   if (GetParam().use_interceptors) {
 837     EXPECT_EQ(0, DummyInterceptor::GetNumTimesCancel());
 838   }
 839 }
 840
 841 TEST_P(ClientCallbackEnd2endTest, GenericUnaryReactor) {
 842   MAYBE_SKIP_TEST;
 843   ResetStub();
 844   const grpc::string kMethodName("/grpc.testing.EchoTestService/Echo");
 845   class UnaryClient : public grpc::experimental::ClientUnaryReactor {
 846    public:
 847     UnaryClient(grpc::GenericStub* stub, const grpc::string& method_name) {
 848       cli_ctx_.AddMetadata("key1", "val1");
 849       cli_ctx_.AddMetadata("key2", "val2");
 850       request_.mutable_param()->set_echo_metadata_initially(true);
 851       request_.set_message("Hello metadata");
 852       send_buf_ = SerializeToByteBuffer(&request_);
 853
 854       stub->experimental().PrepareUnaryCall(&cli_ctx_, method_name,
 855                                             send_buf_.get(), &recv_buf_, this);
 856       StartCall();
 857     }
 858     void OnReadInitialMetadataDone(bool ok) override {
 859       EXPECT_TRUE(ok);
 860       EXPECT_EQ(1u, cli_ctx_.GetServerInitialMetadata().count("key1"));
 861       EXPECT_EQ(
 862           "val1",
 863           ToString(cli_ctx_.GetServerInitialMetadata().find("key1")->second));
 864       EXPECT_EQ(1u, cli_ctx_.GetServerInitialMetadata().count("key2"));
 865       EXPECT_EQ(
 866           "val2",
 867           ToString(cli_ctx_.GetServerInitialMetadata().find("key2")->second));
 868       initial_metadata_done_ = true;
 869     }
 870     void OnDone(const Status& s) override {
 871       EXPECT_TRUE(initial_metadata_done_);
 872       EXPECT_EQ(0u, cli_ctx_.GetServerTrailingMetadata().size());
 873       EXPECT_TRUE(s.ok());
 874       EchoResponse response;
 875       EXPECT_TRUE(ParseFromByteBuffer(&recv_buf_, &response));
 876       EXPECT_EQ(request_.message(), response.message());
 877       std::unique_lock<std::mutex> l(mu_);
 878       done_ = true;
 879       cv_.notify_one();
 880     }
 881     void Await() {
 882       std::unique_lock<std::mutex> l(mu_);
 883       while (!done_) {
 884         cv_.wait(l);
 885       }
 886     }
 887
 888    private:
 889     EchoRequest request_;
 890     std::unique_ptr<ByteBuffer> send_buf_;
 891     ByteBuffer recv_buf_;
 892     ClientContext cli_ctx_;
 893     std::mutex mu_;
 894     std::condition_variable cv_;
 895     bool done_{false};
 896     bool initial_metadata_done_{false};
 897   };
 898
 899   UnaryClient test{generic_stub_.get(), kMethodName};
 900   test.Await();
 901   // Make sure that the server interceptors were not notified of a cancel
 902   if (GetParam().use_interceptors) {
 903     EXPECT_EQ(0, DummyInterceptor::GetNumTimesCancel());
 904   }
 905 }
 906
 907 class ReadClient : public grpc::experimental::ClientReadReactor<EchoResponse> {
 908  public:
 909   ReadClient(grpc::testing::EchoTestService::Stub* stub,
 910              ServerTryCancelRequestPhase server_try_cancel,
 911              ClientCancelInfo client_cancel = {})
 912       : server_try_cancel_(server_try_cancel), client_cancel_{client_cancel} {
 913     if (server_try_cancel_ != DO_NOT_CANCEL) {
 914       // Send server_try_cancel value in the client metadata
 915       context_.AddMetadata(kServerTryCancelRequest,
 916                            grpc::to_string(server_try_cancel));
 917     }
 918     request_.set_message("Hello client ");
 919     stub->experimental_async()->ResponseStream(&context_, &request_, this);
 920     if (client_cancel_.cancel &&
 921         reads_complete_ == client_cancel_.ops_before_cancel) {
 922       context_.TryCancel();
 923     }
 924     // Even if we cancel, read until failure because there might be responses
 925     // pending
 926     StartRead(&response_);
 927     StartCall();
 928   }
 929   void OnReadDone(bool ok) override {
 930     if (!ok) {
 931       if (server_try_cancel_ == DO_NOT_CANCEL && !client_cancel_.cancel) {
 932         EXPECT_EQ(reads_complete_, kServerDefaultResponseStreamsToSend);
 933       }
 934     } else {
 935       EXPECT_LE(reads_complete_, kServerDefaultResponseStreamsToSend);
 936       EXPECT_EQ(response_.message(),
 937                 request_.message() + grpc::to_string(reads_complete_));
 938       reads_complete_++;
 939       if (client_cancel_.cancel &&
 940           reads_complete_ == client_cancel_.ops_before_cancel) {
 941         context_.TryCancel();
 942       }
 943       // Even if we cancel, read until failure because there might be responses
 944       // pending
 945       StartRead(&response_);
 946     }
 947   }
 948   void OnDone(const Status& s) override {
 949     gpr_log(GPR_INFO, "Read %d messages", reads_complete_);
 950     switch (server_try_cancel_) {
 951       case DO_NOT_CANCEL:
 952         if (!client_cancel_.cancel || client_cancel_.ops_before_cancel >
 953                                           kServerDefaultResponseStreamsToSend) {
 954           EXPECT_TRUE(s.ok());
 955           EXPECT_EQ(reads_complete_, kServerDefaultResponseStreamsToSend);
 956         } else {
 957           EXPECT_GE(reads_complete_, client_cancel_.ops_before_cancel);
 958           EXPECT_LE(reads_complete_, kServerDefaultResponseStreamsToSend);
 959           // Status might be ok or cancelled depending on whether server
 960           // sent status before client cancel went through
 961           if (!s.ok()) {
 962             EXPECT_EQ(grpc::StatusCode::CANCELLED, s.error_code());
 963           }
 964         }
 965         break;
 966       case CANCEL_BEFORE_PROCESSING:
 967         EXPECT_FALSE(s.ok());
 968         EXPECT_EQ(grpc::StatusCode::CANCELLED, s.error_code());
 969         EXPECT_EQ(reads_complete_, 0);
 970         break;
 971       case CANCEL_DURING_PROCESSING:
 972       case CANCEL_AFTER_PROCESSING:
 973         // If server canceled while writing messages, client must have read
 974         // less than or equal to the expected number of messages. Even if the
 975         // server canceled after writing all messages, the RPC may be canceled
 976         // before the Client got a chance to read all the messages.
 977         EXPECT_FALSE(s.ok());
 978         EXPECT_EQ(grpc::StatusCode::CANCELLED, s.error_code());
 979         EXPECT_LE(reads_complete_, kServerDefaultResponseStreamsToSend);
 980         break;
 981       default:
 982         assert(false);
 983     }
 984     std::unique_lock<std::mutex> l(mu_);
 985     done_ = true;
 986     cv_.notify_one();
 987   }
 988   void Await() {
 989     std::unique_lock<std::mutex> l(mu_);
 990     while (!done_) {
 991       cv_.wait(l);
 992     }
 993   }
 994
 995  private:
 996   EchoRequest request_;
 997   EchoResponse response_;
 998   ClientContext context_;
 999   const ServerTryCancelRequestPhase server_try_cancel_;
1000   int reads_complete_{0};
1001   const ClientCancelInfo client_cancel_;
1002   std::mutex mu_;
1003   std::condition_variable cv_;
1004   bool done_ = false;
1005 };
1006
1007 TEST_P(ClientCallbackEnd2endTest, ResponseStream) {
1008   MAYBE_SKIP_TEST;
1009   ResetStub();
1010   ReadClient test{stub_.get(), DO_NOT_CANCEL};
1011   test.Await();
1012   // Make sure that the server interceptors were not notified of a cancel
1013   if (GetParam().use_interceptors) {
1014     EXPECT_EQ(0, DummyInterceptor::GetNumTimesCancel());
1015   }
1016 }
1017
1018 TEST_P(ClientCallbackEnd2endTest, ClientCancelsResponseStream) {
1019   MAYBE_SKIP_TEST;
1020   ResetStub();
1021   ReadClient test{stub_.get(), DO_NOT_CANCEL, ClientCancelInfo{2}};
1022   test.Await();
1023   // Because cancel in this case races with server finish, we can't be sure that
1024   // server interceptors even see cancellation
1025 }
1026
1027 // Server to cancel before sending any response messages
1028 TEST_P(ClientCallbackEnd2endTest, ResponseStreamServerCancelBefore) {
1029   MAYBE_SKIP_TEST;
1030   ResetStub();
1031   ReadClient test{stub_.get(), CANCEL_BEFORE_PROCESSING};
1032   test.Await();
1033   // Make sure that the server interceptors were notified
1034   if (GetParam().use_interceptors) {
1035     EXPECT_EQ(20, DummyInterceptor::GetNumTimesCancel());
1036   }
1037 }
1038
1039 // Server to cancel while writing a response to the stream in parallel
1040 TEST_P(ClientCallbackEnd2endTest, ResponseStreamServerCancelDuring) {
1041   MAYBE_SKIP_TEST;
1042   ResetStub();
1043   ReadClient test{stub_.get(), CANCEL_DURING_PROCESSING};
1044   test.Await();
1045   // Make sure that the server interceptors were notified
1046   if (GetParam().use_interceptors) {
1047     EXPECT_EQ(20, DummyInterceptor::GetNumTimesCancel());
1048   }
1049 }
1050
1051 // Server to cancel after writing all the respones to the stream but before
1052 // returning to the client
1053 TEST_P(ClientCallbackEnd2endTest, ResponseStreamServerCancelAfter) {
1054   MAYBE_SKIP_TEST;
1055   ResetStub();
1056   ReadClient test{stub_.get(), CANCEL_AFTER_PROCESSING};
1057   test.Await();
1058   // Make sure that the server interceptors were notified
1059   if (GetParam().use_interceptors) {
1060     EXPECT_EQ(20, DummyInterceptor::GetNumTimesCancel());
1061   }
1062 }
1063
1064 class BidiClient
1065     : public grpc::experimental::ClientBidiReactor<EchoRequest, EchoResponse> {
1066  public:
1067   BidiClient(grpc::testing::EchoTestService::Stub* stub,
1068              ServerTryCancelRequestPhase server_try_cancel,
1069              int num_msgs_to_send, ClientCancelInfo client_cancel = {})
1070       : server_try_cancel_(server_try_cancel),
1071         msgs_to_send_{num_msgs_to_send},
1072         client_cancel_{client_cancel} {
1073     if (server_try_cancel_ != DO_NOT_CANCEL) {
1074       // Send server_try_cancel value in the client metadata
1075       context_.AddMetadata(kServerTryCancelRequest,
1076                            grpc::to_string(server_try_cancel));
1077     }
1078     request_.set_message("Hello fren ");
1079     stub->experimental_async()->BidiStream(&context_, this);
1080     MaybeWrite();
1081     StartRead(&response_);
1082     StartCall();
1083   }
1084   void OnReadDone(bool ok) override {
1085     if (!ok) {
1086       if (server_try_cancel_ == DO_NOT_CANCEL) {
1087         if (!client_cancel_.cancel) {
1088           EXPECT_EQ(reads_complete_, msgs_to_send_);
1089         } else {
1090           EXPECT_LE(reads_complete_, writes_complete_);
1091         }
1092       }
1093     } else {
1094       EXPECT_LE(reads_complete_, msgs_to_send_);
1095       EXPECT_EQ(response_.message(), request_.message());
1096       reads_complete_++;
1097       StartRead(&response_);
1098     }
1099   }
1100   void OnWriteDone(bool ok) override {
1101     if (server_try_cancel_ == DO_NOT_CANCEL) {
1102       EXPECT_TRUE(ok);
1103     } else if (!ok) {
1104       return;
1105     }
1106     writes_complete_++;
1107     MaybeWrite();
1108   }
1109   void OnDone(const Status& s) override {
1110     gpr_log(GPR_INFO, "Sent %d messages", writes_complete_);
1111     gpr_log(GPR_INFO, "Read %d messages", reads_complete_);
1112     switch (server_try_cancel_) {
1113       case DO_NOT_CANCEL:
1114         if (!client_cancel_.cancel ||
1115             client_cancel_.ops_before_cancel > msgs_to_send_) {
1116           EXPECT_TRUE(s.ok());
1117           EXPECT_EQ(writes_complete_, msgs_to_send_);
1118           EXPECT_EQ(reads_complete_, writes_complete_);
1119         } else {
1120           EXPECT_FALSE(s.ok());
1121           EXPECT_EQ(grpc::StatusCode::CANCELLED, s.error_code());
1122           EXPECT_EQ(writes_complete_, client_cancel_.ops_before_cancel);
1123           EXPECT_LE(reads_complete_, writes_complete_);
1124         }
1125         break;
1126       case CANCEL_BEFORE_PROCESSING:
1127         EXPECT_FALSE(s.ok());
1128         EXPECT_EQ(grpc::StatusCode::CANCELLED, s.error_code());
1129         // The RPC is canceled before the server did any work or returned any
1130         // reads, but it's possible that some writes took place first from the
1131         // client
1132         EXPECT_LE(writes_complete_, msgs_to_send_);
1133         EXPECT_EQ(reads_complete_, 0);
1134         break;
1135       case CANCEL_DURING_PROCESSING:
1136         EXPECT_FALSE(s.ok());
1137         EXPECT_EQ(grpc::StatusCode::CANCELLED, s.error_code());
1138         EXPECT_LE(writes_complete_, msgs_to_send_);
1139         EXPECT_LE(reads_complete_, writes_complete_);
1140         break;
1141       case CANCEL_AFTER_PROCESSING:
1142         EXPECT_FALSE(s.ok());
1143         EXPECT_EQ(grpc::StatusCode::CANCELLED, s.error_code());
1144         EXPECT_EQ(writes_complete_, msgs_to_send_);
1145         // The Server canceled after reading the last message and after writing
1146         // the message to the client. However, the RPC cancellation might have
1147         // taken effect before the client actually read the response.
1148         EXPECT_LE(reads_complete_, writes_complete_);
1149         break;
1150       default:
1151         assert(false);
1152     }
1153     std::unique_lock<std::mutex> l(mu_);
1154     done_ = true;
1155     cv_.notify_one();
1156   }
1157   void Await() {
1158     std::unique_lock<std::mutex> l(mu_);
1159     while (!done_) {
1160       cv_.wait(l);
1161     }
1162   }
1163
1164  private:
1165   void MaybeWrite() {
1166     if (client_cancel_.cancel &&
1167         writes_complete_ == client_cancel_.ops_before_cancel) {
1168       context_.TryCancel();
1169     } else if (writes_complete_ == msgs_to_send_) {
1170       StartWritesDone();
1171     } else {
1172       StartWrite(&request_);
1173     }
1174   }
1175   EchoRequest request_;
1176   EchoResponse response_;
1177   ClientContext context_;
1178   const ServerTryCancelRequestPhase server_try_cancel_;
1179   int reads_complete_{0};
1180   int writes_complete_{0};
1181   const int msgs_to_send_;
1182   const ClientCancelInfo client_cancel_;
1183   std::mutex mu_;
1184   std::condition_variable cv_;
1185   bool done_ = false;
1186 };
1187
1188 TEST_P(ClientCallbackEnd2endTest, BidiStream) {
1189   MAYBE_SKIP_TEST;
1190   ResetStub();
1191   BidiClient test{stub_.get(), DO_NOT_CANCEL,
1192                   kServerDefaultResponseStreamsToSend};
1193   test.Await();
1194   // Make sure that the server interceptors were not notified of a cancel
1195   if (GetParam().use_interceptors) {
1196     EXPECT_EQ(0, DummyInterceptor::GetNumTimesCancel());
1197   }
1198 }
1199
1200 TEST_P(ClientCallbackEnd2endTest, ClientCancelsBidiStream) {
1201   MAYBE_SKIP_TEST;
1202   ResetStub();
1203   BidiClient test{stub_.get(), DO_NOT_CANCEL,
1204                   kServerDefaultResponseStreamsToSend, ClientCancelInfo{2}};
1205   test.Await();
1206   // Make sure that the server interceptors were notified of a cancel
1207   if (GetParam().use_interceptors) {
1208     EXPECT_EQ(20, DummyInterceptor::GetNumTimesCancel());
1209   }
1210 }
1211
1212 // Server to cancel before reading/writing any requests/responses on the stream
1213 TEST_P(ClientCallbackEnd2endTest, BidiStreamServerCancelBefore) {
1214   MAYBE_SKIP_TEST;
1215   ResetStub();
1216   BidiClient test{stub_.get(), CANCEL_BEFORE_PROCESSING, 2};
1217   test.Await();
1218   // Make sure that the server interceptors were notified
1219   if (GetParam().use_interceptors) {
1220     EXPECT_EQ(20, DummyInterceptor::GetNumTimesCancel());
1221   }
1222 }
1223
1224 // Server to cancel while reading/writing requests/responses on the stream in
1225 // parallel
1226 TEST_P(ClientCallbackEnd2endTest, BidiStreamServerCancelDuring) {
1227   MAYBE_SKIP_TEST;
1228   ResetStub();
1229   BidiClient test{stub_.get(), CANCEL_DURING_PROCESSING, 10};
1230   test.Await();
1231   // Make sure that the server interceptors were notified
1232   if (GetParam().use_interceptors) {
1233     EXPECT_EQ(20, DummyInterceptor::GetNumTimesCancel());
1234   }
1235 }
1236
1237 // Server to cancel after reading/writing all requests/responses on the stream
1238 // but before returning to the client
1239 TEST_P(ClientCallbackEnd2endTest, BidiStreamServerCancelAfter) {
1240   MAYBE_SKIP_TEST;
1241   ResetStub();
1242   BidiClient test{stub_.get(), CANCEL_AFTER_PROCESSING, 5};
1243   test.Await();
1244   // Make sure that the server interceptors were notified
1245   if (GetParam().use_interceptors) {
1246     EXPECT_EQ(20, DummyInterceptor::GetNumTimesCancel());
1247   }
1248 }
1249
1250 TEST_P(ClientCallbackEnd2endTest, SimultaneousReadAndWritesDone) {
1251   MAYBE_SKIP_TEST;
1252   ResetStub();
1253   class Client : public grpc::experimental::ClientBidiReactor<EchoRequest,
1254                                                               EchoResponse> {
1255    public:
1256     Client(grpc::testing::EchoTestService::Stub* stub) {
1257       request_.set_message("Hello bidi ");
1258       stub->experimental_async()->BidiStream(&context_, this);
1259       StartWrite(&request_);
1260       StartCall();
1261     }
1262     void OnReadDone(bool ok) override {
1263       EXPECT_TRUE(ok);
1264       EXPECT_EQ(response_.message(), request_.message());
1265     }
1266     void OnWriteDone(bool ok) override {
1267       EXPECT_TRUE(ok);
1268       // Now send out the simultaneous Read and WritesDone
1269       StartWritesDone();
1270       StartRead(&response_);
1271     }
1272     void OnDone(const Status& s) override {
1273       EXPECT_TRUE(s.ok());
1274       EXPECT_EQ(response_.message(), request_.message());
1275       std::unique_lock<std::mutex> l(mu_);
1276       done_ = true;
1277       cv_.notify_one();
1278     }
1279     void Await() {
1280       std::unique_lock<std::mutex> l(mu_);
1281       while (!done_) {
1282         cv_.wait(l);
1283       }
1284     }
1285
1286    private:
1287     EchoRequest request_;
1288     EchoResponse response_;
1289     ClientContext context_;
1290     std::mutex mu_;
1291     std::condition_variable cv_;
1292     bool done_ = false;
1293   } test{stub_.get()};
1294
1295   test.Await();
1296 }
1297
1298 TEST_P(ClientCallbackEnd2endTest, UnimplementedRpc) {
1299   MAYBE_SKIP_TEST;
1300   ChannelArguments args;
1301   const auto& channel_creds = GetCredentialsProvider()->GetChannelCredentials(
1302       GetParam().credentials_type, &args);
1303   std::shared_ptr<Channel> channel =
1304       (GetParam().protocol == Protocol::TCP)
1305           ? ::grpc::CreateCustomChannel(server_address_.str(), channel_creds,
1306                                         args)
1307           : server_->InProcessChannel(args);
1308   std::unique_ptr<grpc::testing::UnimplementedEchoService::Stub> stub;
1309   stub = grpc::testing::UnimplementedEchoService::NewStub(channel);
1310   EchoRequest request;
1311   EchoResponse response;
1312   ClientContext cli_ctx;
1313   request.set_message("Hello world.");
1314   std::mutex mu;
1315   std::condition_variable cv;
1316   bool done = false;
1317   stub->experimental_async()->Unimplemented(
1318       &cli_ctx, &request, &response, [&done, &mu, &cv](Status s) {
1319         EXPECT_EQ(StatusCode::UNIMPLEMENTED, s.error_code());
1320         EXPECT_EQ("", s.error_message());
1321
1322         std::lock_guard<std::mutex> l(mu);
1323         done = true;
1324         cv.notify_one();
1325       });
1326   std::unique_lock<std::mutex> l(mu);
1327   while (!done) {
1328     cv.wait(l);
1329   }
1330 }
1331
1332 TEST_P(ClientCallbackEnd2endTest,
1333        ResponseStreamExtraReactionFlowReadsUntilDone) {
1334   MAYBE_SKIP_TEST;
1335   ResetStub();
1336   class ReadAllIncomingDataClient
1337       : public grpc::experimental::ClientReadReactor<EchoResponse> {
1338    public:
1339     ReadAllIncomingDataClient(grpc::testing::EchoTestService::Stub* stub) {
1340       request_.set_message("Hello client ");
1341       stub->experimental_async()->ResponseStream(&context_, &request_, this);
1342     }
1343     bool WaitForReadDone() {
1344       std::unique_lock<std::mutex> l(mu_);
1345       while (!read_done_) {
1346         read_cv_.wait(l);
1347       }
1348       read_done_ = false;
1349       return read_ok_;
1350     }
1351     void Await() {
1352       std::unique_lock<std::mutex> l(mu_);
1353       while (!done_) {
1354         done_cv_.wait(l);
1355       }
1356     }
1357     const Status& status() {
1358       std::unique_lock<std::mutex> l(mu_);
1359       return status_;
1360     }
1361
1362    private:
1363     void OnReadDone(bool ok) override {
1364       std::unique_lock<std::mutex> l(mu_);
1365       read_ok_ = ok;
1366       read_done_ = true;
1367       read_cv_.notify_one();
1368     }
1369     void OnDone(const Status& s) override {
1370       std::unique_lock<std::mutex> l(mu_);
1371       done_ = true;
1372       status_ = s;
1373       done_cv_.notify_one();
1374     }
1375
1376     EchoRequest request_;
1377     EchoResponse response_;
1378     ClientContext context_;
1379     bool read_ok_ = false;
1380     bool read_done_ = false;
1381     std::mutex mu_;
1382     std::condition_variable read_cv_;
1383     std::condition_variable done_cv_;
1384     bool done_ = false;
1385     Status status_;
1386   } client{stub_.get()};
1387
1388   int reads_complete = 0;
1389   client.AddHold();
1390   client.StartCall();
1391
1392   EchoResponse response;
1393   bool read_ok = true;
1394   while (read_ok) {
1395     client.StartRead(&response);
1396     read_ok = client.WaitForReadDone();
1397     if (read_ok) {
1398       ++reads_complete;
1399     }
1400   }
1401   client.RemoveHold();
1402   client.Await();
1403
1404   EXPECT_EQ(kServerDefaultResponseStreamsToSend, reads_complete);
1405   EXPECT_EQ(client.status().error_code(), grpc::StatusCode::OK);
1406 }
1407
1408 std::vector<TestScenario> CreateTestScenarios(bool test_insecure) {
1409 #if TARGET_OS_IPHONE
1410   // Workaround Apple CFStream bug
1411   gpr_setenv("grpc_cfstream", "0");
1412 #endif
1413
1414   std::vector<TestScenario> scenarios;
1415   std::vector<grpc::string> credentials_types{
1416       GetCredentialsProvider()->GetSecureCredentialsTypeList()};
1417   auto insec_ok = [] {
1418     // Only allow insecure credentials type when it is registered with the
1419     // provider. User may create providers that do not have insecure.
1420     return GetCredentialsProvider()->GetChannelCredentials(
1421                kInsecureCredentialsType, nullptr) != nullptr;
1422   };
1423   if (test_insecure && insec_ok()) {
1424     credentials_types.push_back(kInsecureCredentialsType);
1425   }
1426   GPR_ASSERT(!credentials_types.empty());
1427
1428   bool barr[]{false, true};
1429   Protocol parr[]{Protocol::INPROC, Protocol::TCP};
1430   for (Protocol p : parr) {
1431     for (const auto& cred : credentials_types) {
1432       // TODO(vjpai): Test inproc with secure credentials when feasible
1433       if (p == Protocol::INPROC &&
1434           (cred != kInsecureCredentialsType || !insec_ok())) {
1435         continue;
1436       }
1437       for (bool callback_server : barr) {
1438         for (bool use_interceptors : barr) {
1439           scenarios.emplace_back(callback_server, p, use_interceptors, cred);
1440         }
1441       }
1442     }
1443   }
1444   return scenarios;
1445 }
1446
1447 INSTANTIATE_TEST_SUITE_P(ClientCallbackEnd2endTest, ClientCallbackEnd2endTest,
1448                          ::testing::ValuesIn(CreateTestScenarios(true)));
1449
1450 }  // namespace
1451 }  // namespace testing
1452 }  // namespace grpc
1453
1454 int main(int argc, char** argv) {
1455   ::testing::InitGoogleTest(&argc, argv);
1456   grpc::testing::TestEnvironment env(argc, argv);
1457   grpc_init();
1458   int ret = RUN_ALL_TESTS();
1459   grpc_shutdown();
1460   return ret;
1461 }