3 * Copyright 2016 gRPC authors.
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
9 * http://www.apache.org/licenses/LICENSE-2.0
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.
23 #include <grpc/grpc.h>
24 #include <grpc/support/time.h>
25 #include <grpcpp/channel.h>
26 #include <grpcpp/client_context.h>
27 #include <grpcpp/create_channel.h>
28 #include <grpcpp/generic/async_generic_service.h>
29 #include <grpcpp/generic/generic_stub.h>
30 #include <grpcpp/impl/codegen/proto_utils.h>
31 #include <grpcpp/server.h>
32 #include <grpcpp/server_builder.h>
33 #include <grpcpp/server_context.h>
34 #include <grpcpp/support/config.h>
35 #include <grpcpp/support/slice.h>
37 #include "absl/memory/memory.h"
39 #include "src/cpp/common/channel_filter.h"
40 #include "src/proto/grpc/testing/echo.grpc.pb.h"
41 #include "test/core/util/port.h"
42 #include "test/core/util/test_config.h"
43 #include "test/cpp/util/byte_buffer_proto_helper.h"
45 #include <gtest/gtest.h>
47 using grpc::testing::EchoRequest;
48 using grpc::testing::EchoResponse;
49 using std::chrono::system_clock;
55 void* tag(int i) { return (void*)static_cast<intptr_t>(i); }
57 void verify_ok(CompletionQueue* cq, int i, bool expect_ok) {
60 EXPECT_TRUE(cq->Next(&got_tag, &ok));
61 EXPECT_EQ(expect_ok, ok);
62 EXPECT_EQ(tag(i), got_tag);
67 int global_num_connections = 0;
68 int global_num_calls = 0;
71 void IncrementConnectionCounter() {
72 std::unique_lock<std::mutex> lock(global_mu);
73 ++global_num_connections;
76 void ResetConnectionCounter() {
77 std::unique_lock<std::mutex> lock(global_mu);
78 global_num_connections = 0;
81 int GetConnectionCounterValue() {
82 std::unique_lock<std::mutex> lock(global_mu);
83 return global_num_connections;
86 void IncrementCallCounter() {
87 std::unique_lock<std::mutex> lock(global_mu);
91 void ResetCallCounter() {
92 std::unique_lock<std::mutex> lock(global_mu);
96 int GetCallCounterValue() {
97 std::unique_lock<std::mutex> lock(global_mu);
98 return global_num_calls;
103 class ChannelDataImpl : public ChannelData {
105 grpc_error* Init(grpc_channel_element* /*elem*/,
106 grpc_channel_element_args* /*args*/) override {
107 IncrementConnectionCounter();
108 return GRPC_ERROR_NONE;
112 class CallDataImpl : public CallData {
114 void StartTransportStreamOpBatch(grpc_call_element* elem,
115 TransportStreamOpBatch* op) override {
116 // Incrementing the counter could be done from Init(), but we want
117 // to test that the individual methods are actually called correctly.
118 if (op->recv_initial_metadata() != nullptr) IncrementCallCounter();
119 grpc_call_next_op(elem, op->op());
123 class FilterEnd2endTest : public ::testing::Test {
125 FilterEnd2endTest() : server_host_("localhost") {}
127 static void SetUpTestCase() {
129 // https://github.com/google/google-toolbox-for-mac/issues/242
130 static bool setup_done = false;
133 grpc::RegisterChannelFilter<ChannelDataImpl, CallDataImpl>(
134 "test-filter", GRPC_SERVER_CHANNEL, INT_MAX, nullptr);
138 void SetUp() override {
139 int port = grpc_pick_unused_port_or_die();
140 server_address_ << server_host_ << ":" << port;
142 ServerBuilder builder;
143 builder.AddListeningPort(server_address_.str(),
144 InsecureServerCredentials());
145 builder.RegisterAsyncGenericService(&generic_service_);
146 srv_cq_ = builder.AddCompletionQueue();
147 server_ = builder.BuildAndStart();
150 void TearDown() override {
156 while (cli_cq_.Next(&ignored_tag, &ignored_ok))
158 while (srv_cq_->Next(&ignored_tag, &ignored_ok))
163 std::shared_ptr<Channel> channel = grpc::CreateChannel(
164 server_address_.str(), InsecureChannelCredentials());
165 generic_stub_ = absl::make_unique<GenericStub>(channel);
166 ResetConnectionCounter();
170 void server_ok(int i) { verify_ok(srv_cq_.get(), i, true); }
171 void client_ok(int i) { verify_ok(&cli_cq_, i, true); }
172 void server_fail(int i) { verify_ok(srv_cq_.get(), i, false); }
173 void client_fail(int i) { verify_ok(&cli_cq_, i, false); }
175 void SendRpc(int num_rpcs) {
176 const std::string kMethodName("/grpc.cpp.test.util.EchoTestService/Echo");
177 for (int i = 0; i < num_rpcs; i++) {
178 EchoRequest send_request;
179 EchoRequest recv_request;
180 EchoResponse send_response;
181 EchoResponse recv_response;
184 ClientContext cli_ctx;
185 GenericServerContext srv_ctx;
186 GenericServerAsyncReaderWriter stream(&srv_ctx);
188 // The string needs to be long enough to test heap-based slice.
189 send_request.set_message("Hello world. Hello world. Hello world.");
190 std::thread request_call([this]() { server_ok(4); });
191 std::unique_ptr<GenericClientAsyncReaderWriter> call =
192 generic_stub_->PrepareCall(&cli_ctx, kMethodName, &cli_cq_);
193 call->StartCall(tag(1));
195 std::unique_ptr<ByteBuffer> send_buffer =
196 SerializeToByteBuffer(&send_request);
197 call->Write(*send_buffer, tag(2));
198 // Send ByteBuffer can be destroyed after calling Write.
201 call->WritesDone(tag(3));
204 generic_service_.RequestCall(&srv_ctx, &stream, srv_cq_.get(),
205 srv_cq_.get(), tag(4));
208 EXPECT_EQ(server_host_, srv_ctx.host().substr(0, server_host_.length()));
209 EXPECT_EQ(kMethodName, srv_ctx.method());
210 ByteBuffer recv_buffer;
211 stream.Read(&recv_buffer, tag(5));
213 EXPECT_TRUE(ParseFromByteBuffer(&recv_buffer, &recv_request));
214 EXPECT_EQ(send_request.message(), recv_request.message());
216 send_response.set_message(recv_request.message());
217 send_buffer = SerializeToByteBuffer(&send_response);
218 stream.Write(*send_buffer, tag(6));
222 stream.Finish(Status::OK, tag(7));
226 call->Read(&recv_buffer, tag(8));
228 EXPECT_TRUE(ParseFromByteBuffer(&recv_buffer, &recv_response));
230 call->Finish(&recv_status, tag(9));
233 EXPECT_EQ(send_response.message(), recv_response.message());
234 EXPECT_TRUE(recv_status.ok());
238 CompletionQueue cli_cq_;
239 std::unique_ptr<ServerCompletionQueue> srv_cq_;
240 std::unique_ptr<grpc::testing::EchoTestService::Stub> stub_;
241 std::unique_ptr<grpc::GenericStub> generic_stub_;
242 std::unique_ptr<Server> server_;
243 AsyncGenericService generic_service_;
244 const std::string server_host_;
245 std::ostringstream server_address_;
248 TEST_F(FilterEnd2endTest, SimpleRpc) {
250 EXPECT_EQ(0, GetConnectionCounterValue());
251 EXPECT_EQ(0, GetCallCounterValue());
253 EXPECT_EQ(1, GetConnectionCounterValue());
254 EXPECT_EQ(1, GetCallCounterValue());
257 TEST_F(FilterEnd2endTest, SequentialRpcs) {
259 EXPECT_EQ(0, GetConnectionCounterValue());
260 EXPECT_EQ(0, GetCallCounterValue());
262 EXPECT_EQ(1, GetConnectionCounterValue());
263 EXPECT_EQ(10, GetCallCounterValue());
266 // One ping, one pong.
267 TEST_F(FilterEnd2endTest, SimpleBidiStreaming) {
269 EXPECT_EQ(0, GetConnectionCounterValue());
270 EXPECT_EQ(0, GetCallCounterValue());
272 const std::string kMethodName(
273 "/grpc.cpp.test.util.EchoTestService/BidiStream");
274 EchoRequest send_request;
275 EchoRequest recv_request;
276 EchoResponse send_response;
277 EchoResponse recv_response;
279 ClientContext cli_ctx;
280 GenericServerContext srv_ctx;
281 GenericServerAsyncReaderWriter srv_stream(&srv_ctx);
283 cli_ctx.set_compression_algorithm(GRPC_COMPRESS_GZIP);
284 send_request.set_message("Hello");
285 std::thread request_call([this]() { server_ok(2); });
286 std::unique_ptr<GenericClientAsyncReaderWriter> cli_stream =
287 generic_stub_->PrepareCall(&cli_ctx, kMethodName, &cli_cq_);
288 cli_stream->StartCall(tag(1));
291 generic_service_.RequestCall(&srv_ctx, &srv_stream, srv_cq_.get(),
292 srv_cq_.get(), tag(2));
295 EXPECT_EQ(server_host_, srv_ctx.host().substr(0, server_host_.length()));
296 EXPECT_EQ(kMethodName, srv_ctx.method());
298 std::unique_ptr<ByteBuffer> send_buffer =
299 SerializeToByteBuffer(&send_request);
300 cli_stream->Write(*send_buffer, tag(3));
304 ByteBuffer recv_buffer;
305 srv_stream.Read(&recv_buffer, tag(4));
307 EXPECT_TRUE(ParseFromByteBuffer(&recv_buffer, &recv_request));
308 EXPECT_EQ(send_request.message(), recv_request.message());
310 send_response.set_message(recv_request.message());
311 send_buffer = SerializeToByteBuffer(&send_response);
312 srv_stream.Write(*send_buffer, tag(5));
316 cli_stream->Read(&recv_buffer, tag(6));
318 EXPECT_TRUE(ParseFromByteBuffer(&recv_buffer, &recv_response));
319 EXPECT_EQ(send_response.message(), recv_response.message());
321 cli_stream->WritesDone(tag(7));
324 srv_stream.Read(&recv_buffer, tag(8));
327 srv_stream.Finish(Status::OK, tag(9));
330 cli_stream->Finish(&recv_status, tag(10));
333 EXPECT_EQ(send_response.message(), recv_response.message());
334 EXPECT_TRUE(recv_status.ok());
336 EXPECT_EQ(1, GetCallCounterValue());
337 EXPECT_EQ(1, GetConnectionCounterValue());
341 } // namespace testing
344 int main(int argc, char** argv) {
345 grpc::testing::TestEnvironment env(argc, argv);
346 ::testing::InitGoogleTest(&argc, argv);
347 return RUN_ALL_TESTS();