3 * Copyright 2015 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.
19 #include "test/cpp/util/cli_call.h"
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>
35 void* tag(intptr_t t) { return reinterpret_cast<void*>(t); }
38 Status CliCall::Call(const std::shared_ptr<grpc::Channel>& channel,
39 const std::string& method, const std::string& request,
40 std::string* response,
41 const OutgoingMetadataContainer& metadata,
42 IncomingMetadataContainer* server_initial_metadata,
43 IncomingMetadataContainer* server_trailing_metadata) {
44 CliCall call(channel, method, metadata);
47 if (!call.Read(response, server_initial_metadata)) {
48 fprintf(stderr, "Failed to read response.\n");
50 return call.Finish(server_trailing_metadata);
53 CliCall::CliCall(const std::shared_ptr<grpc::Channel>& channel,
54 const std::string& method,
55 const OutgoingMetadataContainer& metadata, CliArgs args)
56 : stub_(new grpc::GenericStub(channel)) {
57 gpr_mu_init(&write_mu_);
58 gpr_cv_init(&write_cv_);
59 if (!metadata.empty()) {
60 for (OutgoingMetadataContainer::const_iterator iter = metadata.begin();
61 iter != metadata.end(); ++iter) {
62 ctx_.AddMetadata(iter->first, iter->second);
66 // Set deadline if timeout > 0 (default value -1 if no timeout specified)
67 if (args.timeout > 0) {
68 int64_t timeout_in_ns = ceil(args.timeout * 1e9);
70 // Convert timeout (in nanoseconds) to a deadline
72 gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC),
73 gpr_time_from_nanos(timeout_in_ns, GPR_TIMESPAN));
74 ctx_.set_deadline(deadline);
75 } else if (args.timeout != -1) {
78 "WARNING: Non-positive timeout value, skipping setting deadline.\n");
81 call_ = stub_->PrepareCall(&ctx_, method, &cq_);
82 call_->StartCall(tag(1));
85 cq_.Next(&got_tag, &ok);
90 gpr_cv_destroy(&write_cv_);
91 gpr_mu_destroy(&write_mu_);
94 void CliCall::Write(const std::string& request) {
98 gpr_slice s = gpr_slice_from_copied_buffer(request.data(), request.size());
99 grpc::Slice req_slice(s, grpc::Slice::STEAL_REF);
100 grpc::ByteBuffer send_buffer(&req_slice, 1);
101 call_->Write(send_buffer, tag(2));
102 cq_.Next(&got_tag, &ok);
106 bool CliCall::Read(std::string* response,
107 IncomingMetadataContainer* server_initial_metadata) {
111 grpc::ByteBuffer recv_buffer;
112 call_->Read(&recv_buffer, tag(3));
114 if (!cq_.Next(&got_tag, &ok) || !ok) {
117 std::vector<grpc::Slice> slices;
118 GPR_ASSERT(recv_buffer.Dump(&slices).ok());
121 for (size_t i = 0; i < slices.size(); i++) {
122 response->append(reinterpret_cast<const char*>(slices[i].begin()),
125 if (server_initial_metadata) {
126 *server_initial_metadata = ctx_.GetServerInitialMetadata();
131 void CliCall::WritesDone() {
135 call_->WritesDone(tag(4));
136 cq_.Next(&got_tag, &ok);
140 void CliCall::WriteAndWait(const std::string& request) {
141 grpc::Slice req_slice(request);
142 grpc::ByteBuffer send_buffer(&req_slice, 1);
144 gpr_mu_lock(&write_mu_);
145 call_->Write(send_buffer, tag(2));
147 while (!write_done_) {
148 gpr_cv_wait(&write_cv_, &write_mu_, gpr_inf_future(GPR_CLOCK_MONOTONIC));
150 gpr_mu_unlock(&write_mu_);
153 void CliCall::WritesDoneAndWait() {
154 gpr_mu_lock(&write_mu_);
155 call_->WritesDone(tag(4));
157 while (!write_done_) {
158 gpr_cv_wait(&write_cv_, &write_mu_, gpr_inf_future(GPR_CLOCK_MONOTONIC));
160 gpr_mu_unlock(&write_mu_);
163 bool CliCall::ReadAndMaybeNotifyWrite(
164 std::string* response, IncomingMetadataContainer* server_initial_metadata) {
167 grpc::ByteBuffer recv_buffer;
169 call_->Read(&recv_buffer, tag(3));
170 bool cq_result = cq_.Next(&got_tag, &ok);
172 while (got_tag != tag(3)) {
173 gpr_mu_lock(&write_mu_);
175 gpr_cv_signal(&write_cv_);
176 gpr_mu_unlock(&write_mu_);
178 cq_result = cq_.Next(&got_tag, &ok);
179 if (got_tag == tag(2)) {
184 if (!cq_result || !ok) {
185 // If the RPC is ended on the server side, we should still wait for the
186 // pending write on the client side to be done.
188 gpr_mu_lock(&write_mu_);
190 cq_.Next(&got_tag, &ok);
191 GPR_ASSERT(got_tag != tag(2));
193 gpr_cv_signal(&write_cv_);
195 gpr_mu_unlock(&write_mu_);
200 std::vector<grpc::Slice> slices;
201 GPR_ASSERT(recv_buffer.Dump(&slices).ok());
203 for (size_t i = 0; i < slices.size(); i++) {
204 response->append(reinterpret_cast<const char*>(slices[i].begin()),
207 if (server_initial_metadata) {
208 *server_initial_metadata = ctx_.GetServerInitialMetadata();
213 Status CliCall::Finish(IncomingMetadataContainer* server_trailing_metadata) {
218 call_->Finish(&status, tag(5));
219 cq_.Next(&got_tag, &ok);
221 if (server_trailing_metadata) {
222 *server_trailing_metadata = ctx_.GetServerTrailingMetadata();
228 } // namespace testing