Merge branch 'master' into new_scenario

1 files changed, 51 insertions, 9 deletions

diff --git a/src/core/lib/channel/message_size_filter.c b/src/core/lib/channel/message_size_filter.c
index f067a3a51c..1382f19945 100644
--- a/src/core/lib/channel/message_size_filter.c
+++ b/src/core/lib/channel/message_size_filter.c
@@ -38,6 +38,7 @@
 #include <grpc/support/log.h>
 #include <grpc/support/string_util.h>
 
+#include "src/core/ext/client_config/method_config.h"
 #include "src/core/lib/channel/channel_args.h"
 
 #define DEFAULT_MAX_SEND_MESSAGE_LENGTH -1  // Unlimited.
@@ -45,6 +46,8 @@
 #define DEFAULT_MAX_RECV_MESSAGE_LENGTH (4 * 1024 * 1024)
 
 typedef struct call_data {
+  int max_send_size;
+  int max_recv_size;
   // Receive closures are chained: we inject this closure as the
   // recv_message_ready up-call on transport_stream_op, and remember to
   // call our next_recv_message_ready member after handling it.
@@ -58,6 +61,8 @@ typedef struct call_data {
 typedef struct channel_data {
   int max_send_size;
   int max_recv_size;
+  // Method config table.
+  grpc_method_config_table* method_config_table;
 } channel_data;
 
 // Callback invoked when we receive a message.  Here we check the max
@@ -66,13 +71,12 @@ static void recv_message_ready(grpc_exec_ctx* exec_ctx, void* user_data,
                                grpc_error* error) {
   grpc_call_element* elem = user_data;
   call_data* calld = elem->call_data;
-  channel_data* chand = elem->channel_data;
-  if (*calld->recv_message != NULL && chand->max_recv_size >= 0 &&
-      (*calld->recv_message)->length > (size_t)chand->max_recv_size) {
+  if (*calld->recv_message != NULL && calld->max_recv_size >= 0 &&
+      (*calld->recv_message)->length > (size_t)calld->max_recv_size) {
     char* message_string;
     gpr_asprintf(&message_string,
                  "Received message larger than max (%u vs. %d)",
-                 (*calld->recv_message)->length, chand->max_recv_size);
+                 (*calld->recv_message)->length, calld->max_recv_size);
     grpc_error* new_error = grpc_error_set_int(
         GRPC_ERROR_CREATE(message_string), GRPC_ERROR_INT_GRPC_STATUS,
         GRPC_STATUS_INVALID_ARGUMENT);
@@ -93,13 +97,12 @@ static void start_transport_stream_op(grpc_exec_ctx* exec_ctx,
                                       grpc_call_element* elem,
                                       grpc_transport_stream_op* op) {
   call_data* calld = elem->call_data;
-  channel_data* chand = elem->channel_data;
   // Check max send message size.
-  if (op->send_message != NULL && chand->max_send_size >= 0 &&
-      op->send_message->length > (size_t)chand->max_send_size) {
+  if (op->send_message != NULL && calld->max_send_size >= 0 &&
+      op->send_message->length > (size_t)calld->max_send_size) {
     char* message_string;
     gpr_asprintf(&message_string, "Sent message larger than max (%u vs. %d)",
-                 op->send_message->length, chand->max_send_size);
+                 op->send_message->length, calld->max_send_size);
     gpr_slice message = gpr_slice_from_copied_string(message_string);
     gpr_free(message_string);
     grpc_call_element_send_close_with_message(
@@ -119,9 +122,37 @@ static void start_transport_stream_op(grpc_exec_ctx* exec_ctx,
 static grpc_error* init_call_elem(grpc_exec_ctx* exec_ctx,
                                   grpc_call_element* elem,
                                   grpc_call_element_args* args) {
+  channel_data* chand = elem->channel_data;
   call_data* calld = elem->call_data;
   calld->next_recv_message_ready = NULL;
   grpc_closure_init(&calld->recv_message_ready, recv_message_ready, elem);
+  // 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
+  // size to the receive limit.
+  calld->max_send_size = chand->max_send_size;
+  calld->max_recv_size = chand->max_recv_size;
+  if (chand->method_config_table != NULL) {
+    grpc_method_config* method_config =
+        grpc_method_config_table_get_method_config(chand->method_config_table,
+                                                   args->path);
+    if (method_config != NULL) {
+      const int32_t* max_request_message_bytes =
+          grpc_method_config_get_max_request_message_bytes(method_config);
+      if (max_request_message_bytes != NULL &&
+          (*max_request_message_bytes < calld->max_send_size ||
+           calld->max_send_size < 0)) {
+        calld->max_send_size = *max_request_message_bytes;
+      }
+      const int32_t* max_response_message_bytes =
+          grpc_method_config_get_max_response_message_bytes(method_config);
+      if (max_response_message_bytes != NULL &&
+          (*max_response_message_bytes < calld->max_recv_size ||
+           calld->max_recv_size < 0)) {
+        calld->max_recv_size = *max_response_message_bytes;
+      }
+    }
+  }
   return GRPC_ERROR_NONE;
 }
 
@@ -155,11 +186,22 @@ static void init_channel_elem(grpc_exec_ctx* exec_ctx,
           grpc_channel_arg_get_integer(&args->channel_args->args[i], options);
     }
   }
+  // Get method config table from channel args.
+  const grpc_arg* channel_arg =
+      grpc_channel_args_find(args->channel_args, GRPC_ARG_SERVICE_CONFIG);
+  if (channel_arg != NULL) {
+    GPR_ASSERT(channel_arg->type == GRPC_ARG_POINTER);
+    chand->method_config_table = grpc_method_config_table_ref(
+        (grpc_method_config_table*)channel_arg->value.pointer.p);
+  }
 }
 
 // Destructor for channel_data.
 static void destroy_channel_elem(grpc_exec_ctx* exec_ctx,
-                                 grpc_channel_element* elem) {}
+                                 grpc_channel_element* elem) {
+  channel_data* chand = elem->channel_data;
+  grpc_method_config_table_unref(chand->method_config_table);
+}
 
 const grpc_channel_filter grpc_message_size_filter = {
     start_transport_stream_op,