test/cpp/end2end/interceptors_util.cc

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