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:
LoggingInterceptor(experimental::ServerRpcInfo * info)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
Intercept(experimental::InterceptorBatchMethods * methods)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:
CreateServerInterceptor(experimental::ServerRpcInfo * info)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:
SyncSendMessageTester(experimental::ServerRpcInfo *)150 SyncSendMessageTester(experimental::ServerRpcInfo* /*info*/) {}
151
Intercept(experimental::InterceptorBatchMethods * methods)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:
CreateServerInterceptor(experimental::ServerRpcInfo * info)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:
SyncSendMessageVerifier(experimental::ServerRpcInfo *)180 SyncSendMessageVerifier(experimental::ServerRpcInfo* /*info*/) {}
181
Intercept(experimental::InterceptorBatchMethods * methods)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:
CreateServerInterceptor(experimental::ServerRpcInfo * info)206 virtual experimental::Interceptor* CreateServerInterceptor(
207 experimental::ServerRpcInfo* info) override {
208 return new SyncSendMessageVerifier(info);
209 }
210 };
211
MakeBidiStreamingCall(const std::shared_ptr<Channel> & channel)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:
ServerInterceptorsEnd2endSyncUnaryTest()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
TEST_F(ServerInterceptorsEnd2endSyncUnaryTest,UnaryTest)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:
ServerInterceptorsEnd2endSyncStreamingTest()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
TEST_F(ServerInterceptorsEnd2endSyncStreamingTest,ClientStreamingTest)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
TEST_F(ServerInterceptorsEnd2endSyncStreamingTest,ServerStreamingTest)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
TEST_F(ServerInterceptorsEnd2endSyncStreamingTest,BidiStreamingTest)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
TEST_F(ServerInterceptorsAsyncEnd2endTest,UnaryTest)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
TEST_F(ServerInterceptorsAsyncEnd2endTest,BidiStreamingTest)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
TEST_F(ServerInterceptorsAsyncEnd2endTest,GenericRPCTest)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 std::unique_ptr<GenericClientAsyncReaderWriter> call =
540 generic_stub.PrepareCall(&cli_ctx, kMethodName, &cli_cq);
541 call->StartCall(tag(1));
542 Verifier().Expect(1, true).Verify(&cli_cq);
543 std::unique_ptr<ByteBuffer> send_buffer =
544 SerializeToByteBuffer(&send_request);
545 call->Write(*send_buffer, tag(2));
546 // Send ByteBuffer can be destroyed after calling Write.
547 send_buffer.reset();
548 Verifier().Expect(2, true).Verify(&cli_cq);
549 call->WritesDone(tag(3));
550 Verifier().Expect(3, true).Verify(&cli_cq);
551
552 service.RequestCall(&srv_ctx, &stream, srv_cq.get(), srv_cq.get(), tag(4));
553
554 Verifier().Expect(4, true).Verify(srv_cq.get());
555 EXPECT_EQ(kMethodName, srv_ctx.method());
556 EXPECT_TRUE(CheckMetadata(srv_ctx.client_metadata(), "testkey", "testvalue"));
557 srv_ctx.AddTrailingMetadata("testkey", "testvalue");
558
559 ByteBuffer recv_buffer;
560 stream.Read(&recv_buffer, tag(5));
561 Verifier().Expect(5, true).Verify(srv_cq.get());
562 EXPECT_TRUE(ParseFromByteBuffer(&recv_buffer, &recv_request));
563 EXPECT_EQ(send_request.message(), recv_request.message());
564
565 send_response.set_message(recv_request.message());
566 send_buffer = SerializeToByteBuffer(&send_response);
567 stream.Write(*send_buffer, tag(6));
568 send_buffer.reset();
569 Verifier().Expect(6, true).Verify(srv_cq.get());
570
571 stream.Finish(Status::OK, tag(7));
572 // Shutdown srv_cq before we try to get the tag back, to verify that the
573 // interception API handles completion queue shutdowns that take place before
574 // all the tags are returned
575 srv_cq->Shutdown();
576 Verifier().Expect(7, true).Verify(srv_cq.get());
577
578 recv_buffer.Clear();
579 call->Read(&recv_buffer, tag(8));
580 Verifier().Expect(8, true).Verify(&cli_cq);
581 EXPECT_TRUE(ParseFromByteBuffer(&recv_buffer, &recv_response));
582
583 call->Finish(&recv_status, tag(9));
584 cli_cq.Shutdown();
585 Verifier().Expect(9, true).Verify(&cli_cq);
586
587 EXPECT_EQ(send_response.message(), recv_response.message());
588 EXPECT_TRUE(recv_status.ok());
589 EXPECT_TRUE(CheckMetadata(cli_ctx.GetServerTrailingMetadata(), "testkey",
590 "testvalue"));
591
592 // Make sure all 20 dummy interceptors were run
593 EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20);
594
595 server->Shutdown();
596 void* ignored_tag;
597 bool ignored_ok;
598 while (cli_cq.Next(&ignored_tag, &ignored_ok))
599 ;
600 while (srv_cq->Next(&ignored_tag, &ignored_ok))
601 ;
602 grpc_recycle_unused_port(port);
603 }
604
TEST_F(ServerInterceptorsAsyncEnd2endTest,UnimplementedRpcTest)605 TEST_F(ServerInterceptorsAsyncEnd2endTest, UnimplementedRpcTest) {
606 DummyInterceptor::Reset();
607 int port = grpc_pick_unused_port_or_die();
608 string server_address = "localhost:" + std::to_string(port);
609 ServerBuilder builder;
610 builder.AddListeningPort(server_address, InsecureServerCredentials());
611 std::vector<std::unique_ptr<experimental::ServerInterceptorFactoryInterface>>
612 creators;
613 creators.reserve(20);
614 for (auto i = 0; i < 20; i++) {
615 creators.push_back(std::unique_ptr<DummyInterceptorFactory>(
616 new DummyInterceptorFactory()));
617 }
618 builder.experimental().SetInterceptorCreators(std::move(creators));
619 auto cq = builder.AddCompletionQueue();
620 auto server = builder.BuildAndStart();
621
622 ChannelArguments args;
623 std::shared_ptr<Channel> channel =
624 grpc::CreateChannel(server_address, InsecureChannelCredentials());
625 std::unique_ptr<grpc::testing::UnimplementedEchoService::Stub> stub;
626 stub = grpc::testing::UnimplementedEchoService::NewStub(channel);
627 EchoRequest send_request;
628 EchoResponse recv_response;
629 Status recv_status;
630
631 ClientContext cli_ctx;
632 send_request.set_message("Hello");
633 std::unique_ptr<ClientAsyncResponseReader<EchoResponse>> response_reader(
634 stub->AsyncUnimplemented(&cli_ctx, send_request, cq.get()));
635
636 response_reader->Finish(&recv_response, &recv_status, tag(4));
637 Verifier().Expect(4, true).Verify(cq.get());
638
639 EXPECT_EQ(StatusCode::UNIMPLEMENTED, recv_status.error_code());
640 EXPECT_EQ("", recv_status.error_message());
641
642 // Make sure all 20 dummy interceptors were run
643 EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20);
644
645 server->Shutdown();
646 cq->Shutdown();
647 void* ignored_tag;
648 bool ignored_ok;
649 while (cq->Next(&ignored_tag, &ignored_ok))
650 ;
651 grpc_recycle_unused_port(port);
652 }
653
654 class ServerInterceptorsSyncUnimplementedEnd2endTest : public ::testing::Test {
655 };
656
TEST_F(ServerInterceptorsSyncUnimplementedEnd2endTest,UnimplementedRpcTest)657 TEST_F(ServerInterceptorsSyncUnimplementedEnd2endTest, UnimplementedRpcTest) {
658 DummyInterceptor::Reset();
659 int port = grpc_pick_unused_port_or_die();
660 string server_address = "localhost:" + std::to_string(port);
661 ServerBuilder builder;
662 TestServiceImpl service;
663 builder.RegisterService(&service);
664 builder.AddListeningPort(server_address, InsecureServerCredentials());
665 std::vector<std::unique_ptr<experimental::ServerInterceptorFactoryInterface>>
666 creators;
667 creators.reserve(20);
668 for (auto i = 0; i < 20; i++) {
669 creators.push_back(std::unique_ptr<DummyInterceptorFactory>(
670 new DummyInterceptorFactory()));
671 }
672 builder.experimental().SetInterceptorCreators(std::move(creators));
673 auto server = builder.BuildAndStart();
674
675 ChannelArguments args;
676 std::shared_ptr<Channel> channel =
677 grpc::CreateChannel(server_address, InsecureChannelCredentials());
678 std::unique_ptr<grpc::testing::UnimplementedEchoService::Stub> stub;
679 stub = grpc::testing::UnimplementedEchoService::NewStub(channel);
680 EchoRequest send_request;
681 EchoResponse recv_response;
682
683 ClientContext cli_ctx;
684 send_request.set_message("Hello");
685 Status recv_status =
686 stub->Unimplemented(&cli_ctx, send_request, &recv_response);
687
688 EXPECT_EQ(StatusCode::UNIMPLEMENTED, recv_status.error_code());
689 EXPECT_EQ("", recv_status.error_message());
690
691 // Make sure all 20 dummy interceptors were run
692 EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20);
693
694 server->Shutdown();
695 grpc_recycle_unused_port(port);
696 }
697
698 } // namespace
699 } // namespace testing
700 } // namespace grpc
701
main(int argc,char ** argv)702 int main(int argc, char** argv) {
703 grpc::testing::TestEnvironment env(argc, argv);
704 ::testing::InitGoogleTest(&argc, argv);
705 return RUN_ALL_TESTS();
706 }
707