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 "test/cpp/util/cli_call.h"
20
21 #include <cmath>
22 #include <iostream>
23 #include <utility>
24
25 #include <grpc/grpc.h>
26 #include <grpc/slice.h>
27 #include <grpc/support/log.h>
28 #include <grpcpp/channel.h>
29 #include <grpcpp/client_context.h>
30 #include <grpcpp/support/byte_buffer.h>
31
32 #include "src/core/lib/gprpp/crash.h"
33
34 namespace grpc {
35 namespace testing {
36 namespace {
tag(intptr_t t)37 void* tag(intptr_t t) { return reinterpret_cast<void*>(t); }
38 } // namespace
39
Call(const std::string & request,std::string * response,IncomingMetadataContainer * server_initial_metadata,IncomingMetadataContainer * server_trailing_metadata)40 Status CliCall::Call(const std::string& request, std::string* response,
41 IncomingMetadataContainer* server_initial_metadata,
42 IncomingMetadataContainer* server_trailing_metadata) {
43 Write(request);
44 WritesDone();
45 if (!Read(response, server_initial_metadata)) {
46 fprintf(stderr, "Failed to read response.\n");
47 }
48 return Finish(server_trailing_metadata);
49 }
50
CliCall(const std::shared_ptr<grpc::Channel> & channel,const std::string & method,const OutgoingMetadataContainer & metadata,CliArgs args)51 CliCall::CliCall(const std::shared_ptr<grpc::Channel>& channel,
52 const std::string& method,
53 const OutgoingMetadataContainer& metadata, CliArgs args)
54 : stub_(new grpc::GenericStub(channel)) {
55 gpr_mu_init(&write_mu_);
56 gpr_cv_init(&write_cv_);
57 if (!metadata.empty()) {
58 for (OutgoingMetadataContainer::const_iterator iter = metadata.begin();
59 iter != metadata.end(); ++iter) {
60 ctx_.AddMetadata(iter->first, iter->second);
61 }
62 }
63
64 // Set deadline if timeout > 0 (default value -1 if no timeout specified)
65 if (args.timeout > 0) {
66 int64_t timeout_in_ns = ceil(args.timeout * 1e9);
67
68 // Convert timeout (in nanoseconds) to a deadline
69 auto deadline =
70 gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC),
71 gpr_time_from_nanos(timeout_in_ns, GPR_TIMESPAN));
72 ctx_.set_deadline(deadline);
73 } else if (args.timeout != -1) {
74 fprintf(
75 stderr,
76 "WARNING: Non-positive timeout value, skipping setting deadline.\n");
77 }
78
79 call_ = stub_->PrepareCall(&ctx_, method, &cq_);
80 call_->StartCall(tag(1));
81 void* got_tag;
82 bool ok;
83 cq_.Next(&got_tag, &ok);
84 GPR_ASSERT(ok);
85 }
86
~CliCall()87 CliCall::~CliCall() {
88 gpr_cv_destroy(&write_cv_);
89 gpr_mu_destroy(&write_mu_);
90 }
91
Write(const std::string & request)92 void CliCall::Write(const std::string& request) {
93 void* got_tag;
94 bool ok;
95
96 grpc_slice s = grpc_slice_from_copied_buffer(request.data(), request.size());
97 grpc::Slice req_slice(s, grpc::Slice::STEAL_REF);
98 grpc::ByteBuffer send_buffer(&req_slice, 1);
99 call_->Write(send_buffer, tag(2));
100 cq_.Next(&got_tag, &ok);
101 GPR_ASSERT(ok);
102 }
103
Read(std::string * response,IncomingMetadataContainer * server_initial_metadata)104 bool CliCall::Read(std::string* response,
105 IncomingMetadataContainer* server_initial_metadata) {
106 void* got_tag;
107 bool ok;
108
109 grpc::ByteBuffer recv_buffer;
110 call_->Read(&recv_buffer, tag(3));
111
112 if (!cq_.Next(&got_tag, &ok) || !ok) {
113 return false;
114 }
115 std::vector<grpc::Slice> slices;
116 GPR_ASSERT(recv_buffer.Dump(&slices).ok());
117
118 response->clear();
119 for (size_t i = 0; i < slices.size(); i++) {
120 response->append(reinterpret_cast<const char*>(slices[i].begin()),
121 slices[i].size());
122 }
123 if (server_initial_metadata) {
124 *server_initial_metadata = ctx_.GetServerInitialMetadata();
125 }
126 return true;
127 }
128
WritesDone()129 void CliCall::WritesDone() {
130 void* got_tag;
131 bool ok;
132
133 call_->WritesDone(tag(4));
134 cq_.Next(&got_tag, &ok);
135 GPR_ASSERT(ok);
136 }
137
WriteAndWait(const std::string & request)138 void CliCall::WriteAndWait(const std::string& request) {
139 grpc::Slice req_slice(request);
140 grpc::ByteBuffer send_buffer(&req_slice, 1);
141
142 gpr_mu_lock(&write_mu_);
143 call_->Write(send_buffer, tag(2));
144 write_done_ = false;
145 while (!write_done_) {
146 gpr_cv_wait(&write_cv_, &write_mu_, gpr_inf_future(GPR_CLOCK_MONOTONIC));
147 }
148 gpr_mu_unlock(&write_mu_);
149 }
150
WritesDoneAndWait()151 void CliCall::WritesDoneAndWait() {
152 gpr_mu_lock(&write_mu_);
153 call_->WritesDone(tag(4));
154 write_done_ = false;
155 while (!write_done_) {
156 gpr_cv_wait(&write_cv_, &write_mu_, gpr_inf_future(GPR_CLOCK_MONOTONIC));
157 }
158 gpr_mu_unlock(&write_mu_);
159 }
160
ReadAndMaybeNotifyWrite(std::string * response,IncomingMetadataContainer * server_initial_metadata)161 bool CliCall::ReadAndMaybeNotifyWrite(
162 std::string* response, IncomingMetadataContainer* server_initial_metadata) {
163 void* got_tag;
164 bool ok;
165 grpc::ByteBuffer recv_buffer;
166
167 call_->Read(&recv_buffer, tag(3));
168 bool cq_result = cq_.Next(&got_tag, &ok);
169
170 while (got_tag != tag(3)) {
171 gpr_mu_lock(&write_mu_);
172 write_done_ = true;
173 gpr_cv_signal(&write_cv_);
174 gpr_mu_unlock(&write_mu_);
175
176 cq_result = cq_.Next(&got_tag, &ok);
177 if (got_tag == tag(2)) {
178 GPR_ASSERT(ok);
179 }
180 }
181
182 if (!cq_result || !ok) {
183 // If the RPC is ended on the server side, we should still wait for the
184 // pending write on the client side to be done.
185 if (!ok) {
186 gpr_mu_lock(&write_mu_);
187 if (!write_done_) {
188 cq_.Next(&got_tag, &ok);
189 GPR_ASSERT(got_tag != tag(2));
190 write_done_ = true;
191 gpr_cv_signal(&write_cv_);
192 }
193 gpr_mu_unlock(&write_mu_);
194 }
195 return false;
196 }
197
198 std::vector<grpc::Slice> slices;
199 GPR_ASSERT(recv_buffer.Dump(&slices).ok());
200 response->clear();
201 for (size_t i = 0; i < slices.size(); i++) {
202 response->append(reinterpret_cast<const char*>(slices[i].begin()),
203 slices[i].size());
204 }
205 if (server_initial_metadata) {
206 *server_initial_metadata = ctx_.GetServerInitialMetadata();
207 }
208 return true;
209 }
210
Finish(IncomingMetadataContainer * server_trailing_metadata)211 Status CliCall::Finish(IncomingMetadataContainer* server_trailing_metadata) {
212 void* got_tag;
213 bool ok;
214 grpc::Status status;
215
216 call_->Finish(&status, tag(5));
217 cq_.Next(&got_tag, &ok);
218 GPR_ASSERT(ok);
219 if (server_trailing_metadata) {
220 *server_trailing_metadata = ctx_.GetServerTrailingMetadata();
221 }
222
223 return status;
224 }
225
226 } // namespace testing
227 } // namespace grpc
228