Separate stream compression from message-wise compression

1 files changed, 98 insertions, 0 deletions

diff --git a/src/core/lib/channel/channel_args.c b/src/core/lib/channel/channel_args.c
index 8fdef0bc64..02db798b5c 100644
--- a/src/core/lib/channel/channel_args.c
+++ b/src/core/lib/channel/channel_args.c
@@ -221,6 +221,21 @@ grpc_compression_algorithm grpc_channel_args_get_compression_algorithm(
   return GRPC_COMPRESS_NONE;
 }
 
+grpc_stream_compression_algorithm
+grpc_channel_args_get_stream_compression_algorithm(const grpc_channel_args *a) {
+  size_t i;
+  if (a == NULL) return 0;
+  for (i = 0; i < a->num_args; ++i) {
+    if (a->args[i].type == GRPC_ARG_INTEGER &&
+        !strcmp(GRPC_STREAM_COMPRESSION_CHANNEL_DEFAULT_ALGORITHM,
+                a->args[i].key)) {
+      return (grpc_stream_compression_algorithm)a->args[i].value.integer;
+      break;
+    }
+  }
+  return GRPC_STREAM_COMPRESS_NONE;
+}
+
 grpc_channel_args *grpc_channel_args_set_compression_algorithm(
     grpc_channel_args *a, grpc_compression_algorithm algorithm) {
   GPR_ASSERT(algorithm < GRPC_COMPRESS_ALGORITHMS_COUNT);
@@ -231,6 +246,16 @@ grpc_channel_args *grpc_channel_args_set_compression_algorithm(
   return grpc_channel_args_copy_and_add(a, &tmp, 1);
 }
 
+grpc_channel_args *grpc_channel_args_set_stream_compression_algorithm(
+    grpc_channel_args *a, grpc_stream_compression_algorithm algorithm) {
+  GPR_ASSERT(algorithm < GRPC_STREAM_COMPRESS_ALGORITHMS_COUNT);
+  grpc_arg tmp;
+  tmp.type = GRPC_ARG_INTEGER;
+  tmp.key = GRPC_STREAM_COMPRESSION_CHANNEL_DEFAULT_ALGORITHM;
+  tmp.value.integer = algorithm;
+  return grpc_channel_args_copy_and_add(a, &tmp, 1);
+}
+
 /** Returns 1 if the argument for compression algorithm's enabled states bitset
  * was found in \a a, returning the arg's value in \a states. Otherwise, returns
  * 0. */
@@ -251,6 +276,26 @@ static int find_compression_algorithm_states_bitset(const grpc_channel_args *a,
   return 0; /* GPR_FALSE */
 }
 
+/** Returns 1 if the argument for compression algorithm's enabled states bitset
+ * was found in \a a, returning the arg's value in \a states. Otherwise, returns
+ * 0. */
+static int find_stream_compression_algorithm_states_bitset(
+    const grpc_channel_args *a, int **states_arg) {
+  if (a != NULL) {
+    size_t i;
+    for (i = 0; i < a->num_args; ++i) {
+      if (a->args[i].type == GRPC_ARG_INTEGER &&
+          !strcmp(GRPC_STREAM_COMPRESSION_CHANNEL_ENABLED_ALGORITHMS_BITSET,
+                  a->args[i].key)) {
+        *states_arg = &a->args[i].value.integer;
+        **states_arg |= 0x1; /* forcefully enable support for no compression */
+        return 1;
+      }
+    }
+  }
+  return 0; /* GPR_FALSE */
+}
+
 grpc_channel_args *grpc_channel_args_compression_algorithm_set_state(
     grpc_exec_ctx *exec_ctx, grpc_channel_args **a,
     grpc_compression_algorithm algorithm, int state) {
@@ -292,6 +337,48 @@ grpc_channel_args *grpc_channel_args_compression_algorithm_set_state(
   return result;
 }
 
+grpc_channel_args *grpc_channel_args_stream_compression_algorithm_set_state(
+    grpc_exec_ctx *exec_ctx, grpc_channel_args **a,
+    grpc_stream_compression_algorithm algorithm, int state) {
+  int *states_arg = NULL;
+  grpc_channel_args *result = *a;
+  const int states_arg_found =
+      find_stream_compression_algorithm_states_bitset(*a, &states_arg);
+
+  if (grpc_channel_args_get_stream_compression_algorithm(*a) == algorithm &&
+      state == 0) {
+    char *algo_name = NULL;
+    GPR_ASSERT(grpc_stream_compression_algorithm_name(algorithm, &algo_name) !=
+               0);
+    gpr_log(GPR_ERROR,
+            "Tried to disable default stream compression algorithm '%s'. The "
+            "operation has been ignored.",
+            algo_name);
+  } else if (states_arg_found) {
+    if (state != 0) {
+      GPR_BITSET((unsigned *)states_arg, algorithm);
+    } else if (algorithm != GRPC_STREAM_COMPRESS_NONE) {
+      GPR_BITCLEAR((unsigned *)states_arg, algorithm);
+    }
+  } else {
+    /* create a new arg */
+    grpc_arg tmp;
+    tmp.type = GRPC_ARG_INTEGER;
+    tmp.key = GRPC_STREAM_COMPRESSION_CHANNEL_ENABLED_ALGORITHMS_BITSET;
+    /* all enabled by default */
+    tmp.value.integer = (1u << GRPC_STREAM_COMPRESS_ALGORITHMS_COUNT) - 1;
+    if (state != 0) {
+      GPR_BITSET((unsigned *)&tmp.value.integer, algorithm);
+    } else if (algorithm != GRPC_STREAM_COMPRESS_NONE) {
+      GPR_BITCLEAR((unsigned *)&tmp.value.integer, algorithm);
+    }
+    result = grpc_channel_args_copy_and_add(*a, &tmp, 1);
+    grpc_channel_args_destroy(exec_ctx, *a);
+    *a = result;
+  }
+  return result;
+}
+
 uint32_t grpc_channel_args_compression_algorithm_get_states(
     const grpc_channel_args *a) {
   int *states_arg;
@@ -302,6 +389,17 @@ uint32_t grpc_channel_args_compression_algorithm_get_states(
   }
 }
 
+uint32_t grpc_channel_args_stream_compression_algorithm_get_states(
+    const grpc_channel_args *a) {
+  int *states_arg;
+  if (find_stream_compression_algorithm_states_bitset(a, &states_arg)) {
+    return (uint32_t)*states_arg;
+  } else {
+    return (1u << GRPC_STREAM_COMPRESS_ALGORITHMS_COUNT) -
+           1; /* All algs. enabled */
+  }
+}
+
 grpc_channel_args *grpc_channel_args_set_socket_mutator(
     grpc_channel_args *a, grpc_socket_mutator *mutator) {
   grpc_arg tmp = grpc_socket_mutator_to_arg(mutator);