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 "test/cpp/end2end/interceptors_util.h"
20
21 #include "absl/memory/memory.h"
22
23 namespace grpc {
24 namespace testing {
25
26 std::atomic<int> PhonyInterceptor::num_times_run_;
27 std::atomic<int> PhonyInterceptor::num_times_run_reverse_;
28 std::atomic<int> PhonyInterceptor::num_times_cancel_;
29
MakeCall(const std::shared_ptr<Channel> & channel,const StubOptions & options)30 void MakeCall(const std::shared_ptr<Channel>& channel,
31 const StubOptions& options) {
32 auto stub = grpc::testing::EchoTestService::NewStub(channel, options);
33 ClientContext ctx;
34 EchoRequest req;
35 req.mutable_param()->set_echo_metadata(true);
36 ctx.AddMetadata("testkey", "testvalue");
37 req.set_message("Hello");
38 EchoResponse resp;
39 Status s = stub->Echo(&ctx, req, &resp);
40 EXPECT_EQ(s.ok(), true);
41 EXPECT_EQ(resp.message(), "Hello");
42 }
43
MakeClientStreamingCall(const std::shared_ptr<Channel> & channel)44 void MakeClientStreamingCall(const std::shared_ptr<Channel>& channel) {
45 auto stub = grpc::testing::EchoTestService::NewStub(channel);
46 ClientContext ctx;
47 EchoRequest req;
48 req.mutable_param()->set_echo_metadata(true);
49 ctx.AddMetadata("testkey", "testvalue");
50 req.set_message("Hello");
51 EchoResponse resp;
52 string expected_resp = "";
53 auto writer = stub->RequestStream(&ctx, &resp);
54 for (int i = 0; i < kNumStreamingMessages; i++) {
55 writer->Write(req);
56 expected_resp += "Hello";
57 }
58 writer->WritesDone();
59 Status s = writer->Finish();
60 EXPECT_EQ(s.ok(), true);
61 EXPECT_EQ(resp.message(), expected_resp);
62 }
63
MakeServerStreamingCall(const std::shared_ptr<Channel> & channel)64 void MakeServerStreamingCall(const std::shared_ptr<Channel>& channel) {
65 auto stub = grpc::testing::EchoTestService::NewStub(channel);
66 ClientContext ctx;
67 EchoRequest req;
68 req.mutable_param()->set_echo_metadata(true);
69 ctx.AddMetadata("testkey", "testvalue");
70 req.set_message("Hello");
71 EchoResponse resp;
72 auto reader = stub->ResponseStream(&ctx, req);
73 int count = 0;
74 while (reader->Read(&resp)) {
75 EXPECT_EQ(resp.message(), "Hello");
76 count++;
77 }
78 ASSERT_EQ(count, kNumStreamingMessages);
79 Status s = reader->Finish();
80 EXPECT_EQ(s.ok(), true);
81 }
82
MakeBidiStreamingCall(const std::shared_ptr<Channel> & channel)83 void MakeBidiStreamingCall(const std::shared_ptr<Channel>& channel) {
84 auto stub = grpc::testing::EchoTestService::NewStub(channel);
85 ClientContext ctx;
86 EchoRequest req;
87 EchoResponse resp;
88 ctx.AddMetadata("testkey", "testvalue");
89 req.mutable_param()->set_echo_metadata(true);
90 auto stream = stub->BidiStream(&ctx);
91 for (auto i = 0; i < kNumStreamingMessages; i++) {
92 req.set_message("Hello" + std::to_string(i));
93 stream->Write(req);
94 stream->Read(&resp);
95 EXPECT_EQ(req.message(), resp.message());
96 }
97 ASSERT_TRUE(stream->WritesDone());
98 Status s = stream->Finish();
99 EXPECT_EQ(s.ok(), true);
100 }
101
MakeAsyncCQCall(const std::shared_ptr<Channel> & channel)102 void MakeAsyncCQCall(const std::shared_ptr<Channel>& channel) {
103 auto stub = grpc::testing::EchoTestService::NewStub(channel);
104 CompletionQueue cq;
105 EchoRequest send_request;
106 EchoResponse recv_response;
107 Status recv_status;
108 ClientContext cli_ctx;
109
110 send_request.set_message("Hello");
111 cli_ctx.AddMetadata("testkey", "testvalue");
112 std::unique_ptr<ClientAsyncResponseReader<EchoResponse>> response_reader(
113 stub->AsyncEcho(&cli_ctx, send_request, &cq));
114 response_reader->Finish(&recv_response, &recv_status, tag(1));
115 Verifier().Expect(1, true).Verify(&cq);
116 EXPECT_EQ(send_request.message(), recv_response.message());
117 EXPECT_TRUE(recv_status.ok());
118 }
119
MakeAsyncCQClientStreamingCall(const std::shared_ptr<Channel> &)120 void MakeAsyncCQClientStreamingCall(
121 const std::shared_ptr<Channel>& /*channel*/) {
122 // TODO(yashykt) : Fill this out
123 }
124
MakeAsyncCQServerStreamingCall(const std::shared_ptr<Channel> & channel)125 void MakeAsyncCQServerStreamingCall(const std::shared_ptr<Channel>& channel) {
126 auto stub = grpc::testing::EchoTestService::NewStub(channel);
127 CompletionQueue cq;
128 EchoRequest send_request;
129 EchoResponse recv_response;
130 Status recv_status;
131 ClientContext cli_ctx;
132
133 cli_ctx.AddMetadata("testkey", "testvalue");
134 send_request.set_message("Hello");
135 std::unique_ptr<ClientAsyncReader<EchoResponse>> cli_stream(
136 stub->AsyncResponseStream(&cli_ctx, send_request, &cq, tag(1)));
137 Verifier().Expect(1, true).Verify(&cq);
138 // Read the expected number of messages
139 for (int i = 0; i < kNumStreamingMessages; i++) {
140 cli_stream->Read(&recv_response, tag(2));
141 Verifier().Expect(2, true).Verify(&cq);
142 ASSERT_EQ(recv_response.message(), send_request.message());
143 }
144 // The next read should fail
145 cli_stream->Read(&recv_response, tag(3));
146 Verifier().Expect(3, false).Verify(&cq);
147 // Get the status
148 cli_stream->Finish(&recv_status, tag(4));
149 Verifier().Expect(4, true).Verify(&cq);
150 EXPECT_TRUE(recv_status.ok());
151 }
152
MakeAsyncCQBidiStreamingCall(const std::shared_ptr<Channel> &)153 void MakeAsyncCQBidiStreamingCall(const std::shared_ptr<Channel>& /*channel*/) {
154 // TODO(yashykt) : Fill this out
155 }
156
MakeCallbackCall(const std::shared_ptr<Channel> & channel)157 void MakeCallbackCall(const std::shared_ptr<Channel>& channel) {
158 auto stub = grpc::testing::EchoTestService::NewStub(channel);
159 ClientContext ctx;
160 EchoRequest req;
161 std::mutex mu;
162 std::condition_variable cv;
163 bool done = false;
164 req.mutable_param()->set_echo_metadata(true);
165 ctx.AddMetadata("testkey", "testvalue");
166 req.set_message("Hello");
167 EchoResponse resp;
168 stub->async()->Echo(&ctx, &req, &resp, [&resp, &mu, &done, &cv](Status s) {
169 EXPECT_EQ(s.ok(), true);
170 EXPECT_EQ(resp.message(), "Hello");
171 std::lock_guard<std::mutex> l(mu);
172 done = true;
173 cv.notify_one();
174 });
175 std::unique_lock<std::mutex> l(mu);
176 while (!done) {
177 cv.wait(l);
178 }
179 }
180
CheckMetadata(const std::multimap<grpc::string_ref,grpc::string_ref> & map,const string & key,const string & value)181 bool CheckMetadata(const std::multimap<grpc::string_ref, grpc::string_ref>& map,
182 const string& key, const string& value) {
183 for (const auto& pair : map) {
184 if (pair.first.starts_with(key) && pair.second.starts_with(value)) {
185 return true;
186 }
187 }
188 return false;
189 }
190
CheckMetadata(const std::multimap<std::string,std::string> & map,const string & key,const string & value)191 bool CheckMetadata(const std::multimap<std::string, std::string>& map,
192 const string& key, const string& value) {
193 for (const auto& pair : map) {
194 if (pair.first == key && pair.second == value) {
195 return true;
196 }
197 }
198 return false;
199 }
200
201 std::vector<std::unique_ptr<experimental::ClientInterceptorFactoryInterface>>
CreatePhonyClientInterceptors()202 CreatePhonyClientInterceptors() {
203 std::vector<std::unique_ptr<experimental::ClientInterceptorFactoryInterface>>
204 creators;
205 // Add 20 phony interceptors before hijacking interceptor
206 creators.reserve(20);
207 for (auto i = 0; i < 20; i++) {
208 creators.push_back(absl::make_unique<PhonyInterceptorFactory>());
209 }
210 return creators;
211 }
212
213 } // namespace testing
214 } // namespace grpc
215