test/core/surface/lame_client_test.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 <string.h>
  20
  21 #include <grpc/grpc.h>
  22 #include <grpc/support/alloc.h>
  23 #include <grpc/support/log.h>
  24
  25 #include "src/core/lib/channel/channel_stack.h"
  26 #include "src/core/lib/iomgr/closure.h"
  27 #include "src/core/lib/surface/channel.h"
  28 #include "src/core/lib/transport/transport.h"
  29 #include "test/core/end2end/cq_verifier.h"
  30 #include "test/core/util/test_config.h"
  31
  32 class Watcher : public grpc_core::ConnectivityStateWatcherInterface {
  33  public:
  34   void Notify(grpc_connectivity_state new_state,
  35               const absl::Status& /* status */) override {
  36     GPR_ASSERT(new_state == GRPC_CHANNEL_SHUTDOWN);
  37   }
  38 };
  39
  40 static void* tag(intptr_t t) { return reinterpret_cast<void*>(t); }
  41
  42 static grpc_closure transport_op_cb;
  43
  44 static void do_nothing(void* /*arg*/, grpc_error_handle /*error*/) {}
  45
  46 void test_transport_op(grpc_channel* channel) {
  47   grpc_core::ExecCtx exec_ctx;
  48   grpc_transport_op* op = grpc_make_transport_op(nullptr);
  49   op->start_connectivity_watch = grpc_core::MakeOrphanable<Watcher>();
  50   grpc_channel_element* elem =
  51       grpc_channel_stack_element(grpc_channel_get_channel_stack(channel), 0);
  52   elem->filter->start_transport_op(elem, op);
  53
  54   GRPC_CLOSURE_INIT(&transport_op_cb, do_nothing, nullptr,
  55                     grpc_schedule_on_exec_ctx);
  56   op = grpc_make_transport_op(&transport_op_cb);
  57   elem->filter->start_transport_op(elem, op);
  58 }
  59
  60 int main(int argc, char** argv) {
  61   grpc_channel* chan;
  62   grpc_call* call;
  63   grpc_completion_queue* cq;
  64   cq_verifier* cqv;
  65   grpc_op ops[6];
  66   grpc_op* op;
  67   grpc_metadata_array initial_metadata_recv;
  68   grpc_metadata_array trailing_metadata_recv;
  69   grpc_status_code status;
  70   grpc_call_error error;
  71   grpc_slice details;
  72   char* peer;
  73
  74   grpc::testing::TestEnvironment env(argc, argv);
  75   grpc_init();
  76
  77   grpc_metadata_array_init(&initial_metadata_recv);
  78   grpc_metadata_array_init(&trailing_metadata_recv);
  79
  80   const char* error_message = "Rpc sent on a lame channel.";
  81   grpc_status_code error_code = GRPC_STATUS_ABORTED;
  82   chan = grpc_lame_client_channel_create("lampoon:national", error_code,
  83                                          error_message);
  84   GPR_ASSERT(chan);
  85
  86   test_transport_op(chan);
  87
  88   GPR_ASSERT(GRPC_CHANNEL_SHUTDOWN ==
  89              grpc_channel_check_connectivity_state(chan, 0));
  90
  91   cq = grpc_completion_queue_create_for_next(nullptr);
  92
  93   grpc_slice host = grpc_slice_from_static_string("anywhere");
  94   call =
  95       grpc_channel_create_call(chan, nullptr, GRPC_PROPAGATE_DEFAULTS, cq,
  96                                grpc_slice_from_static_string("/Foo"), &host,
  97                                grpc_timeout_seconds_to_deadline(100), nullptr);
  98   GPR_ASSERT(call);
  99   cqv = cq_verifier_create(cq);
 100
 101   memset(ops, 0, sizeof(ops));
 102   op = ops;
 103   op->op = GRPC_OP_SEND_INITIAL_METADATA;
 104   op->data.send_initial_metadata.count = 0;
 105   op->flags = 0;
 106   op->reserved = nullptr;
 107   op++;
 108   op->op = GRPC_OP_RECV_INITIAL_METADATA;
 109   op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
 110   op->flags = 0;
 111   op->reserved = nullptr;
 112   op++;
 113   error = grpc_call_start_batch(call, ops, static_cast<size_t>(op - ops),
 114                                 tag(1), nullptr);
 115   GPR_ASSERT(GRPC_CALL_OK == error);
 116
 117   /* the call should immediately fail */
 118   CQ_EXPECT_COMPLETION(cqv, tag(1), 0);
 119   cq_verify(cqv);
 120
 121   memset(ops, 0, sizeof(ops));
 122   op = ops;
 123   op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
 124   op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
 125   op->data.recv_status_on_client.status = &status;
 126   op->data.recv_status_on_client.status_details = &details;
 127   op->flags = 0;
 128   op->reserved = nullptr;
 129   op++;
 130   error = grpc_call_start_batch(call, ops, static_cast<size_t>(op - ops),
 131                                 tag(2), nullptr);
 132   GPR_ASSERT(GRPC_CALL_OK == error);
 133
 134   /* the call should immediately fail */
 135   CQ_EXPECT_COMPLETION(cqv, tag(2), 1);
 136   cq_verify(cqv);
 137
 138   peer = grpc_call_get_peer(call);
 139   GPR_ASSERT(strcmp(peer, "lampoon:national") == 0);
 140   gpr_free(peer);
 141
 142   GPR_ASSERT(status == error_code);
 143   GPR_ASSERT(grpc_slice_str_cmp(details, error_message) == 0);
 144
 145   grpc_call_unref(call);
 146   grpc_channel_destroy(chan);
 147   cq_verifier_destroy(cqv);
 148   grpc_completion_queue_destroy(cq);
 149
 150   grpc_metadata_array_destroy(&initial_metadata_recv);
 151   grpc_metadata_array_destroy(&trailing_metadata_recv);
 152   grpc_slice_unref(details);
 153
 154   grpc_shutdown();
 155
 156   return 0;
 157 }