Merge branch 'compression_incoming_checks' into compression_md_level_bis

BUILD

@@ -303,7 +303,7 @@ cc_library(
     "src/core/lib/channel/connected_channel.c",
     "src/core/lib/channel/http_client_filter.c",
     "src/core/lib/channel/http_server_filter.c",
-    "src/core/lib/compression/compression_algorithm.c",
+    "src/core/lib/compression/compression.c",
     "src/core/lib/compression/message_compress.c",
     "src/core/lib/debug/trace.c",
     "src/core/lib/http/format_request.c",
@@ -642,7 +642,7 @@ cc_library(
     "src/core/lib/channel/connected_channel.c",
     "src/core/lib/channel/http_client_filter.c",
     "src/core/lib/channel/http_server_filter.c",
-    "src/core/lib/compression/compression_algorithm.c",
+    "src/core/lib/compression/compression.c",
     "src/core/lib/compression/message_compress.c",
     "src/core/lib/debug/trace.c",
     "src/core/lib/http/format_request.c",
@@ -1335,7 +1335,7 @@ objc_library(
     "src/core/lib/channel/connected_channel.c",
     "src/core/lib/channel/http_client_filter.c",
     "src/core/lib/channel/http_server_filter.c",
-    "src/core/lib/compression/compression_algorithm.c",
+    "src/core/lib/compression/compression.c",
     "src/core/lib/compression/message_compress.c",
     "src/core/lib/debug/trace.c",
     "src/core/lib/http/format_request.c",

Makefile

@@ -2500,7 +2500,7 @@ LIBGRPC_SRC = \
     src/core/lib/channel/connected_channel.c \
     src/core/lib/channel/http_client_filter.c \
     src/core/lib/channel/http_server_filter.c \
-    src/core/lib/compression/compression_algorithm.c \
+    src/core/lib/compression/compression.c \
     src/core/lib/compression/message_compress.c \
     src/core/lib/debug/trace.c \
     src/core/lib/http/format_request.c \
@@ -2847,7 +2847,7 @@ LIBGRPC_UNSECURE_SRC = \
     src/core/lib/channel/connected_channel.c \
     src/core/lib/channel/http_client_filter.c \
     src/core/lib/channel/http_server_filter.c \
-    src/core/lib/compression/compression_algorithm.c \
+    src/core/lib/compression/compression.c \
     src/core/lib/compression/message_compress.c \
     src/core/lib/debug/trace.c \
     src/core/lib/http/format_request.c \

binding.gyp

@@ -571,7 +571,7 @@
         'src/core/lib/channel/connected_channel.c',
         'src/core/lib/channel/http_client_filter.c',
         'src/core/lib/channel/http_server_filter.c',
-        'src/core/lib/compression/compression_algorithm.c',
+        'src/core/lib/compression/compression.c',
         'src/core/lib/compression/message_compress.c',
         'src/core/lib/debug/trace.c',
         'src/core/lib/http/format_request.c',

build.yaml

@@ -230,7 +230,7 @@ filegroups:
   - src/core/lib/channel/connected_channel.c
   - src/core/lib/channel/http_client_filter.c
   - src/core/lib/channel/http_server_filter.c
-  - src/core/lib/compression/compression_algorithm.c
+  - src/core/lib/compression/compression.c
   - src/core/lib/compression/message_compress.c
   - src/core/lib/debug/trace.c
   - src/core/lib/http/format_request.c

config.m4

@@ -90,7 +90,7 @@ if test "$PHP_GRPC" != "no"; then
     src/core/lib/channel/connected_channel.c \
     src/core/lib/channel/http_client_filter.c \
     src/core/lib/channel/http_server_filter.c \
-    src/core/lib/compression/compression_algorithm.c \
+    src/core/lib/compression/compression.c \
     src/core/lib/compression/message_compress.c \
     src/core/lib/debug/trace.c \
     src/core/lib/http/format_request.c \

gRPC.podspec

@@ -340,7 +340,7 @@ Pod::Spec.new do |s|
                       'src/core/lib/channel/connected_channel.c',
                       'src/core/lib/channel/http_client_filter.c',
                       'src/core/lib/channel/http_server_filter.c',
-                      'src/core/lib/compression/compression_algorithm.c',
+                      'src/core/lib/compression/compression.c',
                       'src/core/lib/compression/message_compress.c',
                       'src/core/lib/debug/trace.c',
                       'src/core/lib/http/format_request.c',

grpc.def

@@ -34,6 +34,11 @@ EXPORTS
     census_view_reset
     grpc_compression_algorithm_parse
     grpc_compression_algorithm_name
+    grpc_compression_algorithm_for_level
+    grpc_compression_options_init
+    grpc_compression_options_enable_algorithm
+    grpc_compression_options_disable_algorithm
+    grpc_compression_options_is_algorithm_enabled
     grpc_metadata_array_init
     grpc_metadata_array_destroy
     grpc_call_details_init

grpc.gemspec

@@ -319,7 +319,7 @@ Gem::Specification.new do |s|
   s.files += %w( src/core/lib/channel/connected_channel.c )
   s.files += %w( src/core/lib/channel/http_client_filter.c )
   s.files += %w( src/core/lib/channel/http_server_filter.c )
-  s.files += %w( src/core/lib/compression/compression_algorithm.c )
+  s.files += %w( src/core/lib/compression/compression.c )
   s.files += %w( src/core/lib/compression/message_compress.c )
   s.files += %w( src/core/lib/debug/trace.c )
   s.files += %w( src/core/lib/http/format_request.c )

include/grpc/compression.h

@@ -51,10 +51,32 @@ GRPCAPI int grpc_compression_algorithm_parse(
     grpc_compression_algorithm *algorithm);
 
 /** Updates \a name with the encoding name corresponding to a valid \a
- * algorithm.  Returns 1 upon success, 0 otherwise. */
+ * algorithm. Note that \a name is statically allocated and must *not* be freed.
+ * Returns 1 upon success, 0 otherwise. */
 GRPCAPI int grpc_compression_algorithm_name(
     grpc_compression_algorithm algorithm, char **name);
 
+/** Returns the compression algorithm corresponding to \a level for the
+ * compression algorithms encoded in the \a accepted_encodings bitset.
+ *
+ * It abort()s for unknown levels . */
+GRPCAPI grpc_compression_algorithm grpc_compression_algorithm_for_level(
+    grpc_compression_level level, uint32_t accepted_encodings);
+
+GRPCAPI void grpc_compression_options_init(grpc_compression_options *opts);
+
+/** Mark \a algorithm as enabled in \a opts. */
+GRPCAPI void grpc_compression_options_enable_algorithm(
+    grpc_compression_options *opts, grpc_compression_algorithm algorithm);
+
+/** Mark \a algorithm as disabled in \a opts. */
+GRPCAPI void grpc_compression_options_disable_algorithm(
+    grpc_compression_options *opts, grpc_compression_algorithm algorithm);
+
+/** Returns true if \a algorithm is marked as enabled in \a opts. */
+GRPCAPI int grpc_compression_options_is_algorithm_enabled(
+    const grpc_compression_options *opts, grpc_compression_algorithm algorithm);
+
 #ifdef __cplusplus
 }
 #endif

include/grpc/impl/codegen/compression_types.h

@@ -74,6 +74,32 @@ typedef enum {
   GRPC_COMPRESS_LEVEL_COUNT
 } grpc_compression_level;
 
+typedef struct grpc_compression_options {
+  /** All algs are enabled by default. This option corresponds to the channel
+   * argument key behind \a GRPC_COMPRESSION_CHANNEL_ENABLED_ALGORITHMS_BITSET
+   */
+  uint32_t enabled_algorithms_bitset;
+
+  /** The default channel compression level. It'll be used in the absence of
+   * call specific settings. This option corresponds to the channel argument key
+   * behind \a GRPC_COMPRESSION_CHANNEL_DEFAULT_LEVEL. If present, takes
+   * precedence over \a default_algorithm.
+   * TODO(dgq): currently only available for server channels. */
+  struct {
+    bool is_set;
+    grpc_compression_level level;
+  } default_level;
+
+  /** The default channel compression algorithm. It'll be used in the absence of
+   * call specific settings. This option corresponds to the channel argument key
+   * behind \a GRPC_COMPRESSION_CHANNEL_DEFAULT_ALGORITHM. */
+  struct {
+    bool is_set;
+    grpc_compression_algorithm algorithm;
+  } default_algorithm;
+
+} grpc_compression_options;
+
 #ifdef __cplusplus
 }
 #endif

package.xml

@@ -326,7 +326,7 @@
     <file baseinstalldir="/" name="src/core/lib/channel/connected_channel.c" role="src" />
     <file baseinstalldir="/" name="src/core/lib/channel/http_client_filter.c" role="src" />
     <file baseinstalldir="/" name="src/core/lib/channel/http_server_filter.c" role="src" />
-    <file baseinstalldir="/" name="src/core/lib/compression/compression_algorithm.c" role="src" />
+    <file baseinstalldir="/" name="src/core/lib/compression/compression.c" role="src" />
     <file baseinstalldir="/" name="src/core/lib/compression/message_compress.c" role="src" />
     <file baseinstalldir="/" name="src/core/lib/debug/trace.c" role="src" />
     <file baseinstalldir="/" name="src/core/lib/http/format_request.c" role="src" />

src/core/lib/channel/channel_args.c

@@ -35,6 +35,7 @@
 #include <grpc/grpc.h>
 #include "src/core/lib/support/string.h"
 
+#include <grpc/compression.h>
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
 #include <grpc/support/string_util.h>
@@ -180,6 +181,7 @@ grpc_compression_algorithm grpc_channel_args_get_compression_algorithm(
 
 grpc_channel_args *grpc_channel_args_set_compression_algorithm(
     grpc_channel_args *a, grpc_compression_algorithm algorithm) {
+  GPR_ASSERT(algorithm < GRPC_COMPRESS_ALGORITHMS_COUNT);
   grpc_arg tmp;
   tmp.type = GRPC_ARG_INTEGER;
   tmp.key = GRPC_COMPRESSION_CHANNEL_DEFAULT_ALGORITHM;
@@ -214,7 +216,15 @@ grpc_channel_args *grpc_channel_args_compression_algorithm_set_state(
   const int states_arg_found =
       find_compression_algorithm_states_bitset(*a, &states_arg);
 
-  if (states_arg_found) {
+  if (grpc_channel_args_get_compression_algorithm(*a) == algorithm &&
+      state == 0) {
+    char *algo_name = NULL;
+    GPR_ASSERT(grpc_compression_algorithm_name(algorithm, &algo_name) != 0);
+    gpr_log(GPR_ERROR,
+            "Tried to disable default 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_COMPRESS_NONE) {

src/core/lib/compression/compression_algorithm.c → src/core/lib/compression/compression.c

@@ -124,3 +124,81 @@ grpc_mdelem *grpc_compression_encoding_mdelem(
   }
   return NULL;
 }
+
+void grpc_compression_options_init(grpc_compression_options *opts) {
+  memset(opts, 0, sizeof(*opts));
+  /* all enabled by default */
+  opts->enabled_algorithms_bitset = (1u << GRPC_COMPRESS_ALGORITHMS_COUNT) - 1;
+}
+
+void grpc_compression_options_enable_algorithm(
+    grpc_compression_options *opts, grpc_compression_algorithm algorithm) {
+  GPR_BITSET(&opts->enabled_algorithms_bitset, algorithm);
+}
+
+void grpc_compression_options_disable_algorithm(
+    grpc_compression_options *opts, grpc_compression_algorithm algorithm) {
+  GPR_BITCLEAR(&opts->enabled_algorithms_bitset, algorithm);
+}
+
+int grpc_compression_options_is_algorithm_enabled(
+    const grpc_compression_options *opts,
+    grpc_compression_algorithm algorithm) {
+  return GPR_BITGET(opts->enabled_algorithms_bitset, algorithm);
+}
+
+/* TODO(dgq): Add the ability to specify parameters to the individual
+ * compression algorithms */
+grpc_compression_algorithm grpc_compression_algorithm_for_level(
+    grpc_compression_level level, uint32_t accepted_encodings) {
+  GRPC_API_TRACE("grpc_compression_algorithm_for_level(level=%d)", 1,
+                 ((int)level));
+  if (level > GRPC_COMPRESS_LEVEL_HIGH) {
+    gpr_log(GPR_ERROR, "Unknown compression level %d.", (int)level);
+    abort();
+  }
+
+  const size_t num_supported =
+      GPR_BITCOUNT(accepted_encodings) - 1; /* discard NONE */
+  if (level == GRPC_COMPRESS_LEVEL_NONE || num_supported == 0) {
+    return GRPC_COMPRESS_NONE;
+  }
+
+  GPR_ASSERT(level > 0);
+
+  /* Establish a "ranking" or compression algorithms in increasing order of
+   * compression.
+   * This is simplistic and we will probably want to introduce other dimensions
+   * in the future (cpu/memory cost, etc). */
+  const grpc_compression_algorithm algos_ranking[] = {GRPC_COMPRESS_GZIP,
+                                                      GRPC_COMPRESS_DEFLATE};
+
+  /* intersect algos_ranking with the supported ones keeping the ranked order */
+  grpc_compression_algorithm
+      sorted_supported_algos[GRPC_COMPRESS_ALGORITHMS_COUNT];
+  size_t algos_supported_idx = 0;
+  for (size_t i = 0; i < GPR_ARRAY_SIZE(algos_ranking); i++) {
+    const grpc_compression_algorithm alg = algos_ranking[i];
+    for (size_t j = 0; j < num_supported; j++) {
+      if (GPR_BITGET(accepted_encodings, alg) == 1) {
+        /* if \a alg in supported */
+        sorted_supported_algos[algos_supported_idx++] = alg;
+        break;
+      }
+    }
+    if (algos_supported_idx == num_supported) break;
+  }
+
+  switch (level) {
+    case GRPC_COMPRESS_LEVEL_NONE:
+      abort(); /* should have been handled already */
+    case GRPC_COMPRESS_LEVEL_LOW:
+      return sorted_supported_algos[0];
+    case GRPC_COMPRESS_LEVEL_MED:
+      return sorted_supported_algos[num_supported / 2];
+    case GRPC_COMPRESS_LEVEL_HIGH:
+      return sorted_supported_algos[num_supported - 1];
+    default:
+      abort();
+  };
+}

src/core/lib/surface/call.c

@@ -40,6 +40,7 @@
 #include <grpc/grpc.h>
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
+#include <grpc/support/slice.h>
 #include <grpc/support/string_util.h>
 #include <grpc/support/useful.h>
 
@@ -52,7 +53,9 @@
 #include "src/core/lib/surface/call.h"
 #include "src/core/lib/surface/channel.h"
 #include "src/core/lib/surface/completion_queue.h"
+#include "src/core/lib/transport/metadata.h"
 #include "src/core/lib/transport/static_metadata.h"
+#include "src/core/lib/transport/transport.h"
 
 /** The maximum number of concurrent batches possible.
     Based upon the maximum number of individually queueable ops in the batch
@@ -238,6 +241,9 @@ static void execute_op(grpc_exec_ctx *exec_ctx, grpc_call *call,
 static grpc_call_error cancel_with_status(grpc_exec_ctx *exec_ctx, grpc_call *c,
                                           grpc_status_code status,
                                           const char *description);
+static grpc_call_error close_with_status(grpc_exec_ctx *exec_ctx, grpc_call *c,
+                                         grpc_status_code status,
+                                         const char *description);
 static void destroy_call(grpc_exec_ctx *exec_ctx, void *call_stack,
                          bool success);
 static void receiving_slice_ready(grpc_exec_ctx *exec_ctx, void *bctlp,
@@ -408,8 +414,8 @@ static void set_status_code(grpc_call *call, status_source source,
   /* TODO(ctiller): what to do about the flush that was previously here */
 }
 
-static void set_compression_algorithm(grpc_call *call,
-                                      grpc_compression_algorithm algo) {
+static void set_incoming_compression_algorithm(
+    grpc_call *call, grpc_compression_algorithm algo) {
   GPR_ASSERT(algo < GRPC_COMPRESS_ALGORITHMS_COUNT);
   call->incoming_compression_algorithm = algo;
 }
@@ -425,61 +431,8 @@ grpc_compression_algorithm grpc_call_test_only_get_compression_algorithm(
 
 static grpc_compression_algorithm compression_algorithm_for_level_locked(
     grpc_call *call, grpc_compression_level level) {
-  /* Establish a "ranking" or compression algorithms in increasing order of
-   * compression.
-   * This is simplistic and we will probably want to introduce other
-   * dimensions
-   * in the future (cpu/memory cost, etc). */
-  const grpc_compression_algorithm algos_ranking[] = {GRPC_COMPRESS_GZIP,
-                                                      GRPC_COMPRESS_DEFLATE};
-  const uint32_t accepted_encodings = call->encodings_accepted_by_peer;
-  if (level > GRPC_COMPRESS_LEVEL_HIGH) {
-    extern int grpc_compression_trace;
-    if (grpc_compression_trace) {
-      gpr_log(GPR_ERROR,
-              "Unknown compression level %d. Compression will be disabled.",
-              (int)level);
-    }
-    return GRPC_COMPRESS_NONE;
-  }
-
-  const size_t num_supported =
-      GPR_BITCOUNT(accepted_encodings) - 1; /* discard NONE */
-  if (level == GRPC_COMPRESS_LEVEL_NONE || num_supported == 0) {
-    return GRPC_COMPRESS_NONE;
-  }
-
-  GPR_ASSERT(level > 0);
-
-  /* intersect algos_ranking with the supported ones keeping the ranked order
-   */
-  grpc_compression_algorithm
-      sorted_supported_algos[GRPC_COMPRESS_ALGORITHMS_COUNT];
-  size_t algos_supported_idx = 0;
-  for (size_t i = 0; i < GPR_ARRAY_SIZE(algos_ranking); i++) {
-    const grpc_compression_algorithm alg = algos_ranking[i];
-    for (size_t j = 0; j < num_supported; j++) {
-      if (GPR_BITGET(accepted_encodings, alg) == 1) {
-        /* if \a alg in supported */
-        sorted_supported_algos[algos_supported_idx++] = alg;
-        break;
-      }
-    }
-    if (algos_supported_idx == num_supported) break;
-  }
-
-  switch (level) {
-    case GRPC_COMPRESS_LEVEL_NONE:
-      abort(); /* should have been handled already */
-    case GRPC_COMPRESS_LEVEL_LOW:
-      return sorted_supported_algos[0];
-    case GRPC_COMPRESS_LEVEL_MED:
-      return sorted_supported_algos[num_supported / 2];
-    case GRPC_COMPRESS_LEVEL_HIGH:
-      return sorted_supported_algos[num_supported - 1];
-    default:
-      abort();
-  };
+  return grpc_compression_algorithm_for_level(level,
+                                              call->encodings_accepted_by_peer);
 }
 
 uint32_t grpc_call_test_only_get_message_flags(grpc_call *call) {
@@ -793,48 +746,102 @@ grpc_call_error grpc_call_cancel_with_status(grpc_call *c,
   return r;
 }
 
-typedef struct cancel_closure {
+typedef struct termination_closure {
   grpc_closure closure;
   grpc_call *call;
   grpc_status_code status;
-} cancel_closure;
+  gpr_slice optional_message;
+  grpc_closure *op_closure;
+  enum { TC_CANCEL, TC_CLOSE } type;
+} termination_closure;
+
+static void done_termination(grpc_exec_ctx *exec_ctx, void *tcp, bool success) {
+  termination_closure *tc = tcp;
+  if (tc->type == TC_CANCEL) {
+    GRPC_CALL_INTERNAL_UNREF(exec_ctx, tc->call, "cancel");
+  }
+  if (tc->type == TC_CLOSE) {
+    GRPC_CALL_INTERNAL_UNREF(exec_ctx, tc->call, "close");
+  }
+  gpr_slice_unref(tc->optional_message);
+  if (tc->op_closure != NULL) {
+    grpc_exec_ctx_enqueue(exec_ctx, tc->op_closure, true, NULL);
+  }
+  gpr_free(tc);
+}
 
-static void done_cancel(grpc_exec_ctx *exec_ctx, void *ccp, bool success) {
-  cancel_closure *cc = ccp;
-  GRPC_CALL_INTERNAL_UNREF(exec_ctx, cc->call, "cancel");
-  gpr_free(cc);
+static void send_cancel(grpc_exec_ctx *exec_ctx, void *tcp, bool success) {
+  grpc_transport_stream_op op;
+  termination_closure *tc = tcp;
+  memset(&op, 0, sizeof(op));
+  op.cancel_with_status = tc->status;
+  /* reuse closure to catch completion */
+  grpc_closure_init(&tc->closure, done_termination, tc);
+  op.on_complete = &tc->closure;
+  execute_op(exec_ctx, tc->call, &op);
 }
 
-static void send_cancel(grpc_exec_ctx *exec_ctx, void *ccp, bool success) {
+static void send_close(grpc_exec_ctx *exec_ctx, void *tcp, bool success) {
   grpc_transport_stream_op op;
-  cancel_closure *cc = ccp;
+  termination_closure *tc = tcp;
   memset(&op, 0, sizeof(op));
-  op.cancel_with_status = cc->status;
+  tc->optional_message = gpr_slice_ref(tc->optional_message);
+  grpc_transport_stream_op_add_close(&op, tc->status, &tc->optional_message);
   /* reuse closure to catch completion */
-  grpc_closure_init(&cc->closure, done_cancel, cc);
-  op.on_complete = &cc->closure;
-  execute_op(exec_ctx, cc->call, &op);
+  grpc_closure_init(&tc->closure, done_termination, tc);
+  tc->op_closure = op.on_complete;
+  op.on_complete = &tc->closure;
+  execute_op(exec_ctx, tc->call, &op);
+}
+
+static grpc_call_error terminate_with_status(grpc_exec_ctx *exec_ctx,
+                                             termination_closure *tc) {
+  grpc_mdstr *details = NULL;
+  if (GPR_SLICE_LENGTH(tc->optional_message) > 0) {
+    tc->optional_message = gpr_slice_ref(tc->optional_message);
+    details = grpc_mdstr_from_slice(tc->optional_message);
+  }
+
+  set_status_code(tc->call, STATUS_FROM_API_OVERRIDE, (uint32_t)tc->status);
+  set_status_details(tc->call, STATUS_FROM_API_OVERRIDE, details);
+
+  if (tc->type == TC_CANCEL) {
+    grpc_closure_init(&tc->closure, send_cancel, tc);
+    GRPC_CALL_INTERNAL_REF(tc->call, "cancel");
+  } else if (tc->type == TC_CLOSE) {
+    grpc_closure_init(&tc->closure, send_close, tc);
+    GRPC_CALL_INTERNAL_REF(tc->call, "close");
+  }
+  grpc_exec_ctx_enqueue(exec_ctx, &tc->closure, true, NULL);
+  return GRPC_CALL_OK;
 }
 
 static grpc_call_error cancel_with_status(grpc_exec_ctx *exec_ctx, grpc_call *c,
                                           grpc_status_code status,
                                           const char *description) {
-  grpc_mdstr *details =
-      description ? grpc_mdstr_from_string(description) : NULL;
-  cancel_closure *cc = gpr_malloc(sizeof(*cc));
-
+  termination_closure *tc = gpr_malloc(sizeof(*tc));
+  memset(tc, 0, sizeof(termination_closure));
+  tc->type = TC_CANCEL;
+  tc->call = c;
+  tc->optional_message = gpr_slice_from_copied_string(description);
   GPR_ASSERT(status != GRPC_STATUS_OK);
+  tc->status = status;
 
-  set_status_code(c, STATUS_FROM_API_OVERRIDE, (uint32_t)status);
-  set_status_details(c, STATUS_FROM_API_OVERRIDE, details);
+  return terminate_with_status(exec_ctx, tc);
+}
 
-  grpc_closure_init(&cc->closure, send_cancel, cc);
-  cc->call = c;
-  cc->status = status;
-  GRPC_CALL_INTERNAL_REF(c, "cancel");
-  grpc_exec_ctx_enqueue(exec_ctx, &cc->closure, true, NULL);
+static grpc_call_error close_with_status(grpc_exec_ctx *exec_ctx, grpc_call *c,
+                                         grpc_status_code status,
+                                         const char *description) {
+  termination_closure *tc = gpr_malloc(sizeof(*tc));
+  memset(tc, 0, sizeof(termination_closure));
+  tc->type = TC_CLOSE;
+  tc->call = c;
+  tc->optional_message = gpr_slice_from_copied_string(description);
+  GPR_ASSERT(status != GRPC_STATUS_OK);
+  tc->status = status;
 
-  return GRPC_CALL_OK;
+  return terminate_with_status(exec_ctx, tc);
 }
 
 static void execute_op(grpc_exec_ctx *exec_ctx, grpc_call *call,
@@ -976,7 +983,7 @@ static grpc_mdelem *recv_initial_filter(void *callp, grpc_mdelem *elem) {
     return NULL;
   } else if (elem->key == GRPC_MDSTR_GRPC_ENCODING) {
     GPR_TIMER_BEGIN("incoming_compression_algorithm", 0);
-    set_compression_algorithm(call, decode_compression(elem));
+    set_incoming_compression_algorithm(call, decode_compression(elem));
     GPR_TIMER_END("incoming_compression_algorithm", 0);
     return NULL;
   } else if (elem->key == GRPC_MDSTR_GRPC_ACCEPT_ENCODING) {
@@ -1170,6 +1177,56 @@ static void receiving_stream_ready(grpc_exec_ctx *exec_ctx, void *bctlp,
   }
 }
 
+static void validate_filtered_metadata(grpc_exec_ctx *exec_ctx,
+                                       batch_control *bctl) {
+  grpc_call *call = bctl->call;
+  /* validate call->incoming_compression_algorithm */
+  if (call->incoming_compression_algorithm != GRPC_COMPRESS_NONE) {
+    const grpc_compression_algorithm algo =
+        call->incoming_compression_algorithm;
+    char *error_msg = NULL;
+    const grpc_compression_options compression_options =
+        grpc_channel_compression_options(call->channel);
+    /* check if algorithm is known */
+    if (algo >= GRPC_COMPRESS_ALGORITHMS_COUNT) {
+      gpr_asprintf(&error_msg, "Invalid compression algorithm value '%d'.",
+                   algo);
+      gpr_log(GPR_ERROR, error_msg);
+      close_with_status(exec_ctx, call, GRPC_STATUS_UNIMPLEMENTED, error_msg);
+    } else if (grpc_compression_options_is_algorithm_enabled(
+                   &compression_options, algo) == 0) {
+      /* check if algorithm is supported by current channel config */
+      char *algo_name;
+      grpc_compression_algorithm_name(algo, &algo_name);
+      gpr_asprintf(&error_msg, "Compression algorithm '%s' is disabled.",
+                   algo_name);
+      gpr_log(GPR_ERROR, error_msg);
+      close_with_status(exec_ctx, call, GRPC_STATUS_UNIMPLEMENTED, error_msg);
+    } else {
+      call->incoming_compression_algorithm = algo;
+    }
+    gpr_free(error_msg);
+  }
+
+  /* make sure the received grpc-encoding is amongst the ones listed in
+   * grpc-accept-encoding */
+  GPR_ASSERT(call->encodings_accepted_by_peer != 0);
+  if (!GPR_BITGET(call->encodings_accepted_by_peer,
+                  call->incoming_compression_algorithm)) {
+    extern int grpc_compression_trace;
+    if (grpc_compression_trace) {
+      char *algo_name;
+      grpc_compression_algorithm_name(call->incoming_compression_algorithm,
+                                      &algo_name);
+      gpr_log(GPR_ERROR,
+              "Compression algorithm (grpc-encoding = '%s') not present in "
+              "the bitset of accepted encodings (grpc-accept-encodings: "
+              "'0x%x')",
+              algo_name, call->encodings_accepted_by_peer);
+    }
+  }
+}
+
 static void receiving_initial_metadata_ready(grpc_exec_ctx *exec_ctx,
                                              void *bctlp, bool success) {
   batch_control *bctl = bctlp;
@@ -1184,24 +1241,10 @@ static void receiving_initial_metadata_ready(grpc_exec_ctx *exec_ctx,
         &call->metadata_batch[1 /* is_receiving */][0 /* is_trailing */];
     grpc_metadata_batch_filter(md, recv_initial_filter, call);
 
-    /* make sure the received grpc-encoding is amongst the ones listed in
-     * grpc-accept-encoding */
-
-    GPR_ASSERT(call->encodings_accepted_by_peer != 0);
-    if (!GPR_BITGET(call->encodings_accepted_by_peer,
-                    call->incoming_compression_algorithm)) {
-      extern int grpc_compression_trace;
-      if (grpc_compression_trace) {
-        char *algo_name;
-        grpc_compression_algorithm_name(call->incoming_compression_algorithm,
-                                        &algo_name);
-        gpr_log(GPR_ERROR,
-                "Compression algorithm (grpc-encoding = '%s') not present in "
-                "the bitset of accepted encodings (grpc-accept-encodings: "
-                "'0x%x')",
-                algo_name, call->encodings_accepted_by_peer);
-      }
-    }
+    GPR_TIMER_BEGIN("validate_filtered_metadata", 0);
+    validate_filtered_metadata(exec_ctx, bctl);
+    GPR_TIMER_END("validate_filtered_metadata", 0);
+
     if (gpr_time_cmp(md->deadline, gpr_inf_future(md->deadline.clock_type)) !=
             0 &&
         !call->is_client) {
@@ -1356,8 +1399,12 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx,
                   .compression_level;
           level_set = true;
         } else {
-          level_set = grpc_channel_default_compression_level(
-              call->channel, &effective_compression_level);
+          const grpc_compression_options copts =
+              grpc_channel_compression_options(call->channel);
+          level_set = copts.default_level.is_set;
+          if (level_set) {
+            effective_compression_level = copts.default_level.level;
+          }
         }
         if (level_set) {
           const grpc_compression_algorithm calgo =

src/core/lib/surface/channel.c

@@ -36,16 +36,17 @@
 #include <stdlib.h>
 #include <string.h>
 
+#include <grpc/compression.h>
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
 #include <grpc/support/string_util.h>
 
+#include "src/core/lib/channel/channel_args.h"
 #include "src/core/lib/iomgr/iomgr.h"
 #include "src/core/lib/support/string.h"
 #include "src/core/lib/surface/api_trace.h"
 #include "src/core/lib/surface/call.h"
 #include "src/core/lib/surface/channel_init.h"
-#include "src/core/lib/surface/init.h"
 #include "src/core/lib/transport/static_metadata.h"
 
 /** Cache grpc-status: X mdelems for X = 0..NUM_CACHED_STATUS_ELEMS.
@@ -64,17 +65,13 @@ typedef struct registered_call {
 struct grpc_channel {
   int is_client;
   uint32_t max_message_length;
+  grpc_compression_options compression_options;
   grpc_mdelem *default_authority;
 
   gpr_mu registered_call_mu;
   registered_call *registered_calls;
 
   char *target;
-
-  struct {
-    bool is_set;
-    grpc_compression_level default_compression_level;
-  } maybe_default_compression_level;
 };
 
 #define CHANNEL_STACK_FROM_CHANNEL(c) ((grpc_channel_stack *)((c) + 1))
@@ -117,6 +114,7 @@ grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target,
   channel->registered_calls = NULL;
 
   channel->max_message_length = DEFAULT_MAX_MESSAGE_LENGTH;
+  grpc_compression_options_init(&channel->compression_options);
   if (args) {
     for (size_t i = 0; i < args->num_args; i++) {
       if (0 == strcmp(args->args[i].key, GRPC_ARG_MAX_MESSAGE_LENGTH)) {
@@ -159,11 +157,25 @@ grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target,
         }
       } else if (0 == strcmp(args->args[i].key,
                              GRPC_COMPRESSION_CHANNEL_DEFAULT_LEVEL)) {
-        channel->maybe_default_compression_level.is_set = true;
+        channel->compression_options.default_level.is_set = true;
         GPR_ASSERT(args->args[i].value.integer >= 0 &&
                    args->args[i].value.integer < GRPC_COMPRESS_LEVEL_COUNT);
-        channel->maybe_default_compression_level.default_compression_level =
+        channel->compression_options.default_level.level =
             (grpc_compression_level)args->args[i].value.integer;
+      } else if (0 == strcmp(args->args[i].key,
+                             GRPC_COMPRESSION_CHANNEL_DEFAULT_ALGORITHM)) {
+        channel->compression_options.default_algorithm.is_set = true;
+        GPR_ASSERT(args->args[i].value.integer >= 0 &&
+                   args->args[i].value.integer <
+                       GRPC_COMPRESS_ALGORITHMS_COUNT);
+        channel->compression_options.default_algorithm.algorithm =
+            (grpc_compression_algorithm)args->args[i].value.integer;
+      } else if (0 ==
+                 strcmp(args->args[i].key,
+                        GRPC_COMPRESSION_CHANNEL_ENABLED_ALGORITHMS_BITSET)) {
+        channel->compression_options.enabled_algorithms_bitset =
+            (uint32_t)args->args[i].value.integer |
+            0x1; /* always support no compression */
       }
     }
     grpc_channel_args_destroy(args);
@@ -319,6 +331,11 @@ grpc_channel_stack *grpc_channel_get_channel_stack(grpc_channel *channel) {
   return CHANNEL_STACK_FROM_CHANNEL(channel);
 }
 
+grpc_compression_options grpc_channel_compression_options(
+    const grpc_channel *channel) {
+  return channel->compression_options;
+}
+
 grpc_mdelem *grpc_channel_get_reffed_status_elem(grpc_channel *channel, int i) {
   char tmp[GPR_LTOA_MIN_BUFSIZE];
   switch (i) {
@@ -337,15 +354,3 @@ grpc_mdelem *grpc_channel_get_reffed_status_elem(grpc_channel *channel, int i) {
 uint32_t grpc_channel_get_max_message_length(grpc_channel *channel) {
   return channel->max_message_length;
 }
-
-bool grpc_channel_default_compression_level(grpc_channel *channel,
-                                            grpc_compression_level *level) {
-  if (channel->maybe_default_compression_level.is_set) {
-    *level = channel->maybe_default_compression_level.default_compression_level;
-    return true;
-  }
-  return false;
-}
-
-bool grpc_channel_default_compression_algorithm(
-    grpc_channel *channel, grpc_compression_algorithm *algorithm);

src/core/lib/surface/channel.h

@@ -71,9 +71,8 @@ void grpc_channel_internal_unref(grpc_exec_ctx *exec_ctx,
   grpc_channel_internal_unref(exec_ctx, channel)
 #endif
 
-/** If the channel has an associated default compression level, return it in \a
- * level and return true. Otherwise return false. */
-bool grpc_channel_default_compression_level(grpc_channel *channel,
-                                            grpc_compression_level *level);
+/** Return the channel's compression options. */
+grpc_compression_options grpc_channel_compression_options(
+    const grpc_channel *channel);
 
 #endif /* GRPC_CORE_LIB_SURFACE_CHANNEL_H */

src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxi

@@ -442,6 +442,10 @@ cdef extern from "grpc/_cython/loader.h":
     GRPC_COMPRESS_LEVEL_HIGH
     GRPC_COMPRESS_LEVEL_COUNT
 
+  ctypedef struct grpc_compression_options:
+    uint32_t enabled_algorithms_bitset
+    grpc_compression_algorithm default_compression_algorithm
+
   int grpc_compression_algorithm_parse(
       const char *name, size_t name_length,
       grpc_compression_algorithm *algorithm) nogil
@@ -449,3 +453,13 @@ cdef extern from "grpc/_cython/loader.h":
                                       char **name) nogil
   grpc_compression_algorithm grpc_compression_algorithm_for_level(
       grpc_compression_level level, uint32_t accepted_encodings) nogil
+  void grpc_compression_options_init(grpc_compression_options *opts) nogil
+  void grpc_compression_options_enable_algorithm(
+      grpc_compression_options *opts,
+      grpc_compression_algorithm algorithm) nogil
+  void grpc_compression_options_disable_algorithm(
+      grpc_compression_options *opts,
+      grpc_compression_algorithm algorithm) nogil
+  int grpc_compression_options_is_algorithm_enabled(
+      const grpc_compression_options *opts,
+      grpc_compression_algorithm algorithm) nogil

src/python/grpcio/grpc/_cython/_cygrpc/records.pxd.pxi

@@ -123,3 +123,8 @@ cdef class Operations:
   cdef grpc_op *c_ops
   cdef size_t c_nops
   cdef list operations
+
+
+cdef class CompressionOptions:
+
+  cdef grpc_compression_options c_options

src/python/grpcio/grpc/_cython/_cygrpc/records.pyx.pxi

@@ -718,6 +718,32 @@ cdef class Operations:
     return _OperationsIterator(self)
 
 
+cdef class CompressionOptions:
+
+  def __cinit__(self):
+    with nogil:
+      grpc_compression_options_init(&self.c_options)
+
+  def enable_algorithm(self, grpc_compression_algorithm algorithm):
+    with nogil:
+      grpc_compression_options_enable_algorithm(&self.c_options, algorithm)
+
+  def disable_algorithm(self, grpc_compression_algorithm algorithm):
+    with nogil:
+      grpc_compression_options_disable_algorithm(&self.c_options, algorithm)
+
+  def is_algorithm_enabled(self, grpc_compression_algorithm algorithm):
+    cdef int result
+    with nogil:
+      result = grpc_compression_options_is_algorithm_enabled(
+          &self.c_options, algorithm)
+    return result
+
+  def to_channel_arg(self):
+    return ChannelArg(GRPC_COMPRESSION_CHANNEL_ENABLED_ALGORITHMS_BITSET,
+                      self.c_options.enabled_algorithms_bitset)
+
+
 def compression_algorithm_name(grpc_compression_algorithm algorithm):
   cdef char* name
   with nogil:

src/python/grpcio/grpc/_cython/imports.generated.c

@@ -72,6 +72,11 @@ census_view_get_data_type census_view_get_data_import;
 census_view_reset_type census_view_reset_import;
 grpc_compression_algorithm_parse_type grpc_compression_algorithm_parse_import;
 grpc_compression_algorithm_name_type grpc_compression_algorithm_name_import;
+grpc_compression_algorithm_for_level_type grpc_compression_algorithm_for_level_import;
+grpc_compression_options_init_type grpc_compression_options_init_import;
+grpc_compression_options_enable_algorithm_type grpc_compression_options_enable_algorithm_import;
+grpc_compression_options_disable_algorithm_type grpc_compression_options_disable_algorithm_import;
+grpc_compression_options_is_algorithm_enabled_type grpc_compression_options_is_algorithm_enabled_import;
 grpc_metadata_array_init_type grpc_metadata_array_init_import;
 grpc_metadata_array_destroy_type grpc_metadata_array_destroy_import;
 grpc_call_details_init_type grpc_call_details_init_import;
@@ -338,6 +343,11 @@ void pygrpc_load_imports(HMODULE library) {
   census_view_reset_import = (census_view_reset_type) GetProcAddress(library, "census_view_reset");
   grpc_compression_algorithm_parse_import = (grpc_compression_algorithm_parse_type) GetProcAddress(library, "grpc_compression_algorithm_parse");
   grpc_compression_algorithm_name_import = (grpc_compression_algorithm_name_type) GetProcAddress(library, "grpc_compression_algorithm_name");
+  grpc_compression_algorithm_for_level_import = (grpc_compression_algorithm_for_level_type) GetProcAddress(library, "grpc_compression_algorithm_for_level");
+  grpc_compression_options_init_import = (grpc_compression_options_init_type) GetProcAddress(library, "grpc_compression_options_init");
+  grpc_compression_options_enable_algorithm_import = (grpc_compression_options_enable_algorithm_type) GetProcAddress(library, "grpc_compression_options_enable_algorithm");
+  grpc_compression_options_disable_algorithm_import = (grpc_compression_options_disable_algorithm_type) GetProcAddress(library, "grpc_compression_options_disable_algorithm");
+  grpc_compression_options_is_algorithm_enabled_import = (grpc_compression_options_is_algorithm_enabled_type) GetProcAddress(library, "grpc_compression_options_is_algorithm_enabled");
   grpc_metadata_array_init_import = (grpc_metadata_array_init_type) GetProcAddress(library, "grpc_metadata_array_init");
   grpc_metadata_array_destroy_import = (grpc_metadata_array_destroy_type) GetProcAddress(library, "grpc_metadata_array_destroy");
   grpc_call_details_init_import = (grpc_call_details_init_type) GetProcAddress(library, "grpc_call_details_init");

src/python/grpcio/grpc/_cython/imports.generated.h

@@ -167,6 +167,21 @@ extern grpc_compression_algorithm_parse_type grpc_compression_algorithm_parse_im
 typedef int(*grpc_compression_algorithm_name_type)(grpc_compression_algorithm algorithm, char **name);
 extern grpc_compression_algorithm_name_type grpc_compression_algorithm_name_import;
 #define grpc_compression_algorithm_name grpc_compression_algorithm_name_import
+typedef grpc_compression_algorithm(*grpc_compression_algorithm_for_level_type)(grpc_compression_level level, uint32_t accepted_encodings);
+extern grpc_compression_algorithm_for_level_type grpc_compression_algorithm_for_level_import;
+#define grpc_compression_algorithm_for_level grpc_compression_algorithm_for_level_import
+typedef void(*grpc_compression_options_init_type)(grpc_compression_options *opts);
+extern grpc_compression_options_init_type grpc_compression_options_init_import;
+#define grpc_compression_options_init grpc_compression_options_init_import
+typedef void(*grpc_compression_options_enable_algorithm_type)(grpc_compression_options *opts, grpc_compression_algorithm algorithm);
+extern grpc_compression_options_enable_algorithm_type grpc_compression_options_enable_algorithm_import;
+#define grpc_compression_options_enable_algorithm grpc_compression_options_enable_algorithm_import
+typedef void(*grpc_compression_options_disable_algorithm_type)(grpc_compression_options *opts, grpc_compression_algorithm algorithm);
+extern grpc_compression_options_disable_algorithm_type grpc_compression_options_disable_algorithm_import;
+#define grpc_compression_options_disable_algorithm grpc_compression_options_disable_algorithm_import
+typedef int(*grpc_compression_options_is_algorithm_enabled_type)(const grpc_compression_options *opts, grpc_compression_algorithm algorithm);
+extern grpc_compression_options_is_algorithm_enabled_type grpc_compression_options_is_algorithm_enabled_import;
+#define grpc_compression_options_is_algorithm_enabled grpc_compression_options_is_algorithm_enabled_import
 typedef void(*grpc_metadata_array_init_type)(grpc_metadata_array *array);
 extern grpc_metadata_array_init_type grpc_metadata_array_init_import;
 #define grpc_metadata_array_init grpc_metadata_array_init_import

src/python/grpcio/grpc_core_dependencies.py

@@ -84,7 +84,7 @@ CORE_SOURCE_FILES = [
   'src/core/lib/channel/connected_channel.c',
   'src/core/lib/channel/http_client_filter.c',
   'src/core/lib/channel/http_server_filter.c',
-  'src/core/lib/compression/compression_algorithm.c',
+  'src/core/lib/compression/compression.c',
   'src/core/lib/compression/message_compress.c',
   'src/core/lib/debug/trace.c',
   'src/core/lib/http/format_request.c',

src/ruby/ext/grpc/rb_grpc_imports.generated.c

@@ -72,6 +72,11 @@ census_view_get_data_type census_view_get_data_import;
 census_view_reset_type census_view_reset_import;
 grpc_compression_algorithm_parse_type grpc_compression_algorithm_parse_import;
 grpc_compression_algorithm_name_type grpc_compression_algorithm_name_import;
+grpc_compression_algorithm_for_level_type grpc_compression_algorithm_for_level_import;
+grpc_compression_options_init_type grpc_compression_options_init_import;
+grpc_compression_options_enable_algorithm_type grpc_compression_options_enable_algorithm_import;
+grpc_compression_options_disable_algorithm_type grpc_compression_options_disable_algorithm_import;
+grpc_compression_options_is_algorithm_enabled_type grpc_compression_options_is_algorithm_enabled_import;
 grpc_metadata_array_init_type grpc_metadata_array_init_import;
 grpc_metadata_array_destroy_type grpc_metadata_array_destroy_import;
 grpc_call_details_init_type grpc_call_details_init_import;
@@ -334,6 +339,11 @@ void grpc_rb_load_imports(HMODULE library) {
   census_view_reset_import = (census_view_reset_type) GetProcAddress(library, "census_view_reset");
   grpc_compression_algorithm_parse_import = (grpc_compression_algorithm_parse_type) GetProcAddress(library, "grpc_compression_algorithm_parse");
   grpc_compression_algorithm_name_import = (grpc_compression_algorithm_name_type) GetProcAddress(library, "grpc_compression_algorithm_name");
+  grpc_compression_algorithm_for_level_import = (grpc_compression_algorithm_for_level_type) GetProcAddress(library, "grpc_compression_algorithm_for_level");
+  grpc_compression_options_init_import = (grpc_compression_options_init_type) GetProcAddress(library, "grpc_compression_options_init");
+  grpc_compression_options_enable_algorithm_import = (grpc_compression_options_enable_algorithm_type) GetProcAddress(library, "grpc_compression_options_enable_algorithm");
+  grpc_compression_options_disable_algorithm_import = (grpc_compression_options_disable_algorithm_type) GetProcAddress(library, "grpc_compression_options_disable_algorithm");
+  grpc_compression_options_is_algorithm_enabled_import = (grpc_compression_options_is_algorithm_enabled_type) GetProcAddress(library, "grpc_compression_options_is_algorithm_enabled");
   grpc_metadata_array_init_import = (grpc_metadata_array_init_type) GetProcAddress(library, "grpc_metadata_array_init");
   grpc_metadata_array_destroy_import = (grpc_metadata_array_destroy_type) GetProcAddress(library, "grpc_metadata_array_destroy");
   grpc_call_details_init_import = (grpc_call_details_init_type) GetProcAddress(library, "grpc_call_details_init");

src/ruby/ext/grpc/rb_grpc_imports.generated.h

@@ -167,6 +167,21 @@ extern grpc_compression_algorithm_parse_type grpc_compression_algorithm_parse_im
 typedef int(*grpc_compression_algorithm_name_type)(grpc_compression_algorithm algorithm, char **name);
 extern grpc_compression_algorithm_name_type grpc_compression_algorithm_name_import;
 #define grpc_compression_algorithm_name grpc_compression_algorithm_name_import
+typedef grpc_compression_algorithm(*grpc_compression_algorithm_for_level_type)(grpc_compression_level level, uint32_t accepted_encodings);
+extern grpc_compression_algorithm_for_level_type grpc_compression_algorithm_for_level_import;
+#define grpc_compression_algorithm_for_level grpc_compression_algorithm_for_level_import
+typedef void(*grpc_compression_options_init_type)(grpc_compression_options *opts);
+extern grpc_compression_options_init_type grpc_compression_options_init_import;
+#define grpc_compression_options_init grpc_compression_options_init_import
+typedef void(*grpc_compression_options_enable_algorithm_type)(grpc_compression_options *opts, grpc_compression_algorithm algorithm);
+extern grpc_compression_options_enable_algorithm_type grpc_compression_options_enable_algorithm_import;
+#define grpc_compression_options_enable_algorithm grpc_compression_options_enable_algorithm_import
+typedef void(*grpc_compression_options_disable_algorithm_type)(grpc_compression_options *opts, grpc_compression_algorithm algorithm);
+extern grpc_compression_options_disable_algorithm_type grpc_compression_options_disable_algorithm_import;
+#define grpc_compression_options_disable_algorithm grpc_compression_options_disable_algorithm_import
+typedef int(*grpc_compression_options_is_algorithm_enabled_type)(const grpc_compression_options *opts, grpc_compression_algorithm algorithm);
+extern grpc_compression_options_is_algorithm_enabled_type grpc_compression_options_is_algorithm_enabled_import;
+#define grpc_compression_options_is_algorithm_enabled grpc_compression_options_is_algorithm_enabled_import
 typedef void(*grpc_metadata_array_init_type)(grpc_metadata_array *array);
 extern grpc_metadata_array_init_type grpc_metadata_array_init_import;
 #define grpc_metadata_array_init grpc_metadata_array_init_import

test/core/channel/channel_args_test.c

@@ -136,8 +136,10 @@ static void test_compression_algorithm_states(void) {
 
 int main(int argc, char **argv) {
   grpc_test_init(argc, argv);
+  grpc_init();
   test_create();
   test_set_compression_algorithm();
   test_compression_algorithm_states();
+  grpc_shutdown();
   return 0;
 }

test/core/compression/compression_test.c

@@ -92,10 +92,137 @@ static void test_compression_algorithm_name(void) {
   /* the value of "name" is undefined upon failure */
 }
 
+static void test_compression_algorithm_for_level(void) {
+  gpr_log(GPR_DEBUG, "test_compression_algorithm_for_level");
+
+  {
+    /* accept only identity (aka none) */
+    uint32_t accepted_encodings = 0;
+    GPR_BITSET(&accepted_encodings, GRPC_COMPRESS_NONE); /* always */
+
+    GPR_ASSERT(GRPC_COMPRESS_NONE ==
+               grpc_compression_algorithm_for_level(GRPC_COMPRESS_LEVEL_NONE,
+                                                    accepted_encodings));
+
+    GPR_ASSERT(GRPC_COMPRESS_NONE ==
+               grpc_compression_algorithm_for_level(GRPC_COMPRESS_LEVEL_LOW,
+                                                    accepted_encodings));
+
+    GPR_ASSERT(GRPC_COMPRESS_NONE ==
+               grpc_compression_algorithm_for_level(GRPC_COMPRESS_LEVEL_MED,
+                                                    accepted_encodings));
+
+    GPR_ASSERT(GRPC_COMPRESS_NONE ==
+               grpc_compression_algorithm_for_level(GRPC_COMPRESS_LEVEL_HIGH,
+                                                    accepted_encodings));
+  }
+
+  {
+    /* accept only gzip */
+    uint32_t accepted_encodings = 0;
+    GPR_BITSET(&accepted_encodings, GRPC_COMPRESS_NONE); /* always */
+    GPR_BITSET(&accepted_encodings, GRPC_COMPRESS_GZIP);
+
+    GPR_ASSERT(GRPC_COMPRESS_NONE ==
+               grpc_compression_algorithm_for_level(GRPC_COMPRESS_LEVEL_NONE,
+                                                    accepted_encodings));
+
+    GPR_ASSERT(GRPC_COMPRESS_GZIP ==
+               grpc_compression_algorithm_for_level(GRPC_COMPRESS_LEVEL_LOW,
+                                                    accepted_encodings));
+
+    GPR_ASSERT(GRPC_COMPRESS_GZIP ==
+               grpc_compression_algorithm_for_level(GRPC_COMPRESS_LEVEL_MED,
+                                                    accepted_encodings));
+
+    GPR_ASSERT(GRPC_COMPRESS_GZIP ==
+               grpc_compression_algorithm_for_level(GRPC_COMPRESS_LEVEL_HIGH,
+                                                    accepted_encodings));
+  }
+
+  {
+    /* accept only deflate */
+    uint32_t accepted_encodings = 0;
+    GPR_BITSET(&accepted_encodings, GRPC_COMPRESS_NONE); /* always */
+    GPR_BITSET(&accepted_encodings, GRPC_COMPRESS_DEFLATE);
+
+    GPR_ASSERT(GRPC_COMPRESS_NONE ==
+               grpc_compression_algorithm_for_level(GRPC_COMPRESS_LEVEL_NONE,
+                                                    accepted_encodings));
+
+    GPR_ASSERT(GRPC_COMPRESS_DEFLATE ==
+               grpc_compression_algorithm_for_level(GRPC_COMPRESS_LEVEL_LOW,
+                                                    accepted_encodings));
+
+    GPR_ASSERT(GRPC_COMPRESS_DEFLATE ==
+               grpc_compression_algorithm_for_level(GRPC_COMPRESS_LEVEL_MED,
+                                                    accepted_encodings));
+
+    GPR_ASSERT(GRPC_COMPRESS_DEFLATE ==
+               grpc_compression_algorithm_for_level(GRPC_COMPRESS_LEVEL_HIGH,
+                                                    accepted_encodings));
+  }
+
+  {
+    /* accept gzip and deflate */
+    uint32_t accepted_encodings = 0;
+    GPR_BITSET(&accepted_encodings, GRPC_COMPRESS_NONE); /* always */
+    GPR_BITSET(&accepted_encodings, GRPC_COMPRESS_GZIP);
+    GPR_BITSET(&accepted_encodings, GRPC_COMPRESS_DEFLATE);
+
+    GPR_ASSERT(GRPC_COMPRESS_NONE ==
+               grpc_compression_algorithm_for_level(GRPC_COMPRESS_LEVEL_NONE,
+                                                    accepted_encodings));
+
+    GPR_ASSERT(GRPC_COMPRESS_GZIP ==
+               grpc_compression_algorithm_for_level(GRPC_COMPRESS_LEVEL_LOW,
+                                                    accepted_encodings));
+
+    GPR_ASSERT(GRPC_COMPRESS_DEFLATE ==
+               grpc_compression_algorithm_for_level(GRPC_COMPRESS_LEVEL_MED,
+                                                    accepted_encodings));
+
+    GPR_ASSERT(GRPC_COMPRESS_DEFLATE ==
+               grpc_compression_algorithm_for_level(GRPC_COMPRESS_LEVEL_HIGH,
+                                                    accepted_encodings));
+  }
+}
+
+static void test_compression_enable_disable_algorithm(void) {
+  grpc_compression_options options;
+  grpc_compression_algorithm algorithm;
+
+  gpr_log(GPR_DEBUG, "test_compression_enable_disable_algorithm");
+
+  grpc_compression_options_init(&options);
+  for (algorithm = GRPC_COMPRESS_NONE;
+       algorithm < GRPC_COMPRESS_ALGORITHMS_COUNT; algorithm++) {
+    /* all algorithms are enabled by default */
+    GPR_ASSERT(grpc_compression_options_is_algorithm_enabled(&options,
+                                                             algorithm) != 0);
+  }
+  /* disable one by one */
+  for (algorithm = GRPC_COMPRESS_NONE;
+       algorithm < GRPC_COMPRESS_ALGORITHMS_COUNT; algorithm++) {
+    grpc_compression_options_disable_algorithm(&options, algorithm);
+    GPR_ASSERT(grpc_compression_options_is_algorithm_enabled(&options,
+                                                             algorithm) == 0);
+  }
+  /* re-enable one by one */
+  for (algorithm = GRPC_COMPRESS_NONE;
+       algorithm < GRPC_COMPRESS_ALGORITHMS_COUNT; algorithm++) {
+    grpc_compression_options_enable_algorithm(&options, algorithm);
+    GPR_ASSERT(grpc_compression_options_is_algorithm_enabled(&options,
+                                                             algorithm) != 0);
+  }
+}
+
 int main(int argc, char **argv) {
   grpc_init();
   test_compression_algorithm_parse();
   test_compression_algorithm_name();
+  test_compression_algorithm_for_level();
+  test_compression_enable_disable_algorithm();
   grpc_shutdown();
 
   return 0;

test/core/end2end/cq_verifier.c

@@ -284,6 +284,6 @@ static expectation *add(cq_verifier *v, grpc_completion_type type, void *tag) {
   return e;
 }
 
-void cq_expect_completion(cq_verifier *v, void *tag, int success) {
+void cq_expect_completion(cq_verifier *v, void *tag, bool success) {
   add(v, GRPC_OP_COMPLETE, tag)->success = success;
 }

test/core/end2end/cq_verifier.h

@@ -34,6 +34,8 @@
 #ifndef GRPC_TEST_CORE_END2END_CQ_VERIFIER_H
 #define GRPC_TEST_CORE_END2END_CQ_VERIFIER_H
 
+#include <stdbool.h>
+
 #include <grpc/grpc.h>
 #include "test/core/util/test_config.h"
 
@@ -57,7 +59,7 @@ void cq_verify_empty(cq_verifier *v);
    Any functions taking ... expect a NULL terminated list of key/value pairs
    (each pair using two parameter slots) of metadata that MUST be present in
    the event. */
-void cq_expect_completion(cq_verifier *v, void *tag, int success);
+void cq_expect_completion(cq_verifier *v, void *tag, bool success);
 
 int byte_buffer_eq_string(grpc_byte_buffer *byte_buffer, const char *string);
 int contains_metadata(grpc_metadata_array *array, const char *key,

test/core/end2end/tests/compressed_payload.c

@@ -38,9 +38,10 @@
 
 #include <grpc/byte_buffer.h>
 #include <grpc/byte_buffer_reader.h>
-#include <grpc/impl/codegen/compression_types.h>
+#include <grpc/compression.h>
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
 #include <grpc/support/time.h>
 #include <grpc/support/useful.h>
 
@@ -103,6 +104,170 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_destroy(f->cq);
 }
 
+static void request_for_disabled_algorithm(
+    grpc_end2end_test_config config, const char *test_name,
+    uint32_t send_flags_bitmask,
+    grpc_compression_algorithm algorithm_to_disable,
+    grpc_compression_algorithm requested_client_compression_algorithm,
+    grpc_status_code expected_error, grpc_metadata *client_metadata) {
+  grpc_call *c;
+  grpc_call *s;
+  gpr_slice request_payload_slice;
+  grpc_byte_buffer *request_payload;
+  gpr_timespec deadline = five_seconds_time();
+  grpc_channel_args *client_args;
+  grpc_channel_args *server_args;
+  grpc_end2end_test_fixture f;
+  grpc_op ops[6];
+  grpc_op *op;
+  grpc_metadata_array initial_metadata_recv;
+  grpc_metadata_array trailing_metadata_recv;
+  grpc_metadata_array request_metadata_recv;
+  grpc_byte_buffer *request_payload_recv = NULL;
+  grpc_call_details call_details;
+  grpc_status_code status;
+  grpc_call_error error;
+  char *details = NULL;
+  size_t details_capacity = 0;
+  int was_cancelled = 2;
+  cq_verifier *cqv;
+  char str[1024];
+
+  memset(str, 'x', 1023);
+  str[1023] = '\0';
+  request_payload_slice = gpr_slice_from_copied_string(str);
+  request_payload = grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+
+  client_args = grpc_channel_args_set_compression_algorithm(
+      NULL, requested_client_compression_algorithm);
+  server_args =
+      grpc_channel_args_set_compression_algorithm(NULL, GRPC_COMPRESS_NONE);
+  server_args = grpc_channel_args_compression_algorithm_set_state(
+      &server_args, algorithm_to_disable, false);
+
+  f = begin_test(config, test_name, client_args, server_args);
+  cqv = cq_verifier_create(f.cq);
+
+  c = grpc_channel_create_call(f.client, NULL, GRPC_PROPAGATE_DEFAULTS, f.cq,
+                               "/foo", "foo.test.google.fr", deadline, NULL);
+  GPR_ASSERT(c);
+
+  grpc_metadata_array_init(&initial_metadata_recv);
+  grpc_metadata_array_init(&trailing_metadata_recv);
+  grpc_metadata_array_init(&request_metadata_recv);
+  grpc_call_details_init(&call_details);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  if (client_metadata != NULL) {
+    op->data.send_initial_metadata.count = 1;
+    op->data.send_initial_metadata.metadata = client_metadata;
+  } else {
+    op->data.send_initial_metadata.count = 0;
+  }
+  op->flags = 0;
+  op->reserved = NULL;
+  op++;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message = request_payload;
+  op->flags = send_flags_bitmask;
+  op->reserved = NULL;
+  op++;
+  op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+  op->flags = 0;
+  op->reserved = NULL;
+  op++;
+  op->op = GRPC_OP_RECV_INITIAL_METADATA;
+  op->data.recv_initial_metadata = &initial_metadata_recv;
+  op->flags = 0;
+  op->reserved = NULL;
+  op++;
+  op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+  op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+  op->data.recv_status_on_client.status = &status;
+  op->data.recv_status_on_client.status_details = &details;
+  op->data.recv_status_on_client.status_details_capacity = &details_capacity;
+  op->flags = 0;
+  op->reserved = NULL;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), NULL);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  error =
+      grpc_server_request_call(f.server, &s, &call_details,
+                               &request_metadata_recv, f.cq, f.cq, tag(101));
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  cq_expect_completion(cqv, tag(101), true);
+  cq_verify(cqv);
+
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op->flags = 0;
+  op->reserved = NULL;
+  op++;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message = &request_payload_recv;
+  op->flags = 0;
+  op->reserved = NULL;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), NULL);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  cq_expect_completion(cqv, tag(102), false);
+
+  op = ops;
+  op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  op->data.recv_close_on_server.cancelled = &was_cancelled;
+  op->flags = 0;
+  op->reserved = NULL;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(103), NULL);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  cq_expect_completion(cqv, tag(103), true);
+  cq_expect_completion(cqv, tag(1), true);
+  cq_verify(cqv);
+
+  /* call was cancelled (closed) ... */
+  GPR_ASSERT(was_cancelled != 0);
+  /* with a certain error */
+  GPR_ASSERT(status == expected_error);
+
+  char *algo_name = NULL;
+  GPR_ASSERT(grpc_compression_algorithm_name(algorithm_to_disable, &algo_name));
+  char *expected_details = NULL;
+  gpr_asprintf(&expected_details, "Compression algorithm '%s' is disabled.",
+               algo_name);
+  /* and we expect a specific reason for it */
+  GPR_ASSERT(0 == strcmp(details, expected_details));
+  gpr_free(expected_details);
+  GPR_ASSERT(0 == strcmp(call_details.method, "/foo"));
+  GPR_ASSERT(0 == strcmp(call_details.host, "foo.test.google.fr"));
+
+  gpr_free(details);
+  grpc_metadata_array_destroy(&initial_metadata_recv);
+  grpc_metadata_array_destroy(&trailing_metadata_recv);
+  grpc_metadata_array_destroy(&request_metadata_recv);
+  grpc_call_details_destroy(&call_details);
+
+  grpc_call_destroy(c);
+  grpc_call_destroy(s);
+
+  cq_verifier_destroy(cqv);
+
+  gpr_slice_unref(request_payload_slice);
+  grpc_byte_buffer_destroy(request_payload);
+  grpc_byte_buffer_destroy(request_payload_recv);
+
+  grpc_channel_args_destroy(client_args);
+  grpc_channel_args_destroy(server_args);
+
+  end_test(&f);
+  config.tear_down_data(&f);
+}
+
 static void request_with_payload_template(
     grpc_end2end_test_config config, const char *test_name,
     uint32_t client_send_flags_bitmask,
@@ -196,11 +361,11 @@ static void request_with_payload_template(
       grpc_server_request_call(f.server, &s, &call_details,
                                &request_metadata_recv, f.cq, f.cq, tag(100));
   GPR_ASSERT(GRPC_CALL_OK == error);
-  cq_expect_completion(cqv, tag(100), 1);
+  cq_expect_completion(cqv, tag(100), true);
   cq_verify(cqv);
 
-  GPR_ASSERT(
-      GPR_BITCOUNT(grpc_call_test_only_get_encodings_accepted_by_peer(s)) == 3);
+  GPR_ASSERT(GPR_BITCOUNT(grpc_call_test_only_get_encodings_accepted_by_peer(
+                 s)) == GRPC_COMPRESS_ALGORITHMS_COUNT);
   GPR_ASSERT(GPR_BITGET(grpc_call_test_only_get_encodings_accepted_by_peer(s),
                         GRPC_COMPRESS_NONE) != 0);
   GPR_ASSERT(GPR_BITGET(grpc_call_test_only_get_encodings_accepted_by_peer(s),
@@ -423,12 +588,20 @@ static void test_invoke_request_with_compressed_payload_md_override(
       /*ignored*/ GRPC_COMPRESS_LEVEL_NONE);
 }
 
+static void test_invoke_request_with_disabled_algorithm(
+    grpc_end2end_test_config config) {
+  request_for_disabled_algorithm(
+      config, "test_invoke_request_with_disabled_algorithm", 0,
+      GRPC_COMPRESS_GZIP, GRPC_COMPRESS_GZIP, GRPC_STATUS_UNIMPLEMENTED, NULL);
+}
+
 void compressed_payload(grpc_end2end_test_config config) {
   test_invoke_request_with_exceptionally_uncompressed_payload(config);
   test_invoke_request_with_uncompressed_payload(config);
   test_invoke_request_with_compressed_payload(config);
   test_invoke_request_with_server_level(config);
   test_invoke_request_with_compressed_payload_md_override(config);
+  test_invoke_request_with_disabled_algorithm(config);
 }
 
 void compressed_payload_pre_init(void) {}

tools/doxygen/Doxyfile.core.internal

@@ -936,7 +936,7 @@ src/core/lib/channel/compress_filter.c \
 src/core/lib/channel/connected_channel.c \
 src/core/lib/channel/http_client_filter.c \
 src/core/lib/channel/http_server_filter.c \
-src/core/lib/compression/compression_algorithm.c \
+src/core/lib/compression/compression.c \
 src/core/lib/compression/message_compress.c \
 src/core/lib/debug/trace.c \
 src/core/lib/http/format_request.c \

tools/run_tests/sources_and_headers.json

@@ -5727,7 +5727,7 @@
       "src/core/lib/channel/http_server_filter.c", 
       "src/core/lib/channel/http_server_filter.h", 
       "src/core/lib/compression/algorithm_metadata.h", 
-      "src/core/lib/compression/compression_algorithm.c", 
+      "src/core/lib/compression/compression.c", 
       "src/core/lib/compression/message_compress.c", 
       "src/core/lib/compression/message_compress.h", 
       "src/core/lib/debug/trace.c", 

vsprojects/vcxproj/grpc/grpc.vcxproj

@@ -455,7 +455,7 @@
     </ClCompile>
     <ClCompile Include="$(SolutionDir)\..\src\core\lib\channel\http_server_filter.c">
     </ClCompile>
-    <ClCompile Include="$(SolutionDir)\..\src\core\lib\compression\compression_algorithm.c">
+    <ClCompile Include="$(SolutionDir)\..\src\core\lib\compression\compression.c">
     </ClCompile>
     <ClCompile Include="$(SolutionDir)\..\src\core\lib\compression\message_compress.c">
     </ClCompile>

vsprojects/vcxproj/grpc/grpc.vcxproj.filters

@@ -25,7 +25,7 @@
     <ClCompile Include="$(SolutionDir)\..\src\core\lib\channel\http_server_filter.c">
       <Filter>src\core\lib\channel</Filter>
     </ClCompile>
-    <ClCompile Include="$(SolutionDir)\..\src\core\lib\compression\compression_algorithm.c">
+    <ClCompile Include="$(SolutionDir)\..\src\core\lib\compression\compression.c">
       <Filter>src\core\lib\compression</Filter>
     </ClCompile>
     <ClCompile Include="$(SolutionDir)\..\src\core\lib\compression\message_compress.c">

vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj

@@ -430,7 +430,7 @@
     </ClCompile>
     <ClCompile Include="$(SolutionDir)\..\src\core\lib\channel\http_server_filter.c">
     </ClCompile>
-    <ClCompile Include="$(SolutionDir)\..\src\core\lib\compression\compression_algorithm.c">
+    <ClCompile Include="$(SolutionDir)\..\src\core\lib\compression\compression.c">
     </ClCompile>
     <ClCompile Include="$(SolutionDir)\..\src\core\lib\compression\message_compress.c">
     </ClCompile>

vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj.filters

@@ -28,7 +28,7 @@
     <ClCompile Include="$(SolutionDir)\..\src\core\lib\channel\http_server_filter.c">
       <Filter>src\core\lib\channel</Filter>
     </ClCompile>
-    <ClCompile Include="$(SolutionDir)\..\src\core\lib\compression\compression_algorithm.c">
+    <ClCompile Include="$(SolutionDir)\..\src\core\lib\compression\compression.c">
       <Filter>src\core\lib\compression</Filter>
     </ClCompile>
     <ClCompile Include="$(SolutionDir)\..\src\core\lib\compression\message_compress.c">