aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/ext/filters/message_size
diff options
context:
space:
mode:
authorGravatar kpayson64 <kpayson@google.com>2018-07-25 10:51:37 -0700
committerGravatar kpayson64 <kpayson@google.com>2018-07-25 10:51:37 -0700
commitab9b61dfccfb1e14e102332ee2619a6ee12e8081 (patch)
treebff86803b427b642cf9d8fd4189ed1cce497db77 /src/core/ext/filters/message_size
parenta7bd899970b01198a4ba66b7f9fa47e658e506aa (diff)
Revert "Roll back filter changes"
Diffstat (limited to 'src/core/ext/filters/message_size')
-rw-r--r--src/core/ext/filters/message_size/message_size_filter.cc45
1 files changed, 43 insertions, 2 deletions
diff --git a/src/core/ext/filters/message_size/message_size_filter.cc b/src/core/ext/filters/message_size/message_size_filter.cc
index c7fc3f2e62..deb5ae70ec 100644
--- a/src/core/ext/filters/message_size/message_size_filter.cc
+++ b/src/core/ext/filters/message_size/message_size_filter.cc
@@ -99,10 +99,15 @@ struct call_data {
// recv_message_ready up-call on transport_stream_op, and remember to
// call our next_recv_message_ready member after handling it.
grpc_closure recv_message_ready;
+ grpc_closure recv_trailing_metadata_ready;
+ // The error caused by a message that is too large, or GRPC_ERROR_NONE
+ grpc_error* error;
// Used by recv_message_ready.
grpc_core::OrphanablePtr<grpc_core::ByteStream>* recv_message;
// Original recv_message_ready callback, invoked after our own.
grpc_closure* next_recv_message_ready;
+ // Original recv_trailing_metadata callback, invoked after our own.
+ grpc_closure* next_recv_trailing_metadata_ready;
};
struct channel_data {
@@ -130,12 +135,13 @@ static void recv_message_ready(void* user_data, grpc_error* error) {
grpc_error* new_error = grpc_error_set_int(
GRPC_ERROR_CREATE_FROM_COPIED_STRING(message_string),
GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_RESOURCE_EXHAUSTED);
+ GRPC_ERROR_UNREF(calld->error);
if (error == GRPC_ERROR_NONE) {
error = new_error;
} else {
error = grpc_error_add_child(error, new_error);
- GRPC_ERROR_UNREF(new_error);
}
+ calld->error = GRPC_ERROR_REF(error);
gpr_free(message_string);
} else {
GRPC_ERROR_REF(error);
@@ -144,6 +150,26 @@ static void recv_message_ready(void* user_data, grpc_error* error) {
GRPC_CLOSURE_RUN(calld->next_recv_message_ready, error);
}
+// Callback invoked on completion of recv_trailing_metadata
+// Notifies the recv_trailing_metadata batch of any message size failures
+static void recv_trailing_metadata_ready(void* user_data, grpc_error* error) {
+ grpc_call_element* elem = static_cast<grpc_call_element*>(user_data);
+ call_data* calld = static_cast<call_data*>(elem->call_data);
+ if (calld->error != GRPC_ERROR_NONE) {
+ if (error == GRPC_ERROR_NONE) {
+ error = GRPC_ERROR_REF(calld->error);
+ } else if (error != calld->error) {
+ error = grpc_error_add_child(error, GRPC_ERROR_REF(calld->error));
+ } else {
+ error = GRPC_ERROR_REF(error);
+ }
+ } else {
+ error = GRPC_ERROR_REF(error);
+ }
+ // Invoke the next callback.
+ GRPC_CLOSURE_RUN(calld->next_recv_trailing_metadata_ready, error);
+}
+
// Start transport stream op.
static void start_transport_stream_op_batch(
grpc_call_element* elem, grpc_transport_stream_op_batch* op) {
@@ -172,6 +198,13 @@ static void start_transport_stream_op_batch(
calld->recv_message = op->payload->recv_message.recv_message;
op->payload->recv_message.recv_message_ready = &calld->recv_message_ready;
}
+ // Inject callback for receiving trailing metadata.
+ if (op->recv_trailing_metadata) {
+ calld->next_recv_trailing_metadata_ready =
+ op->payload->recv_trailing_metadata.recv_trailing_metadata_ready;
+ op->payload->recv_trailing_metadata.recv_trailing_metadata_ready =
+ &calld->recv_trailing_metadata_ready;
+ }
// Chain to the next filter.
grpc_call_next_op(elem, op);
}
@@ -183,8 +216,13 @@ static grpc_error* init_call_elem(grpc_call_element* elem,
call_data* calld = static_cast<call_data*>(elem->call_data);
calld->call_combiner = args->call_combiner;
calld->next_recv_message_ready = nullptr;
+ calld->next_recv_trailing_metadata_ready = nullptr;
+ calld->error = GRPC_ERROR_NONE;
GRPC_CLOSURE_INIT(&calld->recv_message_ready, recv_message_ready, elem,
grpc_schedule_on_exec_ctx);
+ GRPC_CLOSURE_INIT(&calld->recv_trailing_metadata_ready,
+ recv_trailing_metadata_ready, elem,
+ grpc_schedule_on_exec_ctx);
// Get max sizes from channel data, then merge in per-method config values.
// Note: Per-method config is only available on the client, so we
// apply the max request size to the send limit and the max response
@@ -213,7 +251,10 @@ static grpc_error* init_call_elem(grpc_call_element* elem,
// Destructor for call_data.
static void destroy_call_elem(grpc_call_element* elem,
const grpc_call_final_info* final_info,
- grpc_closure* ignored) {}
+ grpc_closure* ignored) {
+ call_data* calld = (call_data*)elem->call_data;
+ GRPC_ERROR_UNREF(calld->error);
+}
static int default_size(const grpc_channel_args* args,
int without_minimal_stack) {