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 <grpc/support/port_platform.h>
20
21 #include <grpc/grpc.h>
22
23 #include <string.h>
24
25 #include <grpc/support/alloc.h>
26 #include <grpc/support/log.h>
27
28 #include "src/core/lib/channel/channel_stack.h"
29 #include "src/core/lib/gpr/string.h"
30 #include "src/core/lib/gprpp/atomic.h"
31 #include "src/core/lib/surface/api_trace.h"
32 #include "src/core/lib/surface/call.h"
33 #include "src/core/lib/surface/channel.h"
34 #include "src/core/lib/surface/lame_client.h"
35 #include "src/core/lib/transport/connectivity_state.h"
36 #include "src/core/lib/transport/static_metadata.h"
37
38 namespace grpc_core {
39
40 namespace {
41
42 struct ChannelData {
ChannelDatagrpc_core::__anonc7c8649b0111::ChannelData43 ChannelData() : state_tracker("lame_channel", GRPC_CHANNEL_SHUTDOWN) {}
~ChannelDatagrpc_core::__anonc7c8649b0111::ChannelData44 ~ChannelData() { GRPC_ERROR_UNREF(error); }
45
46 grpc_error* error = GRPC_ERROR_NONE;
47 Mutex mu;
48 ConnectivityStateTracker state_tracker;
49 };
50
51 struct CallData {
52 CallCombiner* call_combiner;
53 };
54
lame_start_transport_stream_op_batch(grpc_call_element * elem,grpc_transport_stream_op_batch * op)55 static void lame_start_transport_stream_op_batch(
56 grpc_call_element* elem, grpc_transport_stream_op_batch* op) {
57 CallData* calld = static_cast<CallData*>(elem->call_data);
58 ChannelData* chand = static_cast<ChannelData*>(elem->channel_data);
59 grpc_transport_stream_op_batch_finish_with_failure(
60 op, GRPC_ERROR_REF(chand->error), calld->call_combiner);
61 }
62
lame_get_channel_info(grpc_channel_element *,const grpc_channel_info *)63 static void lame_get_channel_info(grpc_channel_element* /*elem*/,
64 const grpc_channel_info* /*channel_info*/) {}
65
lame_start_transport_op(grpc_channel_element * elem,grpc_transport_op * op)66 static void lame_start_transport_op(grpc_channel_element* elem,
67 grpc_transport_op* op) {
68 ChannelData* chand = static_cast<ChannelData*>(elem->channel_data);
69 {
70 MutexLock lock(&chand->mu);
71 if (op->start_connectivity_watch != nullptr) {
72 chand->state_tracker.AddWatcher(op->start_connectivity_watch_state,
73 std::move(op->start_connectivity_watch));
74 }
75 if (op->stop_connectivity_watch != nullptr) {
76 chand->state_tracker.RemoveWatcher(op->stop_connectivity_watch);
77 }
78 }
79 if (op->send_ping.on_initiate != nullptr) {
80 ExecCtx::Run(DEBUG_LOCATION, op->send_ping.on_initiate,
81 GRPC_ERROR_CREATE_FROM_STATIC_STRING("lame client channel"));
82 }
83 if (op->send_ping.on_ack != nullptr) {
84 ExecCtx::Run(DEBUG_LOCATION, op->send_ping.on_ack,
85 GRPC_ERROR_CREATE_FROM_STATIC_STRING("lame client channel"));
86 }
87 GRPC_ERROR_UNREF(op->disconnect_with_error);
88 if (op->on_consumed != nullptr) {
89 ExecCtx::Run(DEBUG_LOCATION, op->on_consumed, GRPC_ERROR_NONE);
90 }
91 }
92
lame_init_call_elem(grpc_call_element * elem,const grpc_call_element_args * args)93 static grpc_error* lame_init_call_elem(grpc_call_element* elem,
94 const grpc_call_element_args* args) {
95 CallData* calld = static_cast<CallData*>(elem->call_data);
96 calld->call_combiner = args->call_combiner;
97 return GRPC_ERROR_NONE;
98 }
99
lame_destroy_call_elem(grpc_call_element *,const grpc_call_final_info *,grpc_closure * then_schedule_closure)100 static void lame_destroy_call_elem(grpc_call_element* /*elem*/,
101 const grpc_call_final_info* /*final_info*/,
102 grpc_closure* then_schedule_closure) {
103 ExecCtx::Run(DEBUG_LOCATION, then_schedule_closure, GRPC_ERROR_NONE);
104 }
105
lame_init_channel_elem(grpc_channel_element * elem,grpc_channel_element_args * args)106 static grpc_error* lame_init_channel_elem(grpc_channel_element* elem,
107 grpc_channel_element_args* args) {
108 GPR_ASSERT(args->is_first);
109 GPR_ASSERT(args->is_last);
110 new (elem->channel_data) ChannelData;
111 return GRPC_ERROR_NONE;
112 }
113
lame_destroy_channel_elem(grpc_channel_element * elem)114 static void lame_destroy_channel_elem(grpc_channel_element* elem) {
115 ChannelData* chand = static_cast<ChannelData*>(elem->channel_data);
116 chand->~ChannelData();
117 }
118
119 } // namespace
120
SetLameFilterError(grpc_channel_element * elem,grpc_error * error)121 void SetLameFilterError(grpc_channel_element* elem, grpc_error* error) {
122 GPR_ASSERT(elem->filter == &grpc_lame_filter);
123 auto chand = static_cast<grpc_core::ChannelData*>(elem->channel_data);
124 chand->error = error;
125 }
126
127 } // namespace grpc_core
128
129 const grpc_channel_filter grpc_lame_filter = {
130 grpc_core::lame_start_transport_stream_op_batch,
131 grpc_core::lame_start_transport_op,
132 sizeof(grpc_core::CallData),
133 grpc_core::lame_init_call_elem,
134 grpc_call_stack_ignore_set_pollset_or_pollset_set,
135 grpc_core::lame_destroy_call_elem,
136 sizeof(grpc_core::ChannelData),
137 grpc_core::lame_init_channel_elem,
138 grpc_core::lame_destroy_channel_elem,
139 grpc_core::lame_get_channel_info,
140 "lame-client",
141 };
142
143 #define CHANNEL_STACK_FROM_CHANNEL(c) ((grpc_channel_stack*)((c) + 1))
144
grpc_lame_client_channel_create(const char * target,grpc_status_code error_code,const char * error_message)145 grpc_channel* grpc_lame_client_channel_create(const char* target,
146 grpc_status_code error_code,
147 const char* error_message) {
148 grpc_core::ExecCtx exec_ctx;
149 grpc_channel_element* elem;
150 grpc_channel* channel =
151 grpc_channel_create(target, nullptr, GRPC_CLIENT_LAME_CHANNEL, nullptr);
152 elem = grpc_channel_stack_element(grpc_channel_get_channel_stack(channel), 0);
153 GRPC_API_TRACE(
154 "grpc_lame_client_channel_create(target=%s, error_code=%d, "
155 "error_message=%s)",
156 3, (target, (int)error_code, error_message));
157 grpc_core::SetLameFilterError(
158 elem, grpc_error_set_str(
159 grpc_error_set_int(
160 GRPC_ERROR_CREATE_FROM_STATIC_STRING("lame client channel"),
161 GRPC_ERROR_INT_GRPC_STATUS, error_code),
162 GRPC_ERROR_STR_GRPC_MESSAGE,
163 grpc_slice_from_static_string(error_message)));
164 return channel;
165 }
166