test/cpp/end2end/generic_end2end_test.cc

   1 /*
   2  *
   3  * Copyright 2015 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 <memory>
  20 #include <thread>
  21
  22 #include <grpc/grpc.h>
  23 #include <grpc/support/time.h>
  24 #include <grpcpp/channel.h>
  25 #include <grpcpp/client_context.h>
  26 #include <grpcpp/create_channel.h>
  27 #include <grpcpp/generic/async_generic_service.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/slice.h>
  34
  35 #include "src/proto/grpc/testing/echo.grpc.pb.h"
  36 #include "test/core/util/port.h"
  37 #include "test/core/util/test_config.h"
  38 #include "test/cpp/util/byte_buffer_proto_helper.h"
  39
  40 #include <gtest/gtest.h>
  41
  42 using grpc::testing::EchoRequest;
  43 using grpc::testing::EchoResponse;
  44 using std::chrono::system_clock;
  45
  46 namespace grpc {
  47 namespace testing {
  48 namespace {
  49
  50 void* tag(int i) { return (void*)static_cast<intptr_t>(i); }
  51
  52 void verify_ok(CompletionQueue* cq, int i, bool expect_ok) {
  53   bool ok;
  54   void* got_tag;
  55   EXPECT_TRUE(cq->Next(&got_tag, &ok));
  56   EXPECT_EQ(expect_ok, ok);
  57   EXPECT_EQ(tag(i), got_tag);
  58 }
  59
  60 class GenericEnd2endTest : public ::testing::Test {
  61  protected:
  62   GenericEnd2endTest() : server_host_("localhost") {}
  63
  64   void SetUp() override {
  65     shut_down_ = false;
  66     int port = grpc_pick_unused_port_or_die();
  67     server_address_ << server_host_ << ":" << port;
  68     // Setup server
  69     ServerBuilder builder;
  70     builder.AddListeningPort(server_address_.str(),
  71                              InsecureServerCredentials());
  72     builder.RegisterAsyncGenericService(&generic_service_);
  73     // Include a second call to RegisterAsyncGenericService to make sure that
  74     // we get an error in the log, since it is not allowed to have 2 async
  75     // generic services
  76     builder.RegisterAsyncGenericService(&generic_service_);
  77     srv_cq_ = builder.AddCompletionQueue();
  78     server_ = builder.BuildAndStart();
  79   }
  80
  81   void ShutDownServerAndCQs() {
  82     if (!shut_down_) {
  83       server_->Shutdown();
  84       void* ignored_tag;
  85       bool ignored_ok;
  86       cli_cq_.Shutdown();
  87       srv_cq_->Shutdown();
  88       while (cli_cq_.Next(&ignored_tag, &ignored_ok))
  89         ;
  90       while (srv_cq_->Next(&ignored_tag, &ignored_ok))
  91         ;
  92       shut_down_ = true;
  93     }
  94   }
  95   void TearDown() override { ShutDownServerAndCQs(); }
  96
  97   void ResetStub() {
  98     std::shared_ptr<Channel> channel = grpc::CreateChannel(
  99         server_address_.str(), InsecureChannelCredentials());
 100     stub_ = grpc::testing::EchoTestService::NewStub(channel);
 101     generic_stub_.reset(new GenericStub(channel));
 102   }
 103
 104   void server_ok(int i) { verify_ok(srv_cq_.get(), i, true); }
 105   void client_ok(int i) { verify_ok(&cli_cq_, i, true); }
 106   void server_fail(int i) { verify_ok(srv_cq_.get(), i, false); }
 107   void client_fail(int i) { verify_ok(&cli_cq_, i, false); }
 108
 109   void SendRpc(int num_rpcs) {
 110     SendRpc(num_rpcs, false, gpr_inf_future(GPR_CLOCK_MONOTONIC));
 111   }
 112
 113   void SendRpc(int num_rpcs, bool check_deadline, gpr_timespec deadline) {
 114     const std::string kMethodName("/grpc.cpp.test.util.EchoTestService/Echo");
 115     for (int i = 0; i < num_rpcs; i++) {
 116       EchoRequest send_request;
 117       EchoRequest recv_request;
 118       EchoResponse send_response;
 119       EchoResponse recv_response;
 120       Status recv_status;
 121
 122       ClientContext cli_ctx;
 123       GenericServerContext srv_ctx;
 124       GenericServerAsyncReaderWriter stream(&srv_ctx);
 125
 126       // The string needs to be long enough to test heap-based slice.
 127       send_request.set_message("Hello world. Hello world. Hello world.");
 128
 129       if (check_deadline) {
 130         cli_ctx.set_deadline(deadline);
 131       }
 132
 133       // Rather than using the original kMethodName, make a short-lived
 134       // copy to also confirm that we don't refer to this object beyond
 135       // the initial call preparation
 136       const std::string* method_name = new std::string(kMethodName);
 137
 138       std::unique_ptr<GenericClientAsyncReaderWriter> call =
 139           generic_stub_->PrepareCall(&cli_ctx, *method_name, &cli_cq_);
 140
 141       delete method_name;  // Make sure that this is not needed after invocation
 142
 143       std::thread request_call([this]() { server_ok(4); });
 144       call->StartCall(tag(1));
 145       client_ok(1);
 146       std::unique_ptr<ByteBuffer> send_buffer =
 147           SerializeToByteBuffer(&send_request);
 148       call->Write(*send_buffer, tag(2));
 149       // Send ByteBuffer can be destroyed after calling Write.
 150       send_buffer.reset();
 151       client_ok(2);
 152       call->WritesDone(tag(3));
 153       client_ok(3);
 154
 155       generic_service_.RequestCall(&srv_ctx, &stream, srv_cq_.get(),
 156                                    srv_cq_.get(), tag(4));
 157
 158       request_call.join();
 159       EXPECT_EQ(server_host_, srv_ctx.host().substr(0, server_host_.length()));
 160       EXPECT_EQ(kMethodName, srv_ctx.method());
 161
 162       if (check_deadline) {
 163         EXPECT_TRUE(gpr_time_similar(deadline, srv_ctx.raw_deadline(),
 164                                      gpr_time_from_millis(1000, GPR_TIMESPAN)));
 165       }
 166
 167       ByteBuffer recv_buffer;
 168       stream.Read(&recv_buffer, tag(5));
 169       server_ok(5);
 170       EXPECT_TRUE(ParseFromByteBuffer(&recv_buffer, &recv_request));
 171       EXPECT_EQ(send_request.message(), recv_request.message());
 172
 173       send_response.set_message(recv_request.message());
 174       send_buffer = SerializeToByteBuffer(&send_response);
 175       stream.Write(*send_buffer, tag(6));
 176       send_buffer.reset();
 177       server_ok(6);
 178
 179       stream.Finish(Status::OK, tag(7));
 180       server_ok(7);
 181
 182       recv_buffer.Clear();
 183       call->Read(&recv_buffer, tag(8));
 184       client_ok(8);
 185       EXPECT_TRUE(ParseFromByteBuffer(&recv_buffer, &recv_response));
 186
 187       call->Finish(&recv_status, tag(9));
 188       client_ok(9);
 189
 190       EXPECT_EQ(send_response.message(), recv_response.message());
 191       EXPECT_TRUE(recv_status.ok());
 192     }
 193   }
 194
 195   // Return errors to up to one call that comes in on the supplied completion
 196   // queue, until the CQ is being shut down (and therefore we can no longer
 197   // enqueue further events).
 198   void DriveCompletionQueue() {
 199     enum class Event : uintptr_t {
 200       kCallReceived,
 201       kResponseSent,
 202     };
 203     // Request the call, but only if the main thread hasn't beaten us to
 204     // shutting down the CQ.
 205     grpc::GenericServerContext server_context;
 206     grpc::GenericServerAsyncReaderWriter reader_writer(&server_context);
 207
 208     {
 209       std::lock_guard<std::mutex> lock(shutting_down_mu_);
 210       if (!shutting_down_) {
 211         generic_service_.RequestCall(
 212             &server_context, &reader_writer, srv_cq_.get(), srv_cq_.get(),
 213             reinterpret_cast<void*>(Event::kCallReceived));
 214       }
 215     }
 216     // Process events.
 217     {
 218       Event event;
 219       bool ok;
 220       while (srv_cq_->Next(reinterpret_cast<void**>(&event), &ok)) {
 221         std::lock_guard<std::mutex> lock(shutting_down_mu_);
 222         if (shutting_down_) {
 223           // The main thread has started shutting down. Simply continue to drain
 224           // events.
 225           continue;
 226         }
 227
 228         switch (event) {
 229           case Event::kCallReceived:
 230             reader_writer.Finish(
 231                 ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "go away"),
 232                 reinterpret_cast<void*>(Event::kResponseSent));
 233             break;
 234
 235           case Event::kResponseSent:
 236             // We are done.
 237             break;
 238         }
 239       }
 240     }
 241   }
 242
 243   CompletionQueue cli_cq_;
 244   std::unique_ptr<ServerCompletionQueue> srv_cq_;
 245   std::unique_ptr<grpc::testing::EchoTestService::Stub> stub_;
 246   std::unique_ptr<grpc::GenericStub> generic_stub_;
 247   std::unique_ptr<Server> server_;
 248   AsyncGenericService generic_service_;
 249   const std::string server_host_;
 250   std::ostringstream server_address_;
 251   bool shutting_down_;
 252   bool shut_down_;
 253   std::mutex shutting_down_mu_;
 254 };
 255
 256 TEST_F(GenericEnd2endTest, SimpleRpc) {
 257   ResetStub();
 258   SendRpc(1);
 259 }
 260
 261 TEST_F(GenericEnd2endTest, SequentialRpcs) {
 262   ResetStub();
 263   SendRpc(10);
 264 }
 265
 266 TEST_F(GenericEnd2endTest, SequentialUnaryRpcs) {
 267   ResetStub();
 268   const int num_rpcs = 10;
 269   const std::string kMethodName("/grpc.cpp.test.util.EchoTestService/Echo");
 270   for (int i = 0; i < num_rpcs; i++) {
 271     EchoRequest send_request;
 272     EchoRequest recv_request;
 273     EchoResponse send_response;
 274     EchoResponse recv_response;
 275     Status recv_status;
 276
 277     ClientContext cli_ctx;
 278     GenericServerContext srv_ctx;
 279     GenericServerAsyncReaderWriter stream(&srv_ctx);
 280
 281     // The string needs to be long enough to test heap-based slice.
 282     send_request.set_message("Hello world. Hello world. Hello world.");
 283
 284     std::unique_ptr<ByteBuffer> cli_send_buffer =
 285         SerializeToByteBuffer(&send_request);
 286     std::thread request_call([this]() { server_ok(4); });
 287     std::unique_ptr<GenericClientAsyncResponseReader> call =
 288         generic_stub_->PrepareUnaryCall(&cli_ctx, kMethodName,
 289                                         *cli_send_buffer.get(), &cli_cq_);
 290     call->StartCall();
 291     ByteBuffer cli_recv_buffer;
 292     call->Finish(&cli_recv_buffer, &recv_status, tag(1));
 293     std::thread client_check([this] { client_ok(1); });
 294
 295     generic_service_.RequestCall(&srv_ctx, &stream, srv_cq_.get(),
 296                                  srv_cq_.get(), tag(4));
 297     request_call.join();
 298     EXPECT_EQ(server_host_, srv_ctx.host().substr(0, server_host_.length()));
 299     EXPECT_EQ(kMethodName, srv_ctx.method());
 300
 301     ByteBuffer srv_recv_buffer;
 302     stream.Read(&srv_recv_buffer, tag(5));
 303     server_ok(5);
 304     EXPECT_TRUE(ParseFromByteBuffer(&srv_recv_buffer, &recv_request));
 305     EXPECT_EQ(send_request.message(), recv_request.message());
 306
 307     send_response.set_message(recv_request.message());
 308     std::unique_ptr<ByteBuffer> srv_send_buffer =
 309         SerializeToByteBuffer(&send_response);
 310     stream.Write(*srv_send_buffer, tag(6));
 311     server_ok(6);
 312
 313     stream.Finish(Status::OK, tag(7));
 314     server_ok(7);
 315
 316     client_check.join();
 317     EXPECT_TRUE(ParseFromByteBuffer(&cli_recv_buffer, &recv_response));
 318     EXPECT_EQ(send_response.message(), recv_response.message());
 319     EXPECT_TRUE(recv_status.ok());
 320   }
 321 }
 322
 323 // One ping, one pong.
 324 TEST_F(GenericEnd2endTest, SimpleBidiStreaming) {
 325   ResetStub();
 326
 327   const std::string kMethodName(
 328       "/grpc.cpp.test.util.EchoTestService/BidiStream");
 329   EchoRequest send_request;
 330   EchoRequest recv_request;
 331   EchoResponse send_response;
 332   EchoResponse recv_response;
 333   Status recv_status;
 334   ClientContext cli_ctx;
 335   GenericServerContext srv_ctx;
 336   GenericServerAsyncReaderWriter srv_stream(&srv_ctx);
 337
 338   cli_ctx.set_compression_algorithm(GRPC_COMPRESS_GZIP);
 339   send_request.set_message("Hello");
 340   std::thread request_call([this]() { server_ok(2); });
 341   std::unique_ptr<GenericClientAsyncReaderWriter> cli_stream =
 342       generic_stub_->PrepareCall(&cli_ctx, kMethodName, &cli_cq_);
 343   cli_stream->StartCall(tag(1));
 344   client_ok(1);
 345
 346   generic_service_.RequestCall(&srv_ctx, &srv_stream, srv_cq_.get(),
 347                                srv_cq_.get(), tag(2));
 348   request_call.join();
 349
 350   EXPECT_EQ(server_host_, srv_ctx.host().substr(0, server_host_.length()));
 351   EXPECT_EQ(kMethodName, srv_ctx.method());
 352
 353   std::unique_ptr<ByteBuffer> send_buffer =
 354       SerializeToByteBuffer(&send_request);
 355   cli_stream->Write(*send_buffer, tag(3));
 356   send_buffer.reset();
 357   client_ok(3);
 358
 359   ByteBuffer recv_buffer;
 360   srv_stream.Read(&recv_buffer, tag(4));
 361   server_ok(4);
 362   EXPECT_TRUE(ParseFromByteBuffer(&recv_buffer, &recv_request));
 363   EXPECT_EQ(send_request.message(), recv_request.message());
 364
 365   send_response.set_message(recv_request.message());
 366   send_buffer = SerializeToByteBuffer(&send_response);
 367   srv_stream.Write(*send_buffer, tag(5));
 368   send_buffer.reset();
 369   server_ok(5);
 370
 371   cli_stream->Read(&recv_buffer, tag(6));
 372   client_ok(6);
 373   EXPECT_TRUE(ParseFromByteBuffer(&recv_buffer, &recv_response));
 374   EXPECT_EQ(send_response.message(), recv_response.message());
 375
 376   cli_stream->WritesDone(tag(7));
 377   client_ok(7);
 378
 379   srv_stream.Read(&recv_buffer, tag(8));
 380   server_fail(8);
 381
 382   srv_stream.Finish(Status::OK, tag(9));
 383   server_ok(9);
 384
 385   cli_stream->Finish(&recv_status, tag(10));
 386   client_ok(10);
 387
 388   EXPECT_EQ(send_response.message(), recv_response.message());
 389   EXPECT_TRUE(recv_status.ok());
 390 }
 391
 392 TEST_F(GenericEnd2endTest, Deadline) {
 393   ResetStub();
 394   SendRpc(1, true,
 395           gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC),
 396                        gpr_time_from_seconds(10, GPR_TIMESPAN)));
 397 }
 398
 399 TEST_F(GenericEnd2endTest, ShortDeadline) {
 400   ResetStub();
 401
 402   ClientContext cli_ctx;
 403   EchoRequest request;
 404   EchoResponse response;
 405
 406   shutting_down_ = false;
 407   std::thread driver([this] { DriveCompletionQueue(); });
 408
 409   request.set_message("");
 410   cli_ctx.set_deadline(gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC),
 411                                     gpr_time_from_micros(500, GPR_TIMESPAN)));
 412   Status s = stub_->Echo(&cli_ctx, request, &response);
 413   EXPECT_FALSE(s.ok());
 414   {
 415     std::lock_guard<std::mutex> lock(shutting_down_mu_);
 416     shutting_down_ = true;
 417   }
 418   ShutDownServerAndCQs();
 419   driver.join();
 420 }
 421
 422 }  // namespace
 423 }  // namespace testing
 424 }  // namespace grpc
 425
 426 int main(int argc, char** argv) {
 427   grpc::testing::TestEnvironment env(argc, argv);
 428   ::testing::InitGoogleTest(&argc, argv);
 429   return RUN_ALL_TESTS();
 430 }