test/cpp/util/cli_call.cc

   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 namespace grpc {
  33 namespace testing {
  34 namespace {
  35 void* tag(intptr_t t) { return reinterpret_cast<void*>(t); }
  36 }  // namespace
  37
  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);
  45   call.Write(request);
  46   call.WritesDone();
  47   if (!call.Read(response, server_initial_metadata)) {
  48     fprintf(stderr, "Failed to read response.\n");
  49   }
  50   return call.Finish(server_trailing_metadata);
  51 }
  52
  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);
  63     }
  64   }
  65
  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);
  69
  70     // Convert timeout (in nanoseconds) to a deadline
  71     auto 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) {
  76     fprintf(
  77         stderr,
  78         "WARNING: Non-positive timeout value, skipping setting deadline.\n");
  79   }
  80
  81   call_ = stub_->PrepareCall(&ctx_, method, &cq_);
  82   call_->StartCall(tag(1));
  83   void* got_tag;
  84   bool ok;
  85   cq_.Next(&got_tag, &ok);
  86   GPR_ASSERT(ok);
  87 }
  88
  89 CliCall::~CliCall() {
  90   gpr_cv_destroy(&write_cv_);
  91   gpr_mu_destroy(&write_mu_);
  92 }
  93
  94 void CliCall::Write(const std::string& request) {
  95   void* got_tag;
  96   bool ok;
  97
  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);
 103   GPR_ASSERT(ok);
 104 }
 105
 106 bool CliCall::Read(std::string* response,
 107                    IncomingMetadataContainer* server_initial_metadata) {
 108   void* got_tag;
 109   bool ok;
 110
 111   grpc::ByteBuffer recv_buffer;
 112   call_->Read(&recv_buffer, tag(3));
 113
 114   if (!cq_.Next(&got_tag, &ok) || !ok) {
 115     return false;
 116   }
 117   std::vector<grpc::Slice> slices;
 118   GPR_ASSERT(recv_buffer.Dump(&slices).ok());
 119
 120   response->clear();
 121   for (size_t i = 0; i < slices.size(); i++) {
 122     response->append(reinterpret_cast<const char*>(slices[i].begin()),
 123                      slices[i].size());
 124   }
 125   if (server_initial_metadata) {
 126     *server_initial_metadata = ctx_.GetServerInitialMetadata();
 127   }
 128   return true;
 129 }
 130
 131 void CliCall::WritesDone() {
 132   void* got_tag;
 133   bool ok;
 134
 135   call_->WritesDone(tag(4));
 136   cq_.Next(&got_tag, &ok);
 137   GPR_ASSERT(ok);
 138 }
 139
 140 void CliCall::WriteAndWait(const std::string& request) {
 141   grpc::Slice req_slice(request);
 142   grpc::ByteBuffer send_buffer(&req_slice, 1);
 143
 144   gpr_mu_lock(&write_mu_);
 145   call_->Write(send_buffer, tag(2));
 146   write_done_ = false;
 147   while (!write_done_) {
 148     gpr_cv_wait(&write_cv_, &write_mu_, gpr_inf_future(GPR_CLOCK_MONOTONIC));
 149   }
 150   gpr_mu_unlock(&write_mu_);
 151 }
 152
 153 void CliCall::WritesDoneAndWait() {
 154   gpr_mu_lock(&write_mu_);
 155   call_->WritesDone(tag(4));
 156   write_done_ = false;
 157   while (!write_done_) {
 158     gpr_cv_wait(&write_cv_, &write_mu_, gpr_inf_future(GPR_CLOCK_MONOTONIC));
 159   }
 160   gpr_mu_unlock(&write_mu_);
 161 }
 162
 163 bool CliCall::ReadAndMaybeNotifyWrite(
 164     std::string* response, IncomingMetadataContainer* server_initial_metadata) {
 165   void* got_tag;
 166   bool ok;
 167   grpc::ByteBuffer recv_buffer;
 168
 169   call_->Read(&recv_buffer, tag(3));
 170   bool cq_result = cq_.Next(&got_tag, &ok);
 171
 172   while (got_tag != tag(3)) {
 173     gpr_mu_lock(&write_mu_);
 174     write_done_ = true;
 175     gpr_cv_signal(&write_cv_);
 176     gpr_mu_unlock(&write_mu_);
 177
 178     cq_result = cq_.Next(&got_tag, &ok);
 179     if (got_tag == tag(2)) {
 180       GPR_ASSERT(ok);
 181     }
 182   }
 183
 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.
 187     if (!ok) {
 188       gpr_mu_lock(&write_mu_);
 189       if (!write_done_) {
 190         cq_.Next(&got_tag, &ok);
 191         GPR_ASSERT(got_tag != tag(2));
 192         write_done_ = true;
 193         gpr_cv_signal(&write_cv_);
 194       }
 195       gpr_mu_unlock(&write_mu_);
 196     }
 197     return false;
 198   }
 199
 200   std::vector<grpc::Slice> slices;
 201   GPR_ASSERT(recv_buffer.Dump(&slices).ok());
 202   response->clear();
 203   for (size_t i = 0; i < slices.size(); i++) {
 204     response->append(reinterpret_cast<const char*>(slices[i].begin()),
 205                      slices[i].size());
 206   }
 207   if (server_initial_metadata) {
 208     *server_initial_metadata = ctx_.GetServerInitialMetadata();
 209   }
 210   return true;
 211 }
 212
 213 Status CliCall::Finish(IncomingMetadataContainer* server_trailing_metadata) {
 214   void* got_tag;
 215   bool ok;
 216   grpc::Status status;
 217
 218   call_->Finish(&status, tag(5));
 219   cq_.Next(&got_tag, &ok);
 220   GPR_ASSERT(ok);
 221   if (server_trailing_metadata) {
 222     *server_trailing_metadata = ctx_.GetServerTrailingMetadata();
 223   }
 224
 225   return status;
 226 }
 227
 228 }  // namespace testing
 229 }  // namespace grpc