test/cpp/end2end/server_interceptors_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 <memory>
  20 #include <vector>
  21
  22 #include <grpcpp/channel.h>
  23 #include <grpcpp/client_context.h>
  24 #include <grpcpp/create_channel.h>
  25 #include <grpcpp/generic/generic_stub.h>
  26 #include <grpcpp/impl/codegen/proto_utils.h>
  27 #include <grpcpp/server.h>
  28 #include <grpcpp/server_builder.h>
  29 #include <grpcpp/server_context.h>
  30 #include <grpcpp/support/server_interceptor.h>
  31
  32 #include "src/proto/grpc/testing/echo.grpc.pb.h"
  33 #include "test/core/util/port.h"
  34 #include "test/core/util/test_config.h"
  35 #include "test/cpp/end2end/interceptors_util.h"
  36 #include "test/cpp/end2end/test_service_impl.h"
  37 #include "test/cpp/util/byte_buffer_proto_helper.h"
  38
  39 #include <gtest/gtest.h>
  40
  41 namespace grpc {
  42 namespace testing {
  43 namespace {
  44
  45 class LoggingInterceptor : public experimental::Interceptor {
  46  public:
  47   LoggingInterceptor(experimental::ServerRpcInfo* info) {
  48     info_ = info;
  49
  50     // Check the method name and compare to the type
  51     const char* method = info->method();
  52     experimental::ServerRpcInfo::Type type = info->type();
  53
  54     // Check that we use one of our standard methods with expected type.
  55     // Also allow the health checking service.
  56     // We accept BIDI_STREAMING for Echo in case it's an AsyncGenericService
  57     // being tested (the GenericRpc test).
  58     // The empty method is for the Unimplemented requests that arise
  59     // when draining the CQ.
  60     EXPECT_TRUE(
  61         strstr(method, "/grpc.health") == method ||
  62         (strcmp(method, "/grpc.testing.EchoTestService/Echo") == 0 &&
  63          (type == experimental::ServerRpcInfo::Type::UNARY ||
  64           type == experimental::ServerRpcInfo::Type::BIDI_STREAMING)) ||
  65         (strcmp(method, "/grpc.testing.EchoTestService/RequestStream") == 0 &&
  66          type == experimental::ServerRpcInfo::Type::CLIENT_STREAMING) ||
  67         (strcmp(method, "/grpc.testing.EchoTestService/ResponseStream") == 0 &&
  68          type == experimental::ServerRpcInfo::Type::SERVER_STREAMING) ||
  69         (strcmp(method, "/grpc.testing.EchoTestService/BidiStream") == 0 &&
  70          type == experimental::ServerRpcInfo::Type::BIDI_STREAMING) ||
  71         strcmp(method, "/grpc.testing.EchoTestService/Unimplemented") == 0 ||
  72         (strcmp(method, "") == 0 &&
  73          type == experimental::ServerRpcInfo::Type::BIDI_STREAMING));
  74   }
  75
  76   void Intercept(experimental::InterceptorBatchMethods* methods) override {
  77     if (methods->QueryInterceptionHookPoint(
  78             experimental::InterceptionHookPoints::PRE_SEND_INITIAL_METADATA)) {
  79       auto* map = methods->GetSendInitialMetadata();
  80       // Got nothing better to do here for now
  81       EXPECT_EQ(map->size(), static_cast<unsigned>(0));
  82     }
  83     if (methods->QueryInterceptionHookPoint(
  84             experimental::InterceptionHookPoints::PRE_SEND_MESSAGE)) {
  85       EchoRequest req;
  86       auto* buffer = methods->GetSerializedSendMessage();
  87       auto copied_buffer = *buffer;
  88       EXPECT_TRUE(
  89           SerializationTraits<EchoRequest>::Deserialize(&copied_buffer, &req)
  90               .ok());
  91       EXPECT_TRUE(req.message().find("Hello") == 0);
  92     }
  93     if (methods->QueryInterceptionHookPoint(
  94             experimental::InterceptionHookPoints::PRE_SEND_STATUS)) {
  95       auto* map = methods->GetSendTrailingMetadata();
  96       bool found = false;
  97       // Check that we received the metadata as an echo
  98       for (const auto& pair : *map) {
  99         found = pair.first.find("testkey") == 0 &&
 100                 pair.second.find("testvalue") == 0;
 101         if (found) break;
 102       }
 103       EXPECT_EQ(found, true);
 104       auto status = methods->GetSendStatus();
 105       EXPECT_EQ(status.ok(), true);
 106     }
 107     if (methods->QueryInterceptionHookPoint(
 108             experimental::InterceptionHookPoints::POST_RECV_INITIAL_METADATA)) {
 109       auto* map = methods->GetRecvInitialMetadata();
 110       bool found = false;
 111       // Check that we received the metadata as an echo
 112       for (const auto& pair : *map) {
 113         found = pair.first.find("testkey") == 0 &&
 114                 pair.second.find("testvalue") == 0;
 115         if (found) break;
 116       }
 117       EXPECT_EQ(found, true);
 118     }
 119     if (methods->QueryInterceptionHookPoint(
 120             experimental::InterceptionHookPoints::POST_RECV_MESSAGE)) {
 121       EchoResponse* resp =
 122           static_cast<EchoResponse*>(methods->GetRecvMessage());
 123       if (resp != nullptr) {
 124         EXPECT_TRUE(resp->message().find("Hello") == 0);
 125       }
 126     }
 127     if (methods->QueryInterceptionHookPoint(
 128             experimental::InterceptionHookPoints::POST_RECV_CLOSE)) {
 129       // Got nothing interesting to do here
 130     }
 131     methods->Proceed();
 132   }
 133
 134  private:
 135   experimental::ServerRpcInfo* info_;
 136 };
 137
 138 class LoggingInterceptorFactory
 139     : public experimental::ServerInterceptorFactoryInterface {
 140  public:
 141   virtual experimental::Interceptor* CreateServerInterceptor(
 142       experimental::ServerRpcInfo* info) override {
 143     return new LoggingInterceptor(info);
 144   }
 145 };
 146
 147 // Test if SendMessage function family works as expected for sync/callback apis
 148 class SyncSendMessageTester : public experimental::Interceptor {
 149  public:
 150   SyncSendMessageTester(experimental::ServerRpcInfo* /*info*/) {}
 151
 152   void Intercept(experimental::InterceptorBatchMethods* methods) override {
 153     if (methods->QueryInterceptionHookPoint(
 154             experimental::InterceptionHookPoints::PRE_SEND_MESSAGE)) {
 155       string old_msg =
 156           static_cast<const EchoRequest*>(methods->GetSendMessage())->message();
 157       EXPECT_EQ(old_msg.find("Hello"), 0u);
 158       new_msg_.set_message("World" + old_msg);
 159       methods->ModifySendMessage(&new_msg_);
 160     }
 161     methods->Proceed();
 162   }
 163
 164  private:
 165   EchoRequest new_msg_;
 166 };
 167
 168 class SyncSendMessageTesterFactory
 169     : public experimental::ServerInterceptorFactoryInterface {
 170  public:
 171   virtual experimental::Interceptor* CreateServerInterceptor(
 172       experimental::ServerRpcInfo* info) override {
 173     return new SyncSendMessageTester(info);
 174   }
 175 };
 176
 177 // Test if SendMessage function family works as expected for sync/callback apis
 178 class SyncSendMessageVerifier : public experimental::Interceptor {
 179  public:
 180   SyncSendMessageVerifier(experimental::ServerRpcInfo* /*info*/) {}
 181
 182   void Intercept(experimental::InterceptorBatchMethods* methods) override {
 183     if (methods->QueryInterceptionHookPoint(
 184             experimental::InterceptionHookPoints::PRE_SEND_MESSAGE)) {
 185       // Make sure that the changes made in SyncSendMessageTester persisted
 186       string old_msg =
 187           static_cast<const EchoRequest*>(methods->GetSendMessage())->message();
 188       EXPECT_EQ(old_msg.find("World"), 0u);
 189
 190       // Remove the "World" part of the string that we added earlier
 191       new_msg_.set_message(old_msg.erase(0, 5));
 192       methods->ModifySendMessage(&new_msg_);
 193
 194       // LoggingInterceptor verifies that changes got reverted
 195     }
 196     methods->Proceed();
 197   }
 198
 199  private:
 200   EchoRequest new_msg_;
 201 };
 202
 203 class SyncSendMessageVerifierFactory
 204     : public experimental::ServerInterceptorFactoryInterface {
 205  public:
 206   virtual experimental::Interceptor* CreateServerInterceptor(
 207       experimental::ServerRpcInfo* info) override {
 208     return new SyncSendMessageVerifier(info);
 209   }
 210 };
 211
 212 void MakeBidiStreamingCall(const std::shared_ptr<Channel>& channel) {
 213   auto stub = grpc::testing::EchoTestService::NewStub(channel);
 214   ClientContext ctx;
 215   EchoRequest req;
 216   EchoResponse resp;
 217   ctx.AddMetadata("testkey", "testvalue");
 218   auto stream = stub->BidiStream(&ctx);
 219   for (auto i = 0; i < 10; i++) {
 220     req.set_message("Hello" + std::to_string(i));
 221     stream->Write(req);
 222     stream->Read(&resp);
 223     EXPECT_EQ(req.message(), resp.message());
 224   }
 225   ASSERT_TRUE(stream->WritesDone());
 226   Status s = stream->Finish();
 227   EXPECT_EQ(s.ok(), true);
 228 }
 229
 230 class ServerInterceptorsEnd2endSyncUnaryTest : public ::testing::Test {
 231  protected:
 232   ServerInterceptorsEnd2endSyncUnaryTest() {
 233     int port = grpc_pick_unused_port_or_die();
 234
 235     ServerBuilder builder;
 236     server_address_ = "localhost:" + std::to_string(port);
 237     builder.AddListeningPort(server_address_, InsecureServerCredentials());
 238     builder.RegisterService(&service_);
 239
 240     std::vector<
 241         std::unique_ptr<experimental::ServerInterceptorFactoryInterface>>
 242         creators;
 243     creators.push_back(
 244         std::unique_ptr<experimental::ServerInterceptorFactoryInterface>(
 245             new SyncSendMessageTesterFactory()));
 246     creators.push_back(
 247         std::unique_ptr<experimental::ServerInterceptorFactoryInterface>(
 248             new SyncSendMessageVerifierFactory()));
 249     creators.push_back(
 250         std::unique_ptr<experimental::ServerInterceptorFactoryInterface>(
 251             new LoggingInterceptorFactory()));
 252     // Add 20 dummy interceptor factories and null interceptor factories
 253     for (auto i = 0; i < 20; i++) {
 254       creators.push_back(std::unique_ptr<DummyInterceptorFactory>(
 255           new DummyInterceptorFactory()));
 256       creators.push_back(std::unique_ptr<NullInterceptorFactory>(
 257           new NullInterceptorFactory()));
 258     }
 259     builder.experimental().SetInterceptorCreators(std::move(creators));
 260     server_ = builder.BuildAndStart();
 261   }
 262   std::string server_address_;
 263   TestServiceImpl service_;
 264   std::unique_ptr<Server> server_;
 265 };
 266
 267 TEST_F(ServerInterceptorsEnd2endSyncUnaryTest, UnaryTest) {
 268   ChannelArguments args;
 269   DummyInterceptor::Reset();
 270   auto channel =
 271       grpc::CreateChannel(server_address_, InsecureChannelCredentials());
 272   MakeCall(channel);
 273   // Make sure all 20 dummy interceptors were run
 274   EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20);
 275 }
 276
 277 class ServerInterceptorsEnd2endSyncStreamingTest : public ::testing::Test {
 278  protected:
 279   ServerInterceptorsEnd2endSyncStreamingTest() {
 280     int port = grpc_pick_unused_port_or_die();
 281
 282     ServerBuilder builder;
 283     server_address_ = "localhost:" + std::to_string(port);
 284     builder.AddListeningPort(server_address_, InsecureServerCredentials());
 285     builder.RegisterService(&service_);
 286
 287     std::vector<
 288         std::unique_ptr<experimental::ServerInterceptorFactoryInterface>>
 289         creators;
 290     creators.push_back(
 291         std::unique_ptr<experimental::ServerInterceptorFactoryInterface>(
 292             new SyncSendMessageTesterFactory()));
 293     creators.push_back(
 294         std::unique_ptr<experimental::ServerInterceptorFactoryInterface>(
 295             new SyncSendMessageVerifierFactory()));
 296     creators.push_back(
 297         std::unique_ptr<experimental::ServerInterceptorFactoryInterface>(
 298             new LoggingInterceptorFactory()));
 299     for (auto i = 0; i < 20; i++) {
 300       creators.push_back(std::unique_ptr<DummyInterceptorFactory>(
 301           new DummyInterceptorFactory()));
 302     }
 303     builder.experimental().SetInterceptorCreators(std::move(creators));
 304     server_ = builder.BuildAndStart();
 305   }
 306   std::string server_address_;
 307   EchoTestServiceStreamingImpl service_;
 308   std::unique_ptr<Server> server_;
 309 };
 310
 311 TEST_F(ServerInterceptorsEnd2endSyncStreamingTest, ClientStreamingTest) {
 312   ChannelArguments args;
 313   DummyInterceptor::Reset();
 314   auto channel =
 315       grpc::CreateChannel(server_address_, InsecureChannelCredentials());
 316   MakeClientStreamingCall(channel);
 317   // Make sure all 20 dummy interceptors were run
 318   EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20);
 319 }
 320
 321 TEST_F(ServerInterceptorsEnd2endSyncStreamingTest, ServerStreamingTest) {
 322   ChannelArguments args;
 323   DummyInterceptor::Reset();
 324   auto channel =
 325       grpc::CreateChannel(server_address_, InsecureChannelCredentials());
 326   MakeServerStreamingCall(channel);
 327   // Make sure all 20 dummy interceptors were run
 328   EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20);
 329 }
 330
 331 TEST_F(ServerInterceptorsEnd2endSyncStreamingTest, BidiStreamingTest) {
 332   ChannelArguments args;
 333   DummyInterceptor::Reset();
 334   auto channel =
 335       grpc::CreateChannel(server_address_, InsecureChannelCredentials());
 336   MakeBidiStreamingCall(channel);
 337   // Make sure all 20 dummy interceptors were run
 338   EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20);
 339 }
 340
 341 class ServerInterceptorsAsyncEnd2endTest : public ::testing::Test {};
 342
 343 TEST_F(ServerInterceptorsAsyncEnd2endTest, UnaryTest) {
 344   DummyInterceptor::Reset();
 345   int port = grpc_pick_unused_port_or_die();
 346   string server_address = "localhost:" + std::to_string(port);
 347   ServerBuilder builder;
 348   EchoTestService::AsyncService service;
 349   builder.AddListeningPort(server_address, InsecureServerCredentials());
 350   builder.RegisterService(&service);
 351   std::vector<std::unique_ptr<experimental::ServerInterceptorFactoryInterface>>
 352       creators;
 353   creators.push_back(
 354       std::unique_ptr<experimental::ServerInterceptorFactoryInterface>(
 355           new LoggingInterceptorFactory()));
 356   for (auto i = 0; i < 20; i++) {
 357     creators.push_back(std::unique_ptr<DummyInterceptorFactory>(
 358         new DummyInterceptorFactory()));
 359   }
 360   builder.experimental().SetInterceptorCreators(std::move(creators));
 361   auto cq = builder.AddCompletionQueue();
 362   auto server = builder.BuildAndStart();
 363
 364   ChannelArguments args;
 365   auto channel =
 366       grpc::CreateChannel(server_address, InsecureChannelCredentials());
 367   auto stub = grpc::testing::EchoTestService::NewStub(channel);
 368
 369   EchoRequest send_request;
 370   EchoRequest recv_request;
 371   EchoResponse send_response;
 372   EchoResponse recv_response;
 373   Status recv_status;
 374
 375   ClientContext cli_ctx;
 376   ServerContext srv_ctx;
 377   grpc::ServerAsyncResponseWriter<EchoResponse> response_writer(&srv_ctx);
 378
 379   send_request.set_message("Hello");
 380   cli_ctx.AddMetadata("testkey", "testvalue");
 381   std::unique_ptr<ClientAsyncResponseReader<EchoResponse>> response_reader(
 382       stub->AsyncEcho(&cli_ctx, send_request, cq.get()));
 383
 384   service.RequestEcho(&srv_ctx, &recv_request, &response_writer, cq.get(),
 385                       cq.get(), tag(2));
 386
 387   response_reader->Finish(&recv_response, &recv_status, tag(4));
 388
 389   Verifier().Expect(2, true).Verify(cq.get());
 390   EXPECT_EQ(send_request.message(), recv_request.message());
 391
 392   EXPECT_TRUE(CheckMetadata(srv_ctx.client_metadata(), "testkey", "testvalue"));
 393   srv_ctx.AddTrailingMetadata("testkey", "testvalue");
 394
 395   send_response.set_message(recv_request.message());
 396   response_writer.Finish(send_response, Status::OK, tag(3));
 397   Verifier().Expect(3, true).Expect(4, true).Verify(cq.get());
 398
 399   EXPECT_EQ(send_response.message(), recv_response.message());
 400   EXPECT_TRUE(recv_status.ok());
 401   EXPECT_TRUE(CheckMetadata(cli_ctx.GetServerTrailingMetadata(), "testkey",
 402                             "testvalue"));
 403
 404   // Make sure all 20 dummy interceptors were run
 405   EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20);
 406
 407   server->Shutdown();
 408   cq->Shutdown();
 409   void* ignored_tag;
 410   bool ignored_ok;
 411   while (cq->Next(&ignored_tag, &ignored_ok))
 412     ;
 413   grpc_recycle_unused_port(port);
 414 }
 415
 416 TEST_F(ServerInterceptorsAsyncEnd2endTest, BidiStreamingTest) {
 417   DummyInterceptor::Reset();
 418   int port = grpc_pick_unused_port_or_die();
 419   string server_address = "localhost:" + std::to_string(port);
 420   ServerBuilder builder;
 421   EchoTestService::AsyncService service;
 422   builder.AddListeningPort(server_address, InsecureServerCredentials());
 423   builder.RegisterService(&service);
 424   std::vector<std::unique_ptr<experimental::ServerInterceptorFactoryInterface>>
 425       creators;
 426   creators.push_back(
 427       std::unique_ptr<experimental::ServerInterceptorFactoryInterface>(
 428           new LoggingInterceptorFactory()));
 429   for (auto i = 0; i < 20; i++) {
 430     creators.push_back(std::unique_ptr<DummyInterceptorFactory>(
 431         new DummyInterceptorFactory()));
 432   }
 433   builder.experimental().SetInterceptorCreators(std::move(creators));
 434   auto cq = builder.AddCompletionQueue();
 435   auto server = builder.BuildAndStart();
 436
 437   ChannelArguments args;
 438   auto channel =
 439       grpc::CreateChannel(server_address, InsecureChannelCredentials());
 440   auto stub = grpc::testing::EchoTestService::NewStub(channel);
 441
 442   EchoRequest send_request;
 443   EchoRequest recv_request;
 444   EchoResponse send_response;
 445   EchoResponse recv_response;
 446   Status recv_status;
 447
 448   ClientContext cli_ctx;
 449   ServerContext srv_ctx;
 450   grpc::ServerAsyncReaderWriter<EchoResponse, EchoRequest> srv_stream(&srv_ctx);
 451
 452   send_request.set_message("Hello");
 453   cli_ctx.AddMetadata("testkey", "testvalue");
 454   std::unique_ptr<ClientAsyncReaderWriter<EchoRequest, EchoResponse>>
 455       cli_stream(stub->AsyncBidiStream(&cli_ctx, cq.get(), tag(1)));
 456
 457   service.RequestBidiStream(&srv_ctx, &srv_stream, cq.get(), cq.get(), tag(2));
 458
 459   Verifier().Expect(1, true).Expect(2, true).Verify(cq.get());
 460
 461   EXPECT_TRUE(CheckMetadata(srv_ctx.client_metadata(), "testkey", "testvalue"));
 462   srv_ctx.AddTrailingMetadata("testkey", "testvalue");
 463
 464   cli_stream->Write(send_request, tag(3));
 465   srv_stream.Read(&recv_request, tag(4));
 466   Verifier().Expect(3, true).Expect(4, true).Verify(cq.get());
 467   EXPECT_EQ(send_request.message(), recv_request.message());
 468
 469   send_response.set_message(recv_request.message());
 470   srv_stream.Write(send_response, tag(5));
 471   cli_stream->Read(&recv_response, tag(6));
 472   Verifier().Expect(5, true).Expect(6, true).Verify(cq.get());
 473   EXPECT_EQ(send_response.message(), recv_response.message());
 474
 475   cli_stream->WritesDone(tag(7));
 476   srv_stream.Read(&recv_request, tag(8));
 477   Verifier().Expect(7, true).Expect(8, false).Verify(cq.get());
 478
 479   srv_stream.Finish(Status::OK, tag(9));
 480   cli_stream->Finish(&recv_status, tag(10));
 481   Verifier().Expect(9, true).Expect(10, true).Verify(cq.get());
 482
 483   EXPECT_TRUE(recv_status.ok());
 484   EXPECT_TRUE(CheckMetadata(cli_ctx.GetServerTrailingMetadata(), "testkey",
 485                             "testvalue"));
 486
 487   // Make sure all 20 dummy interceptors were run
 488   EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20);
 489
 490   server->Shutdown();
 491   cq->Shutdown();
 492   void* ignored_tag;
 493   bool ignored_ok;
 494   while (cq->Next(&ignored_tag, &ignored_ok))
 495     ;
 496   grpc_recycle_unused_port(port);
 497 }
 498
 499 TEST_F(ServerInterceptorsAsyncEnd2endTest, GenericRPCTest) {
 500   DummyInterceptor::Reset();
 501   int port = grpc_pick_unused_port_or_die();
 502   string server_address = "localhost:" + std::to_string(port);
 503   ServerBuilder builder;
 504   AsyncGenericService service;
 505   builder.AddListeningPort(server_address, InsecureServerCredentials());
 506   builder.RegisterAsyncGenericService(&service);
 507   std::vector<std::unique_ptr<experimental::ServerInterceptorFactoryInterface>>
 508       creators;
 509   creators.reserve(20);
 510   for (auto i = 0; i < 20; i++) {
 511     creators.push_back(std::unique_ptr<DummyInterceptorFactory>(
 512         new DummyInterceptorFactory()));
 513   }
 514   builder.experimental().SetInterceptorCreators(std::move(creators));
 515   auto srv_cq = builder.AddCompletionQueue();
 516   CompletionQueue cli_cq;
 517   auto server = builder.BuildAndStart();
 518
 519   ChannelArguments args;
 520   auto channel =
 521       grpc::CreateChannel(server_address, InsecureChannelCredentials());
 522   GenericStub generic_stub(channel);
 523
 524   const std::string kMethodName("/grpc.cpp.test.util.EchoTestService/Echo");
 525   EchoRequest send_request;
 526   EchoRequest recv_request;
 527   EchoResponse send_response;
 528   EchoResponse recv_response;
 529   Status recv_status;
 530
 531   ClientContext cli_ctx;
 532   GenericServerContext srv_ctx;
 533   GenericServerAsyncReaderWriter stream(&srv_ctx);
 534
 535   // The string needs to be long enough to test heap-based slice.
 536   send_request.set_message("Hello");
 537   cli_ctx.AddMetadata("testkey", "testvalue");
 538
 539   CompletionQueue* cq = srv_cq.get();
 540   std::thread request_call([cq]() { Verifier().Expect(4, true).Verify(cq); });
 541   std::unique_ptr<GenericClientAsyncReaderWriter> call =
 542       generic_stub.PrepareCall(&cli_ctx, kMethodName, &cli_cq);
 543   call->StartCall(tag(1));
 544   Verifier().Expect(1, true).Verify(&cli_cq);
 545   std::unique_ptr<ByteBuffer> send_buffer =
 546       SerializeToByteBuffer(&send_request);
 547   call->Write(*send_buffer, tag(2));
 548   // Send ByteBuffer can be destroyed after calling Write.
 549   send_buffer.reset();
 550   Verifier().Expect(2, true).Verify(&cli_cq);
 551   call->WritesDone(tag(3));
 552   Verifier().Expect(3, true).Verify(&cli_cq);
 553
 554   service.RequestCall(&srv_ctx, &stream, srv_cq.get(), srv_cq.get(), tag(4));
 555
 556   request_call.join();
 557   EXPECT_EQ(kMethodName, srv_ctx.method());
 558   EXPECT_TRUE(CheckMetadata(srv_ctx.client_metadata(), "testkey", "testvalue"));
 559   srv_ctx.AddTrailingMetadata("testkey", "testvalue");
 560
 561   ByteBuffer recv_buffer;
 562   stream.Read(&recv_buffer, tag(5));
 563   Verifier().Expect(5, true).Verify(srv_cq.get());
 564   EXPECT_TRUE(ParseFromByteBuffer(&recv_buffer, &recv_request));
 565   EXPECT_EQ(send_request.message(), recv_request.message());
 566
 567   send_response.set_message(recv_request.message());
 568   send_buffer = SerializeToByteBuffer(&send_response);
 569   stream.Write(*send_buffer, tag(6));
 570   send_buffer.reset();
 571   Verifier().Expect(6, true).Verify(srv_cq.get());
 572
 573   stream.Finish(Status::OK, tag(7));
 574   // Shutdown srv_cq before we try to get the tag back, to verify that the
 575   // interception API handles completion queue shutdowns that take place before
 576   // all the tags are returned
 577   srv_cq->Shutdown();
 578   Verifier().Expect(7, true).Verify(srv_cq.get());
 579
 580   recv_buffer.Clear();
 581   call->Read(&recv_buffer, tag(8));
 582   Verifier().Expect(8, true).Verify(&cli_cq);
 583   EXPECT_TRUE(ParseFromByteBuffer(&recv_buffer, &recv_response));
 584
 585   call->Finish(&recv_status, tag(9));
 586   cli_cq.Shutdown();
 587   Verifier().Expect(9, true).Verify(&cli_cq);
 588
 589   EXPECT_EQ(send_response.message(), recv_response.message());
 590   EXPECT_TRUE(recv_status.ok());
 591   EXPECT_TRUE(CheckMetadata(cli_ctx.GetServerTrailingMetadata(), "testkey",
 592                             "testvalue"));
 593
 594   // Make sure all 20 dummy interceptors were run
 595   EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20);
 596
 597   server->Shutdown();
 598   void* ignored_tag;
 599   bool ignored_ok;
 600   while (cli_cq.Next(&ignored_tag, &ignored_ok))
 601     ;
 602   while (srv_cq->Next(&ignored_tag, &ignored_ok))
 603     ;
 604   grpc_recycle_unused_port(port);
 605 }
 606
 607 TEST_F(ServerInterceptorsAsyncEnd2endTest, UnimplementedRpcTest) {
 608   DummyInterceptor::Reset();
 609   int port = grpc_pick_unused_port_or_die();
 610   string server_address = "localhost:" + std::to_string(port);
 611   ServerBuilder builder;
 612   builder.AddListeningPort(server_address, InsecureServerCredentials());
 613   std::vector<std::unique_ptr<experimental::ServerInterceptorFactoryInterface>>
 614       creators;
 615   creators.reserve(20);
 616   for (auto i = 0; i < 20; i++) {
 617     creators.push_back(std::unique_ptr<DummyInterceptorFactory>(
 618         new DummyInterceptorFactory()));
 619   }
 620   builder.experimental().SetInterceptorCreators(std::move(creators));
 621   auto cq = builder.AddCompletionQueue();
 622   auto server = builder.BuildAndStart();
 623
 624   ChannelArguments args;
 625   std::shared_ptr<Channel> channel =
 626       grpc::CreateChannel(server_address, InsecureChannelCredentials());
 627   std::unique_ptr<grpc::testing::UnimplementedEchoService::Stub> stub;
 628   stub = grpc::testing::UnimplementedEchoService::NewStub(channel);
 629   EchoRequest send_request;
 630   EchoResponse recv_response;
 631   Status recv_status;
 632
 633   ClientContext cli_ctx;
 634   send_request.set_message("Hello");
 635   std::unique_ptr<ClientAsyncResponseReader<EchoResponse>> response_reader(
 636       stub->AsyncUnimplemented(&cli_ctx, send_request, cq.get()));
 637
 638   response_reader->Finish(&recv_response, &recv_status, tag(4));
 639   Verifier().Expect(4, true).Verify(cq.get());
 640
 641   EXPECT_EQ(StatusCode::UNIMPLEMENTED, recv_status.error_code());
 642   EXPECT_EQ("", recv_status.error_message());
 643
 644   // Make sure all 20 dummy interceptors were run
 645   EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20);
 646
 647   server->Shutdown();
 648   cq->Shutdown();
 649   void* ignored_tag;
 650   bool ignored_ok;
 651   while (cq->Next(&ignored_tag, &ignored_ok))
 652     ;
 653   grpc_recycle_unused_port(port);
 654 }
 655
 656 class ServerInterceptorsSyncUnimplementedEnd2endTest : public ::testing::Test {
 657 };
 658
 659 TEST_F(ServerInterceptorsSyncUnimplementedEnd2endTest, UnimplementedRpcTest) {
 660   DummyInterceptor::Reset();
 661   int port = grpc_pick_unused_port_or_die();
 662   string server_address = "localhost:" + std::to_string(port);
 663   ServerBuilder builder;
 664   TestServiceImpl service;
 665   builder.RegisterService(&service);
 666   builder.AddListeningPort(server_address, InsecureServerCredentials());
 667   std::vector<std::unique_ptr<experimental::ServerInterceptorFactoryInterface>>
 668       creators;
 669   creators.reserve(20);
 670   for (auto i = 0; i < 20; i++) {
 671     creators.push_back(std::unique_ptr<DummyInterceptorFactory>(
 672         new DummyInterceptorFactory()));
 673   }
 674   builder.experimental().SetInterceptorCreators(std::move(creators));
 675   auto server = builder.BuildAndStart();
 676
 677   ChannelArguments args;
 678   std::shared_ptr<Channel> channel =
 679       grpc::CreateChannel(server_address, InsecureChannelCredentials());
 680   std::unique_ptr<grpc::testing::UnimplementedEchoService::Stub> stub;
 681   stub = grpc::testing::UnimplementedEchoService::NewStub(channel);
 682   EchoRequest send_request;
 683   EchoResponse recv_response;
 684
 685   ClientContext cli_ctx;
 686   send_request.set_message("Hello");
 687   Status recv_status =
 688       stub->Unimplemented(&cli_ctx, send_request, &recv_response);
 689
 690   EXPECT_EQ(StatusCode::UNIMPLEMENTED, recv_status.error_code());
 691   EXPECT_EQ("", recv_status.error_message());
 692
 693   // Make sure all 20 dummy interceptors were run
 694   EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20);
 695
 696   server->Shutdown();
 697   grpc_recycle_unused_port(port);
 698 }
 699
 700 }  // namespace
 701 }  // namespace testing
 702 }  // namespace grpc
 703
 704 int main(int argc, char** argv) {
 705   grpc::testing::TestEnvironment env(argc, argv);
 706   ::testing::InitGoogleTest(&argc, argv);
 707   return RUN_ALL_TESTS();
 708 }