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