test/cpp/end2end/test_service_impl.h

   1 /*
   2  *
   3  * Copyright 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 #ifndef GRPC_TEST_CPP_END2END_TEST_SERVICE_IMPL_H
  20 #define GRPC_TEST_CPP_END2END_TEST_SERVICE_IMPL_H
  21
  22 #include <condition_variable>
  23 #include <memory>
  24 #include <mutex>
  25
  26 #include <grpc/grpc.h>
  27 #include <grpc/support/log.h>
  28 #include <grpcpp/alarm.h>
  29 #include <grpcpp/security/credentials.h>
  30 #include <grpcpp/server_context.h>
  31 #include <gtest/gtest.h>
  32
  33 #include <string>
  34 #include <thread>
  35
  36 #include "src/proto/grpc/testing/echo.grpc.pb.h"
  37 #include "test/cpp/util/string_ref_helper.h"
  38
  39 namespace grpc {
  40 namespace testing {
  41
  42 const int kServerDefaultResponseStreamsToSend = 3;
  43 const char* const kServerResponseStreamsToSend = "server_responses_to_send";
  44 const char* const kServerTryCancelRequest = "server_try_cancel";
  45 const char* const kClientTryCancelRequest = "client_try_cancel";
  46 const char* const kDebugInfoTrailerKey = "debug-info-bin";
  47 const char* const kServerFinishAfterNReads = "server_finish_after_n_reads";
  48 const char* const kServerUseCoalescingApi = "server_use_coalescing_api";
  49 const char* const kCheckClientInitialMetadataKey = "custom_client_metadata";
  50 const char* const kCheckClientInitialMetadataVal = "Value for client metadata";
  51
  52 typedef enum {
  53   DO_NOT_CANCEL = 0,
  54   CANCEL_BEFORE_PROCESSING,
  55   CANCEL_DURING_PROCESSING,
  56   CANCEL_AFTER_PROCESSING
  57 } ServerTryCancelRequestPhase;
  58
  59 namespace internal {
  60 // When echo_deadline is requested, deadline seen in the ServerContext is set in
  61 // the response in seconds.
  62 void MaybeEchoDeadline(ServerContextBase* context, const EchoRequest* request,
  63                        EchoResponse* response);
  64
  65 void CheckServerAuthContext(const ServerContextBase* context,
  66                             const std::string& expected_transport_security_type,
  67                             const std::string& expected_client_identity);
  68
  69 // Returns the number of pairs in metadata that exactly match the given
  70 // key-value pair. Returns -1 if the pair wasn't found.
  71 int MetadataMatchCount(
  72     const std::multimap<grpc::string_ref, grpc::string_ref>& metadata,
  73     const std::string& key, const std::string& value);
  74
  75 int GetIntValueFromMetadataHelper(
  76     const char* key,
  77     const std::multimap<grpc::string_ref, grpc::string_ref>& metadata,
  78     int default_value);
  79
  80 int GetIntValueFromMetadata(
  81     const char* key,
  82     const std::multimap<grpc::string_ref, grpc::string_ref>& metadata,
  83     int default_value);
  84
  85 void ServerTryCancel(ServerContext* context);
  86 }  // namespace internal
  87
  88 class TestServiceSignaller {
  89  public:
  90   void ClientWaitUntilRpcStarted() {
  91     std::unique_lock<std::mutex> lock(mu_);
  92     cv_rpc_started_.wait(lock, [this] { return rpc_started_; });
  93   }
  94   void ServerWaitToContinue() {
  95     std::unique_lock<std::mutex> lock(mu_);
  96     cv_server_continue_.wait(lock, [this] { return server_should_continue_; });
  97   }
  98   void SignalClientThatRpcStarted() {
  99     std::unique_lock<std::mutex> lock(mu_);
 100     rpc_started_ = true;
 101     cv_rpc_started_.notify_one();
 102   }
 103   void SignalServerToContinue() {
 104     std::unique_lock<std::mutex> lock(mu_);
 105     server_should_continue_ = true;
 106     cv_server_continue_.notify_one();
 107   }
 108
 109  private:
 110   std::mutex mu_;
 111   std::condition_variable cv_rpc_started_;
 112   bool rpc_started_ /* GUARDED_BY(mu_) */ = false;
 113   std::condition_variable cv_server_continue_;
 114   bool server_should_continue_ /* GUARDED_BY(mu_) */ = false;
 115 };
 116
 117 template <typename RpcService>
 118 class TestMultipleServiceImpl : public RpcService {
 119  public:
 120   TestMultipleServiceImpl() : signal_client_(false), host_() {}
 121   explicit TestMultipleServiceImpl(const std::string& host)
 122       : signal_client_(false), host_(new std::string(host)) {}
 123
 124   Status Echo(ServerContext* context, const EchoRequest* request,
 125               EchoResponse* response) {
 126     if (request->has_param() &&
 127         request->param().server_notify_client_when_started()) {
 128       signaller_.SignalClientThatRpcStarted();
 129       signaller_.ServerWaitToContinue();
 130     }
 131
 132     // A bit of sleep to make sure that short deadline tests fail
 133     if (request->has_param() && request->param().server_sleep_us() > 0) {
 134       gpr_sleep_until(
 135           gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC),
 136                        gpr_time_from_micros(request->param().server_sleep_us(),
 137                                             GPR_TIMESPAN)));
 138     }
 139
 140     if (request->has_param() && request->param().server_die()) {
 141       gpr_log(GPR_ERROR, "The request should not reach application handler.");
 142       GPR_ASSERT(0);
 143     }
 144     if (request->has_param() && request->param().has_expected_error()) {
 145       const auto& error = request->param().expected_error();
 146       return Status(static_cast<StatusCode>(error.code()),
 147                     error.error_message(), error.binary_error_details());
 148     }
 149     int server_try_cancel = internal::GetIntValueFromMetadata(
 150         kServerTryCancelRequest, context->client_metadata(), DO_NOT_CANCEL);
 151     if (server_try_cancel > DO_NOT_CANCEL) {
 152       // Since this is a unary RPC, by the time this server handler is called,
 153       // the 'request' message is already read from the client. So the scenarios
 154       // in server_try_cancel don't make much sense. Just cancel the RPC as long
 155       // as server_try_cancel is not DO_NOT_CANCEL
 156       internal::ServerTryCancel(context);
 157       return Status::CANCELLED;
 158     }
 159
 160     response->set_message(request->message());
 161     internal::MaybeEchoDeadline(context, request, response);
 162     if (host_) {
 163       response->mutable_param()->set_host(*host_);
 164     }
 165     if (request->has_param() && request->param().client_cancel_after_us()) {
 166       {
 167         std::unique_lock<std::mutex> lock(mu_);
 168         signal_client_ = true;
 169         ++rpcs_waiting_for_client_cancel_;
 170       }
 171       while (!context->IsCancelled()) {
 172         gpr_sleep_until(gpr_time_add(
 173             gpr_now(GPR_CLOCK_REALTIME),
 174             gpr_time_from_micros(request->param().client_cancel_after_us(),
 175                                  GPR_TIMESPAN)));
 176       }
 177       {
 178         std::unique_lock<std::mutex> lock(mu_);
 179         --rpcs_waiting_for_client_cancel_;
 180       }
 181       return Status::CANCELLED;
 182     } else if (request->has_param() &&
 183                request->param().server_cancel_after_us()) {
 184       gpr_sleep_until(gpr_time_add(
 185           gpr_now(GPR_CLOCK_REALTIME),
 186           gpr_time_from_micros(request->param().server_cancel_after_us(),
 187                                GPR_TIMESPAN)));
 188       return Status::CANCELLED;
 189     } else if (!request->has_param() ||
 190                !request->param().skip_cancelled_check()) {
 191       EXPECT_FALSE(context->IsCancelled());
 192     }
 193
 194     if (request->has_param() && request->param().echo_metadata_initially()) {
 195       const std::multimap<grpc::string_ref, grpc::string_ref>& client_metadata =
 196           context->client_metadata();
 197       for (const auto& metadatum : client_metadata) {
 198         context->AddInitialMetadata(ToString(metadatum.first),
 199                                     ToString(metadatum.second));
 200       }
 201     }
 202
 203     if (request->has_param() && request->param().echo_metadata()) {
 204       const std::multimap<grpc::string_ref, grpc::string_ref>& client_metadata =
 205           context->client_metadata();
 206       for (const auto& metadatum : client_metadata) {
 207         context->AddTrailingMetadata(ToString(metadatum.first),
 208                                      ToString(metadatum.second));
 209       }
 210       // Terminate rpc with error and debug info in trailer.
 211       if (request->param().debug_info().stack_entries_size() ||
 212           !request->param().debug_info().detail().empty()) {
 213         std::string serialized_debug_info =
 214             request->param().debug_info().SerializeAsString();
 215         context->AddTrailingMetadata(kDebugInfoTrailerKey,
 216                                      serialized_debug_info);
 217         return Status::CANCELLED;
 218       }
 219     }
 220     if (request->has_param() &&
 221         (request->param().expected_client_identity().length() > 0 ||
 222          request->param().check_auth_context())) {
 223       internal::CheckServerAuthContext(
 224           context, request->param().expected_transport_security_type(),
 225           request->param().expected_client_identity());
 226     }
 227     if (request->has_param() &&
 228         request->param().response_message_length() > 0) {
 229       response->set_message(
 230           std::string(request->param().response_message_length(), '\0'));
 231     }
 232     if (request->has_param() && request->param().echo_peer()) {
 233       response->mutable_param()->set_peer(context->peer());
 234     }
 235     return Status::OK;
 236   }
 237
 238   Status Echo1(ServerContext* context, const EchoRequest* request,
 239                EchoResponse* response) {
 240     return Echo(context, request, response);
 241   }
 242
 243   Status Echo2(ServerContext* context, const EchoRequest* request,
 244                EchoResponse* response) {
 245     return Echo(context, request, response);
 246   }
 247
 248   Status CheckClientInitialMetadata(ServerContext* context,
 249                                     const SimpleRequest* /*request*/,
 250                                     SimpleResponse* /*response*/) {
 251     EXPECT_EQ(internal::MetadataMatchCount(context->client_metadata(),
 252                                            kCheckClientInitialMetadataKey,
 253                                            kCheckClientInitialMetadataVal),
 254               1);
 255     EXPECT_EQ(1u,
 256               context->client_metadata().count(kCheckClientInitialMetadataKey));
 257     return Status::OK;
 258   }
 259
 260   // Unimplemented is left unimplemented to test the returned error.
 261
 262   Status RequestStream(ServerContext* context,
 263                        ServerReader<EchoRequest>* reader,
 264                        EchoResponse* response) {
 265     // If 'server_try_cancel' is set in the metadata, the RPC is cancelled by
 266     // the server by calling ServerContext::TryCancel() depending on the value:
 267     //   CANCEL_BEFORE_PROCESSING: The RPC is cancelled before the server reads
 268     //   any message from the client
 269     //   CANCEL_DURING_PROCESSING: The RPC is cancelled while the server is
 270     //   reading messages from the client
 271     //   CANCEL_AFTER_PROCESSING: The RPC is cancelled after the server reads
 272     //   all the messages from the client
 273     int server_try_cancel = internal::GetIntValueFromMetadata(
 274         kServerTryCancelRequest, context->client_metadata(), DO_NOT_CANCEL);
 275
 276     EchoRequest request;
 277     response->set_message("");
 278
 279     if (server_try_cancel == CANCEL_BEFORE_PROCESSING) {
 280       internal::ServerTryCancel(context);
 281       return Status::CANCELLED;
 282     }
 283
 284     std::thread* server_try_cancel_thd = nullptr;
 285     if (server_try_cancel == CANCEL_DURING_PROCESSING) {
 286       server_try_cancel_thd =
 287           new std::thread([context] { internal::ServerTryCancel(context); });
 288     }
 289
 290     int num_msgs_read = 0;
 291     while (reader->Read(&request)) {
 292       response->mutable_message()->append(request.message());
 293     }
 294     gpr_log(GPR_INFO, "Read: %d messages", num_msgs_read);
 295
 296     if (server_try_cancel_thd != nullptr) {
 297       server_try_cancel_thd->join();
 298       delete server_try_cancel_thd;
 299       return Status::CANCELLED;
 300     }
 301
 302     if (server_try_cancel == CANCEL_AFTER_PROCESSING) {
 303       internal::ServerTryCancel(context);
 304       return Status::CANCELLED;
 305     }
 306
 307     return Status::OK;
 308   }
 309
 310   // Return 'kNumResponseStreamMsgs' messages.
 311   // TODO(yangg) make it generic by adding a parameter into EchoRequest
 312   Status ResponseStream(ServerContext* context, const EchoRequest* request,
 313                         ServerWriter<EchoResponse>* writer) {
 314     // If server_try_cancel is set in the metadata, the RPC is cancelled by the
 315     // server by calling ServerContext::TryCancel() depending on the value:
 316     //   CANCEL_BEFORE_PROCESSING: The RPC is cancelled before the server writes
 317     //   any messages to the client
 318     //   CANCEL_DURING_PROCESSING: The RPC is cancelled while the server is
 319     //   writing messages to the client
 320     //   CANCEL_AFTER_PROCESSING: The RPC is cancelled after the server writes
 321     //   all the messages to the client
 322     int server_try_cancel = internal::GetIntValueFromMetadata(
 323         kServerTryCancelRequest, context->client_metadata(), DO_NOT_CANCEL);
 324
 325     int server_coalescing_api = internal::GetIntValueFromMetadata(
 326         kServerUseCoalescingApi, context->client_metadata(), 0);
 327
 328     int server_responses_to_send = internal::GetIntValueFromMetadata(
 329         kServerResponseStreamsToSend, context->client_metadata(),
 330         kServerDefaultResponseStreamsToSend);
 331
 332     if (server_try_cancel == CANCEL_BEFORE_PROCESSING) {
 333       internal::ServerTryCancel(context);
 334       return Status::CANCELLED;
 335     }
 336
 337     EchoResponse response;
 338     std::thread* server_try_cancel_thd = nullptr;
 339     if (server_try_cancel == CANCEL_DURING_PROCESSING) {
 340       server_try_cancel_thd =
 341           new std::thread([context] { internal::ServerTryCancel(context); });
 342     }
 343
 344     for (int i = 0; i < server_responses_to_send; i++) {
 345       response.set_message(request->message() + std::to_string(i));
 346       if (i == server_responses_to_send - 1 && server_coalescing_api != 0) {
 347         writer->WriteLast(response, WriteOptions());
 348       } else {
 349         writer->Write(response);
 350       }
 351     }
 352
 353     if (server_try_cancel_thd != nullptr) {
 354       server_try_cancel_thd->join();
 355       delete server_try_cancel_thd;
 356       return Status::CANCELLED;
 357     }
 358
 359     if (server_try_cancel == CANCEL_AFTER_PROCESSING) {
 360       internal::ServerTryCancel(context);
 361       return Status::CANCELLED;
 362     }
 363
 364     return Status::OK;
 365   }
 366
 367   Status BidiStream(ServerContext* context,
 368                     ServerReaderWriter<EchoResponse, EchoRequest>* stream) {
 369     // If server_try_cancel is set in the metadata, the RPC is cancelled by the
 370     // server by calling ServerContext::TryCancel() depending on the value:
 371     //   CANCEL_BEFORE_PROCESSING: The RPC is cancelled before the server reads/
 372     //   writes any messages from/to the client
 373     //   CANCEL_DURING_PROCESSING: The RPC is cancelled while the server is
 374     //   reading/writing messages from/to the client
 375     //   CANCEL_AFTER_PROCESSING: The RPC is cancelled after the server
 376     //   reads/writes all messages from/to the client
 377     int server_try_cancel = internal::GetIntValueFromMetadata(
 378         kServerTryCancelRequest, context->client_metadata(), DO_NOT_CANCEL);
 379
 380     EchoRequest request;
 381     EchoResponse response;
 382
 383     if (server_try_cancel == CANCEL_BEFORE_PROCESSING) {
 384       internal::ServerTryCancel(context);
 385       return Status::CANCELLED;
 386     }
 387
 388     std::thread* server_try_cancel_thd = nullptr;
 389     if (server_try_cancel == CANCEL_DURING_PROCESSING) {
 390       server_try_cancel_thd =
 391           new std::thread([context] { internal::ServerTryCancel(context); });
 392     }
 393
 394     // kServerFinishAfterNReads suggests after how many reads, the server should
 395     // write the last message and send status (coalesced using WriteLast)
 396     int server_write_last = internal::GetIntValueFromMetadata(
 397         kServerFinishAfterNReads, context->client_metadata(), 0);
 398
 399     int read_counts = 0;
 400     while (stream->Read(&request)) {
 401       read_counts++;
 402       gpr_log(GPR_INFO, "recv msg %s", request.message().c_str());
 403       response.set_message(request.message());
 404       if (read_counts == server_write_last) {
 405         stream->WriteLast(response, WriteOptions());
 406       } else {
 407         stream->Write(response);
 408       }
 409     }
 410
 411     if (server_try_cancel_thd != nullptr) {
 412       server_try_cancel_thd->join();
 413       delete server_try_cancel_thd;
 414       return Status::CANCELLED;
 415     }
 416
 417     if (server_try_cancel == CANCEL_AFTER_PROCESSING) {
 418       internal::ServerTryCancel(context);
 419       return Status::CANCELLED;
 420     }
 421
 422     return Status::OK;
 423   }
 424
 425   // Unimplemented is left unimplemented to test the returned error.
 426   bool signal_client() {
 427     std::unique_lock<std::mutex> lock(mu_);
 428     return signal_client_;
 429   }
 430   void ClientWaitUntilRpcStarted() { signaller_.ClientWaitUntilRpcStarted(); }
 431   void SignalServerToContinue() { signaller_.SignalServerToContinue(); }
 432   uint64_t RpcsWaitingForClientCancel() {
 433     std::unique_lock<std::mutex> lock(mu_);
 434     return rpcs_waiting_for_client_cancel_;
 435   }
 436
 437  private:
 438   bool signal_client_;
 439   std::mutex mu_;
 440   TestServiceSignaller signaller_;
 441   std::unique_ptr<std::string> host_;
 442   uint64_t rpcs_waiting_for_client_cancel_ = 0;
 443 };
 444
 445 class CallbackTestServiceImpl
 446     : public ::grpc::testing::EchoTestService::CallbackService {
 447  public:
 448   CallbackTestServiceImpl() : signal_client_(false), host_() {}
 449   explicit CallbackTestServiceImpl(const std::string& host)
 450       : signal_client_(false), host_(new std::string(host)) {}
 451
 452   ServerUnaryReactor* Echo(CallbackServerContext* context,
 453                            const EchoRequest* request,
 454                            EchoResponse* response) override;
 455
 456   ServerUnaryReactor* CheckClientInitialMetadata(CallbackServerContext* context,
 457                                                  const SimpleRequest*,
 458                                                  SimpleResponse*) override;
 459
 460   ServerReadReactor<EchoRequest>* RequestStream(
 461       CallbackServerContext* context, EchoResponse* response) override;
 462
 463   ServerWriteReactor<EchoResponse>* ResponseStream(
 464       CallbackServerContext* context, const EchoRequest* request) override;
 465
 466   ServerBidiReactor<EchoRequest, EchoResponse>* BidiStream(
 467       CallbackServerContext* context) override;
 468
 469   // Unimplemented is left unimplemented to test the returned error.
 470   bool signal_client() {
 471     std::unique_lock<std::mutex> lock(mu_);
 472     return signal_client_;
 473   }
 474   void ClientWaitUntilRpcStarted() { signaller_.ClientWaitUntilRpcStarted(); }
 475   void SignalServerToContinue() { signaller_.SignalServerToContinue(); }
 476
 477  private:
 478   bool signal_client_;
 479   std::mutex mu_;
 480   TestServiceSignaller signaller_;
 481   std::unique_ptr<std::string> host_;
 482 };
 483
 484 using TestServiceImpl =
 485     TestMultipleServiceImpl<::grpc::testing::EchoTestService::Service>;
 486
 487 }  // namespace testing
 488 }  // namespace grpc
 489
 490 #endif  // GRPC_TEST_CPP_END2END_TEST_SERVICE_IMPL_H