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 #define GRPC_ARG_LAME_FILTER_ERROR "grpc.lame_filter_error"
39
40 namespace grpc_core {
41
42 namespace {
43
44 struct ChannelData {
ChannelDatagrpc_core::__anon8271cedc0111::ChannelData45 explicit ChannelData(grpc_channel_element_args* args)
46 : state_tracker("lame_channel", GRPC_CHANNEL_SHUTDOWN) {
47 grpc_error_handle err = grpc_channel_args_find_pointer<grpc_error>(
48 args->channel_args, GRPC_ARG_LAME_FILTER_ERROR);
49 if (err != nullptr) error = GRPC_ERROR_REF(err);
50 }
51
~ChannelDatagrpc_core::__anon8271cedc0111::ChannelData52 ~ChannelData() { GRPC_ERROR_UNREF(error); }
53
54 grpc_error_handle error = GRPC_ERROR_NONE;
55 Mutex mu;
56 ConnectivityStateTracker state_tracker;
57 };
58
59 struct CallData {
60 CallCombiner* call_combiner;
61 };
62
lame_start_transport_stream_op_batch(grpc_call_element * elem,grpc_transport_stream_op_batch * op)63 static void lame_start_transport_stream_op_batch(
64 grpc_call_element* elem, grpc_transport_stream_op_batch* op) {
65 CallData* calld = static_cast<CallData*>(elem->call_data);
66 ChannelData* chand = static_cast<ChannelData*>(elem->channel_data);
67 grpc_transport_stream_op_batch_finish_with_failure(
68 op, GRPC_ERROR_REF(chand->error), calld->call_combiner);
69 }
70
lame_get_channel_info(grpc_channel_element *,const grpc_channel_info *)71 static void lame_get_channel_info(grpc_channel_element* /*elem*/,
72 const grpc_channel_info* /*channel_info*/) {}
73
lame_start_transport_op(grpc_channel_element * elem,grpc_transport_op * op)74 static void lame_start_transport_op(grpc_channel_element* elem,
75 grpc_transport_op* op) {
76 ChannelData* chand = static_cast<ChannelData*>(elem->channel_data);
77 {
78 MutexLock lock(&chand->mu);
79 if (op->start_connectivity_watch != nullptr) {
80 chand->state_tracker.AddWatcher(op->start_connectivity_watch_state,
81 std::move(op->start_connectivity_watch));
82 }
83 if (op->stop_connectivity_watch != nullptr) {
84 chand->state_tracker.RemoveWatcher(op->stop_connectivity_watch);
85 }
86 }
87 if (op->send_ping.on_initiate != nullptr) {
88 ExecCtx::Run(DEBUG_LOCATION, op->send_ping.on_initiate,
89 GRPC_ERROR_CREATE_FROM_STATIC_STRING("lame client channel"));
90 }
91 if (op->send_ping.on_ack != nullptr) {
92 ExecCtx::Run(DEBUG_LOCATION, op->send_ping.on_ack,
93 GRPC_ERROR_CREATE_FROM_STATIC_STRING("lame client channel"));
94 }
95 GRPC_ERROR_UNREF(op->disconnect_with_error);
96 if (op->on_consumed != nullptr) {
97 ExecCtx::Run(DEBUG_LOCATION, op->on_consumed, GRPC_ERROR_NONE);
98 }
99 }
100
lame_init_call_elem(grpc_call_element * elem,const grpc_call_element_args * args)101 static grpc_error_handle lame_init_call_elem(
102 grpc_call_element* elem, const grpc_call_element_args* args) {
103 CallData* calld = static_cast<CallData*>(elem->call_data);
104 calld->call_combiner = args->call_combiner;
105 return GRPC_ERROR_NONE;
106 }
107
lame_destroy_call_elem(grpc_call_element *,const grpc_call_final_info *,grpc_closure * then_schedule_closure)108 static void lame_destroy_call_elem(grpc_call_element* /*elem*/,
109 const grpc_call_final_info* /*final_info*/,
110 grpc_closure* then_schedule_closure) {
111 ExecCtx::Run(DEBUG_LOCATION, then_schedule_closure, GRPC_ERROR_NONE);
112 }
113
lame_init_channel_elem(grpc_channel_element * elem,grpc_channel_element_args * args)114 static grpc_error_handle lame_init_channel_elem(
115 grpc_channel_element* elem, grpc_channel_element_args* args) {
116 new (elem->channel_data) ChannelData(args);
117 return GRPC_ERROR_NONE;
118 }
119
lame_destroy_channel_elem(grpc_channel_element * elem)120 static void lame_destroy_channel_elem(grpc_channel_element* elem) {
121 ChannelData* chand = static_cast<ChannelData*>(elem->channel_data);
122 chand->~ChannelData();
123 }
124
125 // Channel arg vtable for a grpc_error_handle.
ErrorCopy(void * p)126 void* ErrorCopy(void* p) {
127 grpc_error_handle error = static_cast<grpc_error_handle>(p);
128 return GRPC_ERROR_REF(error);
129 }
ErrorDestroy(void * p)130 void ErrorDestroy(void* p) {
131 grpc_error_handle error = static_cast<grpc_error_handle>(p);
132 GRPC_ERROR_UNREF(error);
133 }
ErrorCompare(void * p,void * q)134 int ErrorCompare(void* p, void* q) { return GPR_ICMP(p, q); }
135 const grpc_arg_pointer_vtable kLameFilterErrorArgVtable = {
136 ErrorCopy, ErrorDestroy, ErrorCompare};
137
138 } // namespace
139
MakeLameClientErrorArg(grpc_error_handle error)140 grpc_arg MakeLameClientErrorArg(grpc_error_handle error) {
141 return grpc_channel_arg_pointer_create(
142 const_cast<char*>(GRPC_ARG_LAME_FILTER_ERROR), error,
143 &kLameFilterErrorArgVtable);
144 }
145
146 } // namespace grpc_core
147
148 const grpc_channel_filter grpc_lame_filter = {
149 grpc_core::lame_start_transport_stream_op_batch,
150 grpc_core::lame_start_transport_op,
151 sizeof(grpc_core::CallData),
152 grpc_core::lame_init_call_elem,
153 grpc_call_stack_ignore_set_pollset_or_pollset_set,
154 grpc_core::lame_destroy_call_elem,
155 sizeof(grpc_core::ChannelData),
156 grpc_core::lame_init_channel_elem,
157 grpc_core::lame_destroy_channel_elem,
158 grpc_core::lame_get_channel_info,
159 "lame-client",
160 };
161
162 #define CHANNEL_STACK_FROM_CHANNEL(c) ((grpc_channel_stack*)((c) + 1))
163
grpc_lame_client_channel_create(const char * target,grpc_status_code error_code,const char * error_message)164 grpc_channel* grpc_lame_client_channel_create(const char* target,
165 grpc_status_code error_code,
166 const char* error_message) {
167 grpc_core::ExecCtx exec_ctx;
168 GRPC_API_TRACE(
169 "grpc_lame_client_channel_create(target=%s, error_code=%d, "
170 "error_message=%s)",
171 3, (target, (int)error_code, error_message));
172 grpc_error_handle error = grpc_error_set_str(
173 grpc_error_set_int(
174 GRPC_ERROR_CREATE_FROM_STATIC_STRING("lame client channel"),
175 GRPC_ERROR_INT_GRPC_STATUS, error_code),
176 GRPC_ERROR_STR_GRPC_MESSAGE,
177 grpc_slice_from_static_string(error_message));
178 grpc_arg error_arg = grpc_core::MakeLameClientErrorArg(error);
179 grpc_channel_args args = {1, &error_arg};
180 grpc_channel* channel =
181 grpc_channel_create(target, &args, GRPC_CLIENT_LAME_CHANNEL, nullptr);
182 GRPC_ERROR_UNREF(error);
183 return channel;
184 }
185