3 * Copyright 2015 gRPC authors.
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
9 * http://www.apache.org/licenses/LICENSE-2.0
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.
19 #include <grpc/support/port_platform.h>
21 #include <grpc/grpc.h>
25 #include <grpc/support/alloc.h>
26 #include <grpc/support/log.h>
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/static_metadata.h"
42 grpc_core::CallCombiner* call_combiner;
43 grpc_linked_mdelem status;
44 grpc_linked_mdelem details;
45 grpc_core::Atomic<bool> filled_metadata;
49 grpc_status_code error_code;
50 const char* error_message;
53 static void fill_metadata(grpc_call_element* elem, grpc_metadata_batch* mdb) {
54 CallData* calld = static_cast<CallData*>(elem->call_data);
55 bool expected = false;
56 if (!calld->filled_metadata.CompareExchangeStrong(
57 &expected, true, MemoryOrder::RELAXED, MemoryOrder::RELAXED)) {
60 ChannelData* chand = static_cast<ChannelData*>(elem->channel_data);
61 char tmp[GPR_LTOA_MIN_BUFSIZE];
62 gpr_ltoa(chand->error_code, tmp);
63 calld->status.md = grpc_mdelem_from_slices(
64 GRPC_MDSTR_GRPC_STATUS, grpc_core::UnmanagedMemorySlice(tmp));
65 calld->details.md = grpc_mdelem_from_slices(
66 GRPC_MDSTR_GRPC_MESSAGE,
67 grpc_core::UnmanagedMemorySlice(chand->error_message));
68 calld->status.prev = calld->details.next = nullptr;
69 calld->status.next = &calld->details;
70 calld->details.prev = &calld->status;
71 mdb->list.head = &calld->status;
72 mdb->list.tail = &calld->details;
74 mdb->deadline = GRPC_MILLIS_INF_FUTURE;
77 static void lame_start_transport_stream_op_batch(
78 grpc_call_element* elem, grpc_transport_stream_op_batch* op) {
79 CallData* calld = static_cast<CallData*>(elem->call_data);
80 if (op->recv_initial_metadata) {
82 op->payload->recv_initial_metadata.recv_initial_metadata);
83 } else if (op->recv_trailing_metadata) {
85 op->payload->recv_trailing_metadata.recv_trailing_metadata);
87 grpc_transport_stream_op_batch_finish_with_failure(
88 op, GRPC_ERROR_CREATE_FROM_STATIC_STRING("lame client channel"),
89 calld->call_combiner);
92 static void lame_get_channel_info(grpc_channel_element* elem,
93 const grpc_channel_info* channel_info) {}
95 static void lame_start_transport_op(grpc_channel_element* elem,
96 grpc_transport_op* op) {
97 if (op->on_connectivity_state_change) {
98 GPR_ASSERT(*op->connectivity_state != GRPC_CHANNEL_SHUTDOWN);
99 *op->connectivity_state = GRPC_CHANNEL_SHUTDOWN;
100 GRPC_CLOSURE_SCHED(op->on_connectivity_state_change, GRPC_ERROR_NONE);
102 if (op->send_ping.on_initiate != nullptr) {
104 op->send_ping.on_initiate,
105 GRPC_ERROR_CREATE_FROM_STATIC_STRING("lame client channel"));
107 if (op->send_ping.on_ack != nullptr) {
109 op->send_ping.on_ack,
110 GRPC_ERROR_CREATE_FROM_STATIC_STRING("lame client channel"));
112 GRPC_ERROR_UNREF(op->disconnect_with_error);
113 if (op->on_consumed != nullptr) {
114 GRPC_CLOSURE_SCHED(op->on_consumed, GRPC_ERROR_NONE);
118 static grpc_error* init_call_elem(grpc_call_element* elem,
119 const grpc_call_element_args* args) {
120 CallData* calld = static_cast<CallData*>(elem->call_data);
121 calld->call_combiner = args->call_combiner;
122 return GRPC_ERROR_NONE;
125 static void destroy_call_elem(grpc_call_element* elem,
126 const grpc_call_final_info* final_info,
127 grpc_closure* then_schedule_closure) {
128 GRPC_CLOSURE_SCHED(then_schedule_closure, GRPC_ERROR_NONE);
131 static grpc_error* init_channel_elem(grpc_channel_element* elem,
132 grpc_channel_element_args* args) {
133 GPR_ASSERT(args->is_first);
134 GPR_ASSERT(args->is_last);
135 return GRPC_ERROR_NONE;
138 static void destroy_channel_elem(grpc_channel_element* elem) {}
142 } // namespace grpc_core
144 const grpc_channel_filter grpc_lame_filter = {
145 grpc_core::lame_start_transport_stream_op_batch,
146 grpc_core::lame_start_transport_op,
147 sizeof(grpc_core::CallData),
148 grpc_core::init_call_elem,
149 grpc_call_stack_ignore_set_pollset_or_pollset_set,
150 grpc_core::destroy_call_elem,
151 sizeof(grpc_core::ChannelData),
152 grpc_core::init_channel_elem,
153 grpc_core::destroy_channel_elem,
154 grpc_core::lame_get_channel_info,
158 #define CHANNEL_STACK_FROM_CHANNEL(c) ((grpc_channel_stack*)((c) + 1))
160 grpc_channel* grpc_lame_client_channel_create(const char* target,
161 grpc_status_code error_code,
162 const char* error_message) {
163 grpc_core::ExecCtx exec_ctx;
164 grpc_channel_element* elem;
165 grpc_channel* channel =
166 grpc_channel_create(target, nullptr, GRPC_CLIENT_LAME_CHANNEL, nullptr);
167 elem = grpc_channel_stack_element(grpc_channel_get_channel_stack(channel), 0);
169 "grpc_lame_client_channel_create(target=%s, error_code=%d, "
171 3, (target, (int)error_code, error_message));
172 GPR_ASSERT(elem->filter == &grpc_lame_filter);
173 auto chand = static_cast<grpc_core::ChannelData*>(elem->channel_data);
174 chand->error_code = error_code;
175 chand->error_message = error_message;