Merge remote-tracking branch 'upstream/master' into add_udp_server_2

26 files changed, 652 insertions, 238 deletions

diff --git a/src/core/channel/client_channel.c b/src/core/channel/client_channel.c
index 2ee260b799..6c2e6b38a8 100644
--- a/src/core/channel/client_channel.c
+++ b/src/core/channel/client_channel.c
@@ -401,6 +401,7 @@ static void perform_transport_stream_op(grpc_call_element *elem,
             calld->state = CALL_WAITING_FOR_CONFIG;
             add_to_lb_policy_wait_queue_locked_state_config(elem);
             if (!chand->started_resolving && chand->resolver != NULL) {
+              GRPC_CHANNEL_INTERNAL_REF(chand->master, "resolver");
               chand->started_resolving = 1;
               grpc_resolver_next(chand->resolver,
                                  &chand->incoming_configuration,
@@ -526,6 +527,7 @@ static void cc_on_config_changed(void *arg, int iomgr_success) {
   }
 
   if (old_lb_policy != NULL) {
+    grpc_lb_policy_shutdown(old_lb_policy);
     GRPC_LB_POLICY_UNREF(old_lb_policy, "channel");
   }
 
@@ -701,11 +703,11 @@ void grpc_client_channel_set_resolver(grpc_channel_stack *channel_stack,
   gpr_mu_lock(&chand->mu_config);
   GPR_ASSERT(!chand->resolver);
   chand->resolver = resolver;
-  GRPC_CHANNEL_INTERNAL_REF(chand->master, "resolver");
   GRPC_RESOLVER_REF(resolver, "channel");
   if (chand->waiting_for_config_closures != NULL ||
       chand->exit_idle_when_lb_policy_arrives) {
     chand->started_resolving = 1;
+    GRPC_CHANNEL_INTERNAL_REF(chand->master, "resolver");
     grpc_resolver_next(resolver, &chand->incoming_configuration,
                        &chand->on_config_changed);
   }
@@ -724,6 +726,7 @@ grpc_connectivity_state grpc_client_channel_check_connectivity_state(
     } else {
       chand->exit_idle_when_lb_policy_arrives = 1;
       if (!chand->started_resolving && chand->resolver != NULL) {
+        GRPC_CHANNEL_INTERNAL_REF(chand->master, "resolver");
         chand->started_resolving = 1;
         grpc_resolver_next(chand->resolver, &chand->incoming_configuration,
                            &chand->on_config_changed);
diff --git a/src/core/channel/compress_filter.c b/src/core/channel/compress_filter.c
index 9fc8589fbb..8963c13b0f 100644
--- a/src/core/channel/compress_filter.c
+++ b/src/core/channel/compress_filter.c
@@ -204,7 +204,7 @@ static void process_send_ops(grpc_call_element *elem,
           }
           grpc_metadata_batch_add_tail(
               &(sop->data.metadata), &calld->compression_algorithm_storage,
-              grpc_mdelem_ref(channeld->mdelem_compression_algorithms
+              GRPC_MDELEM_REF(channeld->mdelem_compression_algorithms
                                   [calld->compression_algorithm]));
           calld->written_initial_metadata = 1; /* GPR_TRUE */
         }
@@ -295,7 +295,7 @@ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master,
     channeld->mdelem_compression_algorithms[algo_idx] =
         grpc_mdelem_from_metadata_strings(
             mdctx,
-            grpc_mdstr_ref(channeld->mdstr_outgoing_compression_algorithm_key),
+            GRPC_MDSTR_REF(channeld->mdstr_outgoing_compression_algorithm_key),
             grpc_mdstr_from_string(mdctx, algorithm_name, 0));
   }
 
@@ -307,11 +307,11 @@ static void destroy_channel_elem(grpc_channel_element *elem) {
   channel_data *channeld = elem->channel_data;
   grpc_compression_algorithm algo_idx;
 
-  grpc_mdstr_unref(channeld->mdstr_request_compression_algorithm_key);
-  grpc_mdstr_unref(channeld->mdstr_outgoing_compression_algorithm_key);
+  GRPC_MDSTR_UNREF(channeld->mdstr_request_compression_algorithm_key);
+  GRPC_MDSTR_UNREF(channeld->mdstr_outgoing_compression_algorithm_key);
   for (algo_idx = 0; algo_idx < GRPC_COMPRESS_ALGORITHMS_COUNT;
        ++algo_idx) {
-    grpc_mdelem_unref(channeld->mdelem_compression_algorithms[algo_idx]);
+    GRPC_MDELEM_UNREF(channeld->mdelem_compression_algorithms[algo_idx]);
   }
 }
 
diff --git a/src/core/httpcli/httpcli.c b/src/core/httpcli/httpcli.c
index 65997d5f44..9012070e8e 100644
--- a/src/core/httpcli/httpcli.c
+++ b/src/core/httpcli/httpcli.c
@@ -40,9 +40,7 @@
 #include "src/core/iomgr/resolve_address.h"
 #include "src/core/iomgr/tcp_client.h"
 #include "src/core/httpcli/format_request.h"
-#include "src/core/httpcli/httpcli_security_connector.h"
 #include "src/core/httpcli/parser.h"
-#include "src/core/security/secure_transport_setup.h"
 #include "src/core/support/string.h"
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
@@ -57,7 +55,7 @@ typedef struct {
   char *host;
   gpr_timespec deadline;
   int have_read_byte;
-  int use_ssl;
+  const grpc_httpcli_handshaker *handshaker;
   grpc_httpcli_response_cb on_response;
   void *user_data;
   grpc_httpcli_context *context;
@@ -68,6 +66,16 @@ typedef struct {
 static grpc_httpcli_get_override g_get_override = NULL;
 static grpc_httpcli_post_override g_post_override = NULL;
 
+static void plaintext_handshake(void *arg, grpc_endpoint *endpoint,
+                                const char *host,
+                                void (*on_done)(void *arg,
+                                                grpc_endpoint *endpoint)) {
+  on_done(arg, endpoint);
+}
+
+const grpc_httpcli_handshaker grpc_httpcli_plaintext = {"http",
+                                                        plaintext_handshake};
+
 void grpc_httpcli_context_init(grpc_httpcli_context *context) {
   grpc_pollset_set_init(&context->pollset_set);
 }
@@ -163,18 +171,16 @@ static void start_write(internal_request *req) {
   }
 }
 
-static void on_secure_transport_setup_done(void *rp,
-                                           grpc_security_status status,
-                                           grpc_endpoint *wrapped_endpoint,
-                                           grpc_endpoint *secure_endpoint) {
-  internal_request *req = rp;
-  if (status != GRPC_SECURITY_OK) {
-    gpr_log(GPR_ERROR, "Secure transport setup failed with error %d.", status);
-    finish(req, 0);
-  } else {
-    req->ep = secure_endpoint;
-    start_write(req);
+static void on_handshake_done(void *arg, grpc_endpoint *ep) {
+  internal_request *req = arg;
+
+  if (!ep) {
+    next_address(req);
+    return;
   }
+
+  req->ep = ep;
+  start_write(req);
 }
 
 static void on_connected(void *arg, grpc_endpoint *tcp) {
@@ -184,25 +190,7 @@ static void on_connected(void *arg, grpc_endpoint *tcp) {
     next_address(req);
     return;
   }
-  req->ep = tcp;
-  if (req->use_ssl) {
-    grpc_channel_security_connector *sc = NULL;
-    const unsigned char *pem_root_certs = NULL;
-    size_t pem_root_certs_size = grpc_get_default_ssl_roots(&pem_root_certs);
-    if (pem_root_certs == NULL || pem_root_certs_size == 0) {
-      gpr_log(GPR_ERROR, "Could not get default pem root certs.");
-      finish(req, 0);
-      return;
-    }
-    GPR_ASSERT(grpc_httpcli_ssl_channel_security_connector_create(
-                   pem_root_certs, pem_root_certs_size, req->host, &sc) ==
-               GRPC_SECURITY_OK);
-    grpc_setup_secure_transport(&sc->base, tcp, on_secure_transport_setup_done,
-                                req);
-    GRPC_SECURITY_CONNECTOR_UNREF(&sc->base, "httpcli");
-  } else {
-    start_write(req);
-  }
+  req->handshaker->handshake(req, tcp, req->host, on_handshake_done);
 }
 
 static void next_address(internal_request *req) {
@@ -245,18 +233,17 @@ void grpc_httpcli_get(grpc_httpcli_context *context, grpc_pollset *pollset,
   req->on_response = on_response;
   req->user_data = user_data;
   req->deadline = deadline;
-  req->use_ssl = request->use_ssl;
+  req->handshaker =
+      request->handshaker ? request->handshaker : &grpc_httpcli_plaintext;
   req->context = context;
   req->pollset = pollset;
   gpr_asprintf(&name, "HTTP:GET:%s:%s", request->host, request->path);
   grpc_iomgr_register_object(&req->iomgr_obj, name);
   gpr_free(name);
-  if (req->use_ssl) {
-    req->host = gpr_strdup(request->host);
-  }
+  req->host = gpr_strdup(request->host);
 
   grpc_pollset_set_add_pollset(&req->context->pollset_set, req->pollset);
-  grpc_resolve_address(request->host, req->use_ssl ? "https" : "http",
+  grpc_resolve_address(request->host, req->handshaker->default_port,
                        on_resolved, req);
 }
 
@@ -279,18 +266,17 @@ void grpc_httpcli_post(grpc_httpcli_context *context, grpc_pollset *pollset,
   req->on_response = on_response;
   req->user_data = user_data;
   req->deadline = deadline;
-  req->use_ssl = request->use_ssl;
+  req->handshaker =
+      request->handshaker ? request->handshaker : &grpc_httpcli_plaintext;
   req->context = context;
   req->pollset = pollset;
   gpr_asprintf(&name, "HTTP:GET:%s:%s", request->host, request->path);
   grpc_iomgr_register_object(&req->iomgr_obj, name);
   gpr_free(name);
-  if (req->use_ssl) {
-    req->host = gpr_strdup(request->host);
-  }
+  req->host = gpr_strdup(request->host);
 
   grpc_pollset_set_add_pollset(&req->context->pollset_set, req->pollset);
-  grpc_resolve_address(request->host, req->use_ssl ? "https" : "http",
+  grpc_resolve_address(request->host, req->handshaker->default_port,
                        on_resolved, req);
 }
 
diff --git a/src/core/httpcli/httpcli.h b/src/core/httpcli/httpcli.h
index ab98178f8a..c45966714c 100644
--- a/src/core/httpcli/httpcli.h
+++ b/src/core/httpcli/httpcli.h
@@ -38,6 +38,7 @@
 
 #include <grpc/support/time.h>
 
+#include "src/core/iomgr/endpoint.h"
 #include "src/core/iomgr/pollset_set.h"
 
 /* User agent this library reports */
@@ -58,6 +59,15 @@ typedef struct grpc_httpcli_context {
   grpc_pollset_set pollset_set;
 } grpc_httpcli_context;
 
+typedef struct {
+  const char *default_port;
+  void (*handshake)(void *arg, grpc_endpoint *endpoint, const char *host,
+                    void (*on_done)(void *arg, grpc_endpoint *endpoint));
+} grpc_httpcli_handshaker;
+
+extern const grpc_httpcli_handshaker grpc_httpcli_plaintext;
+extern const grpc_httpcli_handshaker grpc_httpcli_ssl;
+
 /* A request */
 typedef struct grpc_httpcli_request {
   /* The host name to connect to */
@@ -69,8 +79,8 @@ typedef struct grpc_httpcli_request {
        Host, Connection, User-Agent */
   size_t hdr_count;
   grpc_httpcli_header *hdrs;
-  /* whether to use ssl for the request */
-  int use_ssl;
+  /* handshaker to use ssl for the request */
+  const grpc_httpcli_handshaker *handshaker;
 } grpc_httpcli_request;
 
 /* A response */
diff --git a/src/core/httpcli/httpcli_security_connector.c b/src/core/httpcli/httpcli_security_connector.c
index ce0d3d5a70..7887f9d530 100644
--- a/src/core/httpcli/httpcli_security_connector.c
+++ b/src/core/httpcli/httpcli_security_connector.c
@@ -31,7 +31,7 @@
  *
  */
 
-#include "src/core/httpcli/httpcli_security_connector.h"
+#include "src/core/httpcli/httpcli.h"
 
 #include <string.h>
 
@@ -96,7 +96,7 @@ static grpc_security_status httpcli_ssl_check_peer(grpc_security_connector *sc,
 static grpc_security_connector_vtable httpcli_ssl_vtable = {
     httpcli_ssl_destroy, httpcli_ssl_create_handshaker, httpcli_ssl_check_peer};
 
-grpc_security_status grpc_httpcli_ssl_channel_security_connector_create(
+static grpc_security_status httpcli_ssl_channel_security_connector_create(
     const unsigned char *pem_root_certs, size_t pem_root_certs_size,
     const char *secure_peer_name, grpc_channel_security_connector **sc) {
   tsi_result result = TSI_OK;
@@ -130,3 +130,48 @@ grpc_security_status grpc_httpcli_ssl_channel_security_connector_create(
   *sc = &c->base;
   return GRPC_SECURITY_OK;
 }
+
+/* handshaker */
+
+typedef struct {
+  void (*func)(void *arg, grpc_endpoint *endpoint);
+  void *arg;
+} on_done_closure;
+
+static void on_secure_transport_setup_done(void *rp,
+                                           grpc_security_status status,
+                                           grpc_endpoint *wrapped_endpoint,
+                                           grpc_endpoint *secure_endpoint) {
+  on_done_closure *c = rp;
+  if (status != GRPC_SECURITY_OK) {
+    gpr_log(GPR_ERROR, "Secure transport setup failed with error %d.", status);
+    c->func(c->arg, NULL);
+  } else {
+    c->func(c->arg, secure_endpoint);
+  }
+  gpr_free(c);
+}
+
+static void ssl_handshake(void *arg, grpc_endpoint *tcp, const char *host,
+                          void (*on_done)(void *arg, grpc_endpoint *endpoint)) {
+  grpc_channel_security_connector *sc = NULL;
+  const unsigned char *pem_root_certs = NULL;
+  on_done_closure *c = gpr_malloc(sizeof(*c));
+  size_t pem_root_certs_size = grpc_get_default_ssl_roots(&pem_root_certs);
+  if (pem_root_certs == NULL || pem_root_certs_size == 0) {
+    gpr_log(GPR_ERROR, "Could not get default pem root certs.");
+    on_done(arg, NULL);
+    gpr_free(c);
+    return;
+  }
+  c->func = on_done;
+  c->arg = arg;
+  GPR_ASSERT(httpcli_ssl_channel_security_connector_create(
+                 pem_root_certs, pem_root_certs_size, host, &sc) ==
+             GRPC_SECURITY_OK);
+  grpc_setup_secure_transport(&sc->base, tcp, on_secure_transport_setup_done,
+                              c);
+  GRPC_SECURITY_CONNECTOR_UNREF(&sc->base, "httpcli");
+}
+
+const grpc_httpcli_handshaker grpc_httpcli_ssl = {"https", ssl_handshake};
diff --git a/src/core/httpcli/httpcli_security_connector.h b/src/core/httpcli/httpcli_security_connector.h
deleted file mode 100644
index c50f25905e..0000000000
--- a/src/core/httpcli/httpcli_security_connector.h
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- *
- * Copyright 2015, Google Inc.
- * All rights reserved.
- *
- * Redistribution and use in source and binary forms, with or without
- * modification, are permitted provided that the following conditions are
- * met:
- *
- *     * Redistributions of source code must retain the above copyright
- * notice, this list of conditions and the following disclaimer.
- *     * Redistributions in binary form must reproduce the above
- * copyright notice, this list of conditions and the following disclaimer
- * in the documentation and/or other materials provided with the
- * distribution.
- *     * Neither the name of Google Inc. nor the names of its
- * contributors may be used to endorse or promote products derived from
- * this software without specific prior written permission.
- *
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
- * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
- * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
- * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
- * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
- * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
- * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
- * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
- * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
- * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
- * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
- *
- */
-
-#ifndef GRPC_INTERNAL_CORE_HTTPCLI_HTTPCLI_SECURITY_CONNECTOR_H
-#define GRPC_INTERNAL_CORE_HTTPCLI_HTTPCLI_SECURITY_CONNECTOR_H
-
-#include "src/core/security/security_connector.h"
-
-grpc_security_status grpc_httpcli_ssl_channel_security_connector_create(
-    const unsigned char *pem_root_certs, size_t pem_root_certs_size,
-    const char *secure_peer_name, grpc_channel_security_connector **sc);
-
-#endif  /* GRPC_INTERNAL_CORE_HTTPCLI_HTTPCLI_SECURITY_CONNECTOR_H */
diff --git a/src/core/security/client_auth_filter.c b/src/core/security/client_auth_filter.c
index e86b5430b2..410852da52 100644
--- a/src/core/security/client_auth_filter.c
+++ b/src/core/security/client_auth_filter.c
@@ -75,12 +75,11 @@ typedef struct {
   grpc_mdstr *status_key;
 } channel_data;
 
-static void bubble_up_error(grpc_call_element *elem, const char *error_msg) {
+static void bubble_up_error(grpc_call_element *elem, grpc_status_code status,
+                            const char *error_msg) {
   call_data *calld = elem->call_data;
-  channel_data *chand = elem->channel_data;
-  grpc_transport_stream_op_add_cancellation(
-      &calld->op, GRPC_STATUS_UNAUTHENTICATED,
-      grpc_mdstr_from_string(chand->md_ctx, error_msg, 0));
+  gpr_log(GPR_ERROR, "Client side authentication failure: %s", error_msg);
+  grpc_transport_stream_op_add_cancellation(&calld->op, status);
   grpc_call_next_op(elem, &calld->op);
 }
 
@@ -95,7 +94,8 @@ static void on_credentials_metadata(void *user_data,
   grpc_metadata_batch *mdb;
   size_t i;
   if (status != GRPC_CREDENTIALS_OK) {
-    bubble_up_error(elem, "Credentials failed to get metadata.");
+    bubble_up_error(elem, GRPC_STATUS_UNAUTHENTICATED,
+                    "Credentials failed to get metadata.");
     return;
   }
   GPR_ASSERT(num_md <= MAX_CREDENTIALS_METADATA_COUNT);
@@ -155,7 +155,7 @@ static void send_security_metadata(grpc_call_element *elem,
   if (channel_creds_has_md && call_creds_has_md) {
     calld->creds = grpc_composite_credentials_create(channel_creds, ctx->creds);
     if (calld->creds == NULL) {
-      bubble_up_error(elem,
+      bubble_up_error(elem, GRPC_STATUS_INVALID_ARGUMENT,
                       "Incompatible credentials set on channel and call.");
       return;
     }
@@ -183,7 +183,7 @@ static void on_host_checked(void *user_data, grpc_security_status status) {
     char *error_msg;
     gpr_asprintf(&error_msg, "Invalid host %s set in :authority metadata.",
                  grpc_mdstr_as_c_string(calld->host));
-    bubble_up_error(elem, error_msg);
+    bubble_up_error(elem, GRPC_STATUS_INVALID_ARGUMENT, error_msg);
     gpr_free(error_msg);
   }
 }
@@ -253,7 +253,7 @@ static void auth_start_transport_op(grpc_call_element *elem,
             gpr_asprintf(&error_msg,
                          "Invalid host %s set in :authority metadata.",
                          call_host);
-            bubble_up_error(elem, error_msg);
+            bubble_up_error(elem, GRPC_STATUS_INVALID_ARGUMENT, error_msg);
             gpr_free(error_msg);
           }
           return; /* early exit */
diff --git a/src/core/security/credentials.c b/src/core/security/credentials.c
index 15268cefbe..6421ce673d 100644
--- a/src/core/security/credentials.c
+++ b/src/core/security/credentials.c
@@ -149,6 +149,12 @@ grpc_security_status grpc_server_credentials_create_security_connector(
   return creds->vtable->create_security_connector(creds, sc);
 }
 
+void grpc_server_credentials_set_auth_metadata_processor(
+    grpc_server_credentials *creds, grpc_auth_metadata_processor processor) {
+  if (creds == NULL) return;
+  creds->processor = processor;
+}
+
 /* -- Ssl credentials. -- */
 
 static void ssl_destroy(grpc_credentials *creds) {
@@ -679,7 +685,7 @@ static void service_account_fetch_oauth2(
   request.path = GRPC_GOOGLE_OAUTH2_SERVICE_TOKEN_PATH;
   request.hdr_count = 1;
   request.hdrs = &header;
-  request.use_ssl = 1;
+  request.handshaker = &grpc_httpcli_ssl;
   grpc_httpcli_post(httpcli_context, pollset, &request, body, strlen(body),
                     deadline, response_cb, metadata_req);
   gpr_free(body);
@@ -738,7 +744,7 @@ static void refresh_token_fetch_oauth2(
   request.path = GRPC_GOOGLE_OAUTH2_SERVICE_TOKEN_PATH;
   request.hdr_count = 1;
   request.hdrs = &header;
-  request.use_ssl = 1;
+  request.handshaker = &grpc_httpcli_ssl;
   grpc_httpcli_post(httpcli_context, pollset, &request, body, strlen(body),
                     deadline, response_cb, metadata_req);
   gpr_free(body);
@@ -765,19 +771,19 @@ grpc_credentials *grpc_refresh_token_credentials_create(
       grpc_auth_refresh_token_create_from_string(json_refresh_token));
 }
 
-/* -- Fake Oauth2 credentials. -- */
+/* -- Metadata-only credentials. -- */
 
-static void fake_oauth2_destroy(grpc_credentials *creds) {
-  grpc_fake_oauth2_credentials *c = (grpc_fake_oauth2_credentials *)creds;
-  grpc_credentials_md_store_unref(c->access_token_md);
+static void md_only_test_destroy(grpc_credentials *creds) {
+  grpc_md_only_test_credentials *c = (grpc_md_only_test_credentials *)creds;
+  grpc_credentials_md_store_unref(c->md_store);
   gpr_free(c);
 }
 
-static int fake_oauth2_has_request_metadata(const grpc_credentials *creds) {
+static int md_only_test_has_request_metadata(const grpc_credentials *creds) {
   return 1;
 }
 
-static int fake_oauth2_has_request_metadata_only(
+static int md_only_test_has_request_metadata_only(
     const grpc_credentials *creds) {
   return 1;
 }
@@ -785,19 +791,19 @@ static int fake_oauth2_has_request_metadata_only(
 void on_simulated_token_fetch_done(void *user_data, int success) {
   grpc_credentials_metadata_request *r =
       (grpc_credentials_metadata_request *)user_data;
-  grpc_fake_oauth2_credentials *c = (grpc_fake_oauth2_credentials *)r->creds;
+  grpc_md_only_test_credentials *c = (grpc_md_only_test_credentials *)r->creds;
   GPR_ASSERT(success);
-  r->cb(r->user_data, c->access_token_md->entries,
-        c->access_token_md->num_entries, GRPC_CREDENTIALS_OK);
+  r->cb(r->user_data, c->md_store->entries,
+        c->md_store->num_entries, GRPC_CREDENTIALS_OK);
   grpc_credentials_metadata_request_destroy(r);
 }
 
-static void fake_oauth2_get_request_metadata(grpc_credentials *creds,
+static void md_only_test_get_request_metadata(grpc_credentials *creds,
                                              grpc_pollset *pollset,
                                              const char *service_url,
                                              grpc_credentials_metadata_cb cb,
                                              void *user_data) {
-  grpc_fake_oauth2_credentials *c = (grpc_fake_oauth2_credentials *)creds;
+  grpc_md_only_test_credentials *c = (grpc_md_only_test_credentials *)creds;
 
   if (c->is_async) {
     grpc_credentials_metadata_request *cb_arg =
@@ -806,26 +812,26 @@ static void fake_oauth2_get_request_metadata(grpc_credentials *creds,
                             on_simulated_token_fetch_done, cb_arg);
     grpc_iomgr_add_callback(cb_arg->on_simulated_token_fetch_done_closure);
   } else {
-    cb(user_data, c->access_token_md->entries, 1, GRPC_CREDENTIALS_OK);
+    cb(user_data, c->md_store->entries, 1, GRPC_CREDENTIALS_OK);
   }
 }
 
-static grpc_credentials_vtable fake_oauth2_vtable = {
-    fake_oauth2_destroy, fake_oauth2_has_request_metadata,
-    fake_oauth2_has_request_metadata_only, fake_oauth2_get_request_metadata,
+static grpc_credentials_vtable md_only_test_vtable = {
+    md_only_test_destroy, md_only_test_has_request_metadata,
+    md_only_test_has_request_metadata_only, md_only_test_get_request_metadata,
     NULL};
 
-grpc_credentials *grpc_fake_oauth2_credentials_create(
-    const char *token_md_value, int is_async) {
-  grpc_fake_oauth2_credentials *c =
-      gpr_malloc(sizeof(grpc_fake_oauth2_credentials));
-  memset(c, 0, sizeof(grpc_fake_oauth2_credentials));
+grpc_credentials *grpc_md_only_test_credentials_create(const char *md_key,
+                                                       const char *md_value,
+                                                       int is_async) {
+  grpc_md_only_test_credentials *c =
+      gpr_malloc(sizeof(grpc_md_only_test_credentials));
+  memset(c, 0, sizeof(grpc_md_only_test_credentials));
   c->base.type = GRPC_CREDENTIALS_TYPE_OAUTH2;
-  c->base.vtable = &fake_oauth2_vtable;
+  c->base.vtable = &md_only_test_vtable;
   gpr_ref_init(&c->base.refcount, 1);
-  c->access_token_md = grpc_credentials_md_store_create(1);
-  grpc_credentials_md_store_add_cstrings(
-      c->access_token_md, GRPC_AUTHORIZATION_METADATA_KEY, token_md_value);
+  c->md_store = grpc_credentials_md_store_create(1);
+  grpc_credentials_md_store_add_cstrings(c->md_store, md_key, md_value);
   c->is_async = is_async;
   return &c->base;
 }
diff --git a/src/core/security/credentials.h b/src/core/security/credentials.h
index 8d40da47c1..04736525dc 100644
--- a/src/core/security/credentials.h
+++ b/src/core/security/credentials.h
@@ -190,9 +190,10 @@ grpc_oauth2_token_fetcher_credentials_parse_server_response(
     grpc_credentials_md_store **token_md, gpr_timespec *token_lifetime);
 void grpc_flush_cached_google_default_credentials(void);
 
-/* Simulates an oauth2 token fetch with the specified value for testing. */
-grpc_credentials *grpc_fake_oauth2_credentials_create(
-    const char *token_md_value, int is_async);
+/* Metadata-only credentials with the specified key and value where
+   asynchronicity can be simulated for testing. */
+grpc_credentials *grpc_md_only_test_credentials_create(
+    const char *md_key, const char *md_value, int is_async);
 
 /* Private constructor for jwt credentials from an already parsed json key.
    Takes ownership of the key. */
@@ -216,6 +217,7 @@ typedef struct {
 struct grpc_server_credentials {
   const grpc_server_credentials_vtable *vtable;
   const char *type;
+  grpc_auth_metadata_processor processor;
 };
 
 grpc_security_status grpc_server_credentials_create_security_connector(
@@ -297,13 +299,13 @@ typedef struct {
   grpc_credentials_md_store *access_token_md;
 } grpc_access_token_credentials;
 
-/* -- Fake Oauth2 credentials. -- */
+/* --  Metadata-only Test credentials. -- */
 
 typedef struct {
   grpc_credentials base;
-  grpc_credentials_md_store *access_token_md;
+  grpc_credentials_md_store *md_store;
   int is_async;
-} grpc_fake_oauth2_credentials;
+} grpc_md_only_test_credentials;
 
 /* -- IAM credentials. -- */
 
diff --git a/src/core/security/google_default_credentials.c b/src/core/security/google_default_credentials.c
index f368819597..d1f228665f 100644
--- a/src/core/security/google_default_credentials.c
+++ b/src/core/security/google_default_credentials.c
@@ -84,6 +84,8 @@ static void on_compute_engine_detection_http_response(
   gpr_mu_unlock(GRPC_POLLSET_MU(&detector->pollset));
 }
 
+static void destroy_pollset(void *p) { grpc_pollset_destroy(p); }
+
 static int is_stack_running_on_compute_engine(void) {
   compute_engine_detector detector;
   grpc_httpcli_request request;
@@ -114,12 +116,12 @@ static int is_stack_running_on_compute_engine(void) {
   while (!detector.is_done) {
     grpc_pollset_worker worker;
     grpc_pollset_work(&detector.pollset, &worker,
-                      gpr_inf_future(GPR_CLOCK_REALTIME));
+                      gpr_inf_future(GPR_CLOCK_MONOTONIC));
   }
   gpr_mu_unlock(GRPC_POLLSET_MU(&detector.pollset));
 
   grpc_httpcli_context_destroy(&context);
-  grpc_pollset_destroy(&detector.pollset);
+  grpc_pollset_shutdown(&detector.pollset, destroy_pollset, &detector.pollset);
 
   return detector.success;
 }
diff --git a/src/core/security/jwt_verifier.c b/src/core/security/jwt_verifier.c
index 1276693da7..38ad134a6a 100644
--- a/src/core/security/jwt_verifier.c
+++ b/src/core/security/jwt_verifier.c
@@ -628,7 +628,7 @@ static void on_openid_config_retrieved(void *user_data,
     goto error;
   }
   jwks_uri += 8;
-  req.use_ssl = 1;
+  req.handshaker = &grpc_httpcli_ssl;
   req.host = gpr_strdup(jwks_uri);
   req.path = strchr(jwks_uri, '/');
   if (req.path == NULL) {
@@ -685,7 +685,7 @@ static void retrieve_key_and_verify(verifier_cb_ctx *ctx) {
   const char *iss;
   grpc_httpcli_request req;
   memset(&req, 0, sizeof(grpc_httpcli_request));
-  req.use_ssl = 1;
+  req.handshaker = &grpc_httpcli_ssl;
 
   GPR_ASSERT(ctx != NULL && ctx->header != NULL && ctx->claims != NULL);
   iss = ctx->claims->iss;
diff --git a/src/core/security/security_connector.c b/src/core/security/security_connector.c
index 726b4c1e12..a354536dcd 100644
--- a/src/core/security/security_connector.c
+++ b/src/core/security/security_connector.c
@@ -263,9 +263,9 @@ static grpc_security_status fake_check_peer(grpc_security_connector *sc,
     goto end;
   }
   GRPC_AUTH_CONTEXT_UNREF(sc->auth_context, "connector");
-  sc->auth_context = grpc_auth_context_create(NULL, 1);
-  sc->auth_context->properties[0] = grpc_auth_property_init_from_cstring(
-      GRPC_TRANSPORT_SECURITY_TYPE_PROPERTY_NAME,
+  sc->auth_context = grpc_auth_context_create(NULL);
+  grpc_auth_context_add_cstring_property(
+      sc->auth_context, GRPC_TRANSPORT_SECURITY_TYPE_PROPERTY_NAME,
       GRPC_FAKE_TRANSPORT_SECURITY_TYPE);
 
 end:
@@ -409,31 +409,35 @@ static int ssl_host_matches_name(const tsi_peer *peer, const char *peer_name) {
 grpc_auth_context *tsi_ssl_peer_to_auth_context(const tsi_peer *peer) {
   size_t i;
   grpc_auth_context *ctx = NULL;
+  const char *peer_identity_property_name = NULL;
 
   /* The caller has checked the certificate type property. */
   GPR_ASSERT(peer->property_count >= 1);
-  ctx = grpc_auth_context_create(NULL, peer->property_count);
-  ctx->properties[0] = grpc_auth_property_init_from_cstring(
-      GRPC_TRANSPORT_SECURITY_TYPE_PROPERTY_NAME,
+  ctx = grpc_auth_context_create(NULL);
+  grpc_auth_context_add_cstring_property(
+      ctx, GRPC_TRANSPORT_SECURITY_TYPE_PROPERTY_NAME,
       GRPC_SSL_TRANSPORT_SECURITY_TYPE);
-  ctx->property_count = 1;
   for (i = 0; i < peer->property_count; i++) {
     const tsi_peer_property *prop = &peer->properties[i];
     if (prop->name == NULL) continue;
     if (strcmp(prop->name, TSI_X509_SUBJECT_COMMON_NAME_PEER_PROPERTY) == 0) {
       /* If there is no subject alt name, have the CN as the identity. */
-      if (ctx->peer_identity_property_name == NULL) {
-        ctx->peer_identity_property_name = GRPC_X509_CN_PROPERTY_NAME;
+      if (peer_identity_property_name == NULL) {
+        peer_identity_property_name = GRPC_X509_CN_PROPERTY_NAME;
       }
-      ctx->properties[ctx->property_count++] = grpc_auth_property_init(
-          GRPC_X509_CN_PROPERTY_NAME, prop->value.data, prop->value.length);
+      grpc_auth_context_add_property(ctx, GRPC_X509_CN_PROPERTY_NAME,
+                                     prop->value.data, prop->value.length);
     } else if (strcmp(prop->name,
                       TSI_X509_SUBJECT_ALTERNATIVE_NAME_PEER_PROPERTY) == 0) {
-      ctx->peer_identity_property_name = GRPC_X509_SAN_PROPERTY_NAME;
-      ctx->properties[ctx->property_count++] = grpc_auth_property_init(
-          GRPC_X509_SAN_PROPERTY_NAME, prop->value.data, prop->value.length);
+      peer_identity_property_name = GRPC_X509_SAN_PROPERTY_NAME;
+      grpc_auth_context_add_property(ctx, GRPC_X509_SAN_PROPERTY_NAME,
+                                     prop->value.data, prop->value.length);
     }
   }
+  if (peer_identity_property_name != NULL) {
+    GPR_ASSERT(grpc_auth_context_set_peer_identity_property_name(
+                   ctx, peer_identity_property_name) == 1);
+  }
   return ctx;
 }
 
diff --git a/src/core/security/security_context.c b/src/core/security/security_context.c
index 8ce7876bd8..1ef0fc9255 100644
--- a/src/core/security/security_context.c
+++ b/src/core/security/security_context.c
@@ -42,6 +42,19 @@
 #include <grpc/support/log.h>
 #include <grpc/support/string_util.h>
 
+/* --- grpc_process_auth_metadata_func --- */
+
+static grpc_auth_metadata_processor server_processor = {NULL, NULL};
+
+grpc_auth_metadata_processor grpc_server_get_auth_metadata_processor(void) {
+  return server_processor;
+}
+
+void grpc_server_register_auth_metadata_processor(
+    grpc_auth_metadata_processor processor) {
+  server_processor = processor;
+}
+
 /* --- grpc_call --- */
 
 grpc_call_error grpc_call_set_credentials(grpc_call *call,
@@ -120,15 +133,15 @@ void grpc_server_security_context_destroy(void *ctx) {
 
 static grpc_auth_property_iterator empty_iterator = {NULL, 0, NULL};
 
-grpc_auth_context *grpc_auth_context_create(grpc_auth_context *chained,
-                                            size_t property_count) {
+grpc_auth_context *grpc_auth_context_create(grpc_auth_context *chained) {
   grpc_auth_context *ctx = gpr_malloc(sizeof(grpc_auth_context));
   memset(ctx, 0, sizeof(grpc_auth_context));
-  ctx->properties = gpr_malloc(property_count * sizeof(grpc_auth_property));
-  memset(ctx->properties, 0, property_count * sizeof(grpc_auth_property));
-  ctx->property_count = property_count;
   gpr_ref_init(&ctx->refcount, 1);
-  if (chained != NULL) ctx->chained = GRPC_AUTH_CONTEXT_REF(chained, "chained");
+  if (chained != NULL) {
+    ctx->chained = GRPC_AUTH_CONTEXT_REF(chained, "chained");
+    ctx->peer_identity_property_name =
+        ctx->chained->peer_identity_property_name;
+  }
   return ctx;
 }
 
@@ -162,11 +175,11 @@ void grpc_auth_context_unref(grpc_auth_context *ctx) {
   if (gpr_unref(&ctx->refcount)) {
     size_t i;
     GRPC_AUTH_CONTEXT_UNREF(ctx->chained, "chained");
-    if (ctx->properties != NULL) {
-      for (i = 0; i < ctx->property_count; i++) {
-        grpc_auth_property_reset(&ctx->properties[i]);
+    if (ctx->properties.array != NULL) {
+      for (i = 0; i < ctx->properties.count; i++) {
+        grpc_auth_property_reset(&ctx->properties.array[i]);
       }
-      gpr_free(ctx->properties);
+      gpr_free(ctx->properties.array);
     }
     gpr_free(ctx);
   }
@@ -177,6 +190,20 @@ const char *grpc_auth_context_peer_identity_property_name(
   return ctx->peer_identity_property_name;
 }
 
+int grpc_auth_context_set_peer_identity_property_name(grpc_auth_context *ctx,
+                                                      const char *name) {
+  grpc_auth_property_iterator it =
+      grpc_auth_context_find_properties_by_name(ctx, name);
+  const grpc_auth_property *prop = grpc_auth_property_iterator_next(&it);
+  if (prop == NULL) {
+    gpr_log(GPR_ERROR, "Property name %s not found in auth context.",
+            name != NULL ? name : "NULL");
+    return 0;
+  }
+  ctx->peer_identity_property_name = prop->name;
+  return 1;
+}
+
 int grpc_auth_context_peer_is_authenticated(
     const grpc_auth_context *ctx) {
   return ctx->peer_identity_property_name == NULL ? 0 : 1;
@@ -193,16 +220,16 @@ grpc_auth_property_iterator grpc_auth_context_property_iterator(
 const grpc_auth_property *grpc_auth_property_iterator_next(
     grpc_auth_property_iterator *it) {
   if (it == NULL || it->ctx == NULL) return NULL;
-  while (it->index == it->ctx->property_count) {
+  while (it->index == it->ctx->properties.count) {
     if (it->ctx->chained == NULL) return NULL;
     it->ctx = it->ctx->chained;
     it->index = 0;
   }
   if (it->name == NULL) {
-    return &it->ctx->properties[it->index++];
+    return &it->ctx->properties.array[it->index++];
   } else {
-    while (it->index < it->ctx->property_count) {
-      const grpc_auth_property *prop = &it->ctx->properties[it->index++];
+    while (it->index < it->ctx->properties.count) {
+      const grpc_auth_property *prop = &it->ctx->properties.array[it->index++];
       GPR_ASSERT(prop->name != NULL);
       if (strcmp(it->name, prop->name) == 0) {
         return prop;
@@ -229,24 +256,37 @@ grpc_auth_property_iterator grpc_auth_context_peer_identity(
       ctx, ctx->peer_identity_property_name);
 }
 
-grpc_auth_property grpc_auth_property_init_from_cstring(const char *name,
-                                                        const char *value) {
-  grpc_auth_property prop;
-  prop.name = gpr_strdup(name);
-  prop.value = gpr_strdup(value);
-  prop.value_length = strlen(value);
-  return prop;
+static void ensure_auth_context_capacity(grpc_auth_context *ctx) {
+  if (ctx->properties.count == ctx->properties.capacity) {
+    ctx->properties.capacity =
+        GPR_MAX(ctx->properties.capacity + 8, ctx->properties.capacity * 2);
+    ctx->properties.array =
+        gpr_realloc(ctx->properties.array,
+                    ctx->properties.capacity * sizeof(grpc_auth_property));
+  }
 }
 
-grpc_auth_property grpc_auth_property_init(const char *name, const char *value,
-                                           size_t value_length) {
-  grpc_auth_property prop;
-  prop.name = gpr_strdup(name);
-  prop.value = gpr_malloc(value_length + 1);
-  memcpy(prop.value, value, value_length);
-  prop.value[value_length] = '\0';
-  prop.value_length = value_length;
-  return prop;
+void grpc_auth_context_add_property(grpc_auth_context *ctx, const char *name,
+                                    const char *value, size_t value_length) {
+  grpc_auth_property *prop;
+  ensure_auth_context_capacity(ctx);
+  prop = &ctx->properties.array[ctx->properties.count++];
+  prop->name = gpr_strdup(name);
+  prop->value = gpr_malloc(value_length + 1);
+  memcpy(prop->value, value, value_length);
+  prop->value[value_length] = '\0';
+  prop->value_length = value_length;
+}
+
+void grpc_auth_context_add_cstring_property(grpc_auth_context *ctx,
+                                            const char *name,
+                                            const char *value) {
+  grpc_auth_property *prop;
+  ensure_auth_context_capacity(ctx);
+  prop = &ctx->properties.array[ctx->properties.count++];
+  prop->name = gpr_strdup(name);
+  prop->value = gpr_strdup(value);
+  prop->value_length = strlen(value);
 }
 
 void grpc_auth_property_reset(grpc_auth_property *property) {
@@ -255,3 +295,35 @@ void grpc_auth_property_reset(grpc_auth_property *property) {
   memset(property, 0, sizeof(grpc_auth_property));
 }
 
+grpc_arg grpc_auth_metadata_processor_to_arg(grpc_auth_metadata_processor *p) {
+  grpc_arg arg;
+  memset(&arg, 0, sizeof(grpc_arg));
+  arg.type = GRPC_ARG_POINTER;
+  arg.key = GRPC_AUTH_METADATA_PROCESSOR_ARG;
+  arg.value.pointer.p = p;
+  return arg;
+}
+
+grpc_auth_metadata_processor *grpc_auth_metadata_processor_from_arg(
+    const grpc_arg *arg) {
+  if (strcmp(arg->key, GRPC_AUTH_METADATA_PROCESSOR_ARG) != 0) return NULL;
+  if (arg->type != GRPC_ARG_POINTER) {
+    gpr_log(GPR_ERROR, "Invalid type %d for arg %s", arg->type,
+            GRPC_AUTH_METADATA_PROCESSOR_ARG);
+    return NULL;
+  }
+  return arg->value.pointer.p;
+}
+
+grpc_auth_metadata_processor *grpc_find_auth_metadata_processor_in_args(
+    const grpc_channel_args *args) {
+  size_t i;
+  if (args == NULL) return NULL;
+  for (i = 0; i < args->num_args; i++) {
+    grpc_auth_metadata_processor *p =
+        grpc_auth_metadata_processor_from_arg(&args->args[i]);
+    if (p != NULL) return p;
+  }
+  return NULL;
+}
+
diff --git a/src/core/security/security_context.h b/src/core/security/security_context.h
index 76a45910bb..7fcd438cf6 100644
--- a/src/core/security/security_context.h
+++ b/src/core/security/security_context.h
@@ -34,29 +34,31 @@
 #ifndef GRPC_INTERNAL_CORE_SECURITY_SECURITY_CONTEXT_H
 #define GRPC_INTERNAL_CORE_SECURITY_SECURITY_CONTEXT_H
 
+#include "src/core/iomgr/pollset.h"
 #include "src/core/security/credentials.h"
 
-#ifdef __cplusplus
-extern "C" {
-#endif
-
 /* --- grpc_auth_context ---
 
    High level authentication context object. Can optionally be chained. */
 
 /* Property names are always NULL terminated. */
 
+typedef struct {
+  grpc_auth_property *array;
+  size_t count;
+  size_t capacity;
+} grpc_auth_property_array;
+
 struct grpc_auth_context {
   struct grpc_auth_context *chained;
-  grpc_auth_property *properties;
-  size_t property_count;
+  grpc_auth_property_array properties;
   gpr_refcount refcount;
   const char *peer_identity_property_name;
+  grpc_pollset *pollset;
 };
 
-/* Constructor. */
-grpc_auth_context *grpc_auth_context_create(grpc_auth_context *chained,
-                                            size_t property_count);
+/* Creation. */
+grpc_auth_context *grpc_auth_context_create(grpc_auth_context *chained);
 
 /* Refcounting. */
 #ifdef GRPC_AUTH_CONTEXT_REFCOUNT_DEBUG
@@ -76,12 +78,6 @@ grpc_auth_context *grpc_auth_context_ref(grpc_auth_context *policy);
 void grpc_auth_context_unref(grpc_auth_context *policy);
 #endif
 
-grpc_auth_property grpc_auth_property_init_from_cstring(const char *name,
-                                                        const char *value);
-
-grpc_auth_property grpc_auth_property_init(const char *name, const char *value,
-                                           size_t value_length);
-
 void grpc_auth_property_reset(grpc_auth_property *property);
 
 /* --- grpc_client_security_context ---
@@ -107,9 +103,14 @@ typedef struct {
 grpc_server_security_context *grpc_server_security_context_create(void);
 void grpc_server_security_context_destroy(void *ctx);
 
-#ifdef __cplusplus
-}
-#endif
+/* --- Auth metadata processing. --- */
+#define GRPC_AUTH_METADATA_PROCESSOR_ARG "grpc.auth_metadata_processor"
+
+grpc_arg grpc_auth_metadata_processor_to_arg(grpc_auth_metadata_processor *p);
+grpc_auth_metadata_processor *grpc_auth_metadata_processor_from_arg(
+    const grpc_arg *arg);
+grpc_auth_metadata_processor *grpc_find_auth_metadata_processor_in_args(
+    const grpc_channel_args *args);
 
 #endif  /* GRPC_INTERNAL_CORE_SECURITY_SECURITY_CONTEXT_H */
 
diff --git a/src/core/security/server_auth_filter.c b/src/core/security/server_auth_filter.c
index 69789c2f0d..2fc689caec 100644
--- a/src/core/security/server_auth_filter.c
+++ b/src/core/security/server_auth_filter.c
@@ -31,20 +31,140 @@
  *
  */
 
+#include <string.h>
+
 #include "src/core/security/auth_filters.h"
 #include "src/core/security/security_connector.h"
 #include "src/core/security/security_context.h"
 
+#include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
 
 typedef struct call_data {
-  int unused; /* C89 requires at least one struct element */
+  gpr_uint8 got_client_metadata;
+  grpc_stream_op_buffer *recv_ops;
+  /* Closure to call when finished with the auth_on_recv hook. */
+  grpc_iomgr_closure *on_done_recv;
+  /* Receive closures are chained: we inject this closure as the on_done_recv
+     up-call on transport_op, and remember to call our on_done_recv member after
+     handling it. */
+  grpc_iomgr_closure auth_on_recv;
+  grpc_transport_stream_op transport_op;
+  const grpc_metadata *consumed_md;
+  size_t num_consumed_md;
+  grpc_stream_op *md_op;
+  grpc_auth_context *auth_context;
 } call_data;
 
 typedef struct channel_data {
   grpc_security_connector *security_connector;
+  grpc_auth_metadata_processor processor;
+  grpc_mdctx *mdctx;
 } channel_data;
 
+static grpc_metadata_array metadata_batch_to_md_array(
+    const grpc_metadata_batch *batch) {
+  grpc_linked_mdelem *l;
+  grpc_metadata_array result;
+  grpc_metadata_array_init(&result);
+  for (l = batch->list.head; l != NULL; l = l->next) {
+    grpc_metadata *usr_md = NULL;
+    grpc_mdelem *md = l->md;
+    grpc_mdstr *key = md->key;
+    grpc_mdstr *value = md->value;
+    if (result.count == result.capacity) {
+      result.capacity = GPR_MAX(result.capacity + 8, result.capacity * 2);
+      result.metadata =
+          gpr_realloc(result.metadata, result.capacity * sizeof(grpc_metadata));
+    }
+    usr_md = &result.metadata[result.count++];
+    usr_md->key = grpc_mdstr_as_c_string(key);
+    usr_md->value = grpc_mdstr_as_c_string(value);
+    usr_md->value_length = GPR_SLICE_LENGTH(value->slice);
+  }
+  return result;
+}
+
+static grpc_mdelem *remove_consumed_md(void *user_data, grpc_mdelem *md) {
+  grpc_call_element *elem = user_data;
+  call_data *calld = elem->call_data;
+  size_t i;
+  for (i = 0; i < calld->num_consumed_md; i++) {
+    /* Maybe we could do a pointer comparison but we do not have any guarantee
+       that the metadata processor used the same pointers for consumed_md in the
+       callback. */
+    if (memcmp(GPR_SLICE_START_PTR(md->key->slice), calld->consumed_md[i].key,
+               GPR_SLICE_LENGTH(md->key->slice)) == 0 &&
+        memcmp(GPR_SLICE_START_PTR(md->value->slice),
+               calld->consumed_md[i].value,
+               GPR_SLICE_LENGTH(md->value->slice)) == 0) {
+      return NULL; /* Delete. */
+    }
+  }
+  return md;
+}
+
+static void on_md_processing_done(void *user_data,
+                                  const grpc_metadata *consumed_md,
+                                  size_t num_consumed_md, int success) {
+  grpc_call_element *elem = user_data;
+  call_data *calld = elem->call_data;
+
+  if (success) {
+    calld->consumed_md = consumed_md;
+    calld->num_consumed_md = num_consumed_md;
+    grpc_metadata_batch_filter(&calld->md_op->data.metadata, remove_consumed_md,
+                               elem);
+    calld->on_done_recv->cb(calld->on_done_recv->cb_arg, success);
+  } else {
+    gpr_slice message = gpr_slice_from_copied_string(
+        "Authentication metadata processing failed.");
+    grpc_sopb_reset(calld->recv_ops);
+    grpc_transport_stream_op_add_close(&calld->transport_op,
+                                       GRPC_STATUS_UNAUTHENTICATED, &message);
+    grpc_call_next_op(elem, &calld->transport_op);
+  }
+}
+
+static void auth_on_recv(void *user_data, int success) {
+  grpc_call_element *elem = user_data;
+  call_data *calld = elem->call_data;
+  channel_data *chand = elem->channel_data;
+  if (success) {
+    size_t i;
+    size_t nops = calld->recv_ops->nops;
+    grpc_stream_op *ops = calld->recv_ops->ops;
+    for (i = 0; i < nops; i++) {
+      grpc_metadata_array md_array;
+      grpc_stream_op *op = &ops[i];
+      if (op->type != GRPC_OP_METADATA || calld->got_client_metadata) continue;
+      calld->got_client_metadata = 1;
+      if (chand->processor.process == NULL) continue;
+      calld->md_op = op;
+      md_array = metadata_batch_to_md_array(&op->data.metadata);
+      chand->processor.process(chand->processor.state, calld->auth_context,
+                               md_array.metadata, md_array.count,
+                               on_md_processing_done, elem);
+      grpc_metadata_array_destroy(&md_array);
+      return;
+    }
+  }
+  calld->on_done_recv->cb(calld->on_done_recv->cb_arg, success);
+}
+
+static void set_recv_ops_md_callbacks(grpc_call_element *elem,
+                                      grpc_transport_stream_op *op) {
+  call_data *calld = elem->call_data;
+
+  if (op->recv_ops && !calld->got_client_metadata) {
+    /* substitute our callback for the higher callback */
+    calld->recv_ops = op->recv_ops;
+    calld->on_done_recv = op->on_done_recv;
+    op->on_done_recv = &calld->auth_on_recv;
+    calld->transport_op = *op;
+  }
+}
+
 /* Called either:
      - in response to an API call (or similar) from above, to send something
      - a network event (or similar) from below, to receive something
@@ -52,9 +172,7 @@ typedef struct channel_data {
    that is being sent or received. */
 static void auth_start_transport_op(grpc_call_element *elem,
                                     grpc_transport_stream_op *op) {
-  /* TODO(jboeuf): Get the metadata and get a new context from it. */
-
-  /* pass control down the stack */
+  set_recv_ops_md_callbacks(elem, op);
   grpc_call_next_op(elem, op);
 }
 
@@ -68,7 +186,8 @@ static void init_call_elem(grpc_call_element *elem,
   grpc_server_security_context *server_ctx = NULL;
 
   /* initialize members */
-  calld->unused = 0;
+  memset(calld, 0, sizeof(*calld));
+  grpc_iomgr_closure_init(&calld->auth_on_recv, auth_on_recv, elem);
 
   GPR_ASSERT(initial_op && initial_op->context != NULL &&
              initial_op->context[GRPC_CONTEXT_SECURITY].value == NULL);
@@ -80,21 +199,29 @@ static void init_call_elem(grpc_call_element *elem,
         initial_op->context[GRPC_CONTEXT_SECURITY].value);
   }
   server_ctx = grpc_server_security_context_create();
-  server_ctx->auth_context = GRPC_AUTH_CONTEXT_REF(
-      chand->security_connector->auth_context, "server_security_context");
+  server_ctx->auth_context =
+      grpc_auth_context_create(chand->security_connector->auth_context);
+  server_ctx->auth_context->pollset = initial_op->bind_pollset;
   initial_op->context[GRPC_CONTEXT_SECURITY].value = server_ctx;
   initial_op->context[GRPC_CONTEXT_SECURITY].destroy =
       grpc_server_security_context_destroy;
+  calld->auth_context = server_ctx->auth_context;
+
+  /* Set the metadata callbacks. */
+  set_recv_ops_md_callbacks(elem, initial_op);
 }
 
 /* Destructor for call_data */
-static void destroy_call_elem(grpc_call_element *elem) {}
+static void destroy_call_elem(grpc_call_element *elem) {
+}
 
 /* Constructor for channel_data */
 static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master,
                               const grpc_channel_args *args, grpc_mdctx *mdctx,
                               int is_first, int is_last) {
   grpc_security_connector *sc = grpc_find_security_connector_in_args(args);
+  grpc_auth_metadata_processor *processor =
+      grpc_find_auth_metadata_processor_in_args(args);
   /* grab pointers to our data from the channel element */
   channel_data *chand = elem->channel_data;
 
@@ -104,11 +231,14 @@ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master,
   GPR_ASSERT(!is_first);
   GPR_ASSERT(!is_last);
   GPR_ASSERT(sc != NULL);
+  GPR_ASSERT(processor != NULL);
 
   /* initialize members */
   GPR_ASSERT(!sc->is_client_side);
   chand->security_connector =
       GRPC_SECURITY_CONNECTOR_REF(sc, "server_auth_filter");
+  chand->mdctx = mdctx;
+  chand->processor = *processor;
 }
 
 /* Destructor for channel data */
diff --git a/src/core/security/server_secure_chttp2.c b/src/core/security/server_secure_chttp2.c
index 3717b8989f..8d9d036d80 100644
--- a/src/core/security/server_secure_chttp2.c
+++ b/src/core/security/server_secure_chttp2.c
@@ -43,6 +43,7 @@
 #include "src/core/security/auth_filters.h"
 #include "src/core/security/credentials.h"
 #include "src/core/security/security_connector.h"
+#include "src/core/security/security_context.h"
 #include "src/core/security/secure_transport_setup.h"
 #include "src/core/surface/server.h"
 #include "src/core/transport/chttp2_transport.h"
@@ -60,6 +61,7 @@ typedef struct grpc_server_secure_state {
   grpc_server *server;
   grpc_tcp_server *tcp;
   grpc_security_connector *sc;
+  grpc_auth_metadata_processor processor;
   tcp_endpoint_list *handshaking_tcp_endpoints;
   int is_shutdown;
   gpr_mu mu;
@@ -86,9 +88,13 @@ static void setup_transport(void *statep, grpc_transport *transport,
   static grpc_channel_filter const *extra_filters[] = {
       &grpc_server_auth_filter, &grpc_http_server_filter};
   grpc_server_secure_state *state = statep;
-  grpc_arg connector_arg = grpc_security_connector_to_arg(state->sc);
-  grpc_channel_args *args_copy = grpc_channel_args_copy_and_add(
-      grpc_server_get_channel_args(state->server), &connector_arg, 1);
+  grpc_channel_args *args_copy;
+  grpc_arg args_to_add[2];
+  args_to_add[0] = grpc_security_connector_to_arg(state->sc);
+  args_to_add[1] = grpc_auth_metadata_processor_to_arg(&state->processor);
+  args_copy = grpc_channel_args_copy_and_add(
+      grpc_server_get_channel_args(state->server), args_to_add,
+      GPR_ARRAY_SIZE(args_to_add));
   grpc_server_setup_transport(state->server, transport, extra_filters,
                               GPR_ARRAY_SIZE(extra_filters), mdctx, args_copy);
   grpc_channel_args_destroy(args_copy);
@@ -252,9 +258,11 @@ int grpc_server_add_secure_http2_port(grpc_server *server, const char *addr,
   grpc_resolved_addresses_destroy(resolved);
 
   state = gpr_malloc(sizeof(*state));
+  memset(state, 0, sizeof(*state));
   state->server = server;
   state->tcp = tcp;
   state->sc = sc;
+  state->processor = creds->processor;
   state->handshaking_tcp_endpoints = NULL;
   state->is_shutdown = 0;
   gpr_mu_init(&state->mu);
diff --git a/src/core/surface/call.c b/src/core/surface/call.c
index d3e66e9c4c..5839d3ac2e 100644
--- a/src/core/surface/call.c
+++ b/src/core/surface/call.c
@@ -456,20 +456,6 @@ void grpc_call_internal_ref(grpc_call *c) {
 static void destroy_call(void *call, int ignored_success) {
   size_t i;
   grpc_call *c = call;
-  grpc_call *parent = c->parent;
-  if (parent) {
-    gpr_mu_lock(&parent->mu);
-    if (call == parent->first_child) {
-      parent->first_child = c->sibling_next;
-      if (c == parent->first_child) {
-        parent->first_child = NULL;
-      }
-      c->sibling_prev->sibling_next = c->sibling_next;
-      c->sibling_next->sibling_prev = c->sibling_prev;
-    }
-    gpr_mu_unlock(&parent->mu);
-    GRPC_CALL_INTERNAL_UNREF(parent, "child", 1);
-  }
   grpc_call_stack_destroy(CALL_STACK_FROM_CALL(c));
   GRPC_CHANNEL_INTERNAL_UNREF(c->channel, "call");
   gpr_mu_destroy(&c->mu);
@@ -1257,6 +1243,22 @@ grpc_call_error grpc_call_start_ioreq_and_call_back(
 
 void grpc_call_destroy(grpc_call *c) {
   int cancel;
+  grpc_call *parent = c->parent;
+
+  if (parent) {
+    gpr_mu_lock(&parent->mu);
+    if (c == parent->first_child) {
+      parent->first_child = c->sibling_next;
+      if (c == parent->first_child) {
+        parent->first_child = NULL;
+      }
+      c->sibling_prev->sibling_next = c->sibling_next;
+      c->sibling_next->sibling_prev = c->sibling_prev;
+    }
+    gpr_mu_unlock(&parent->mu);
+    GRPC_CALL_INTERNAL_UNREF(parent, "child", 1);
+  }
+
   lock(c);
   GPR_ASSERT(!c->destroy_called);
   c->destroy_called = 1;
@@ -1537,6 +1539,7 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops,
         /* Flag validation: currently allow no flags */
         if (op->flags != 0) return GRPC_CALL_ERROR_INVALID_FLAGS;
         req = &reqs[out++];
+	if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG;
         req->op = GRPC_IOREQ_SEND_INITIAL_METADATA;
         req->data.send_metadata.count = op->data.send_initial_metadata.count;
         req->data.send_metadata.metadata =
@@ -1551,6 +1554,7 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops,
           return GRPC_CALL_ERROR_INVALID_MESSAGE;
         }
         req = &reqs[out++];
+	if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG;
         req->op = GRPC_IOREQ_SEND_MESSAGE;
         req->data.send_message = op->data.send_message;
         req->flags = op->flags;
@@ -1562,6 +1566,7 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops,
           return GRPC_CALL_ERROR_NOT_ON_SERVER;
         }
         req = &reqs[out++];
+	if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG;
         req->op = GRPC_IOREQ_SEND_CLOSE;
         req->flags = op->flags;
         break;
@@ -1572,6 +1577,7 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops,
           return GRPC_CALL_ERROR_NOT_ON_CLIENT;
         }
         req = &reqs[out++];
+	if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG;
         req->op = GRPC_IOREQ_SEND_TRAILING_METADATA;
         req->flags = op->flags;
         req->data.send_metadata.count =
@@ -1579,6 +1585,7 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops,
         req->data.send_metadata.metadata =
             op->data.send_status_from_server.trailing_metadata;
         req = &reqs[out++];
+	if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG;
         req->op = GRPC_IOREQ_SEND_STATUS;
         req->data.send_status.code = op->data.send_status_from_server.status;
         req->data.send_status.details =
@@ -1588,6 +1595,7 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops,
                       op->data.send_status_from_server.status_details, 0)
                 : NULL;
         req = &reqs[out++];
+	if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG;
         req->op = GRPC_IOREQ_SEND_CLOSE;
         break;
       case GRPC_OP_RECV_INITIAL_METADATA:
@@ -1597,6 +1605,7 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops,
           return GRPC_CALL_ERROR_NOT_ON_SERVER;
         }
         req = &reqs[out++];
+	if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG;
         req->op = GRPC_IOREQ_RECV_INITIAL_METADATA;
         req->data.recv_metadata = op->data.recv_initial_metadata;
         req->data.recv_metadata->count = 0;
@@ -1606,6 +1615,7 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops,
         /* Flag validation: currently allow no flags */
         if (op->flags != 0) return GRPC_CALL_ERROR_INVALID_FLAGS;
         req = &reqs[out++];
+	if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG;
         req->op = GRPC_IOREQ_RECV_MESSAGE;
         req->data.recv_message = op->data.recv_message;
         req->flags = op->flags;
@@ -1617,22 +1627,26 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops,
           return GRPC_CALL_ERROR_NOT_ON_SERVER;
         }
         req = &reqs[out++];
+	if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG;
         req->op = GRPC_IOREQ_RECV_STATUS;
         req->flags = op->flags;
         req->data.recv_status.set_value = set_status_value_directly;
         req->data.recv_status.user_data = op->data.recv_status_on_client.status;
         req = &reqs[out++];
+	if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG;
         req->op = GRPC_IOREQ_RECV_STATUS_DETAILS;
         req->data.recv_status_details.details =
             op->data.recv_status_on_client.status_details;
         req->data.recv_status_details.details_capacity =
             op->data.recv_status_on_client.status_details_capacity;
         req = &reqs[out++];
+	if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG;
         req->op = GRPC_IOREQ_RECV_TRAILING_METADATA;
         req->data.recv_metadata =
             op->data.recv_status_on_client.trailing_metadata;
         req->data.recv_metadata->count = 0;
         req = &reqs[out++];
+	if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG;
         req->op = GRPC_IOREQ_RECV_CLOSE;
         finish_func = finish_batch_with_close;
         break;
@@ -1640,12 +1654,14 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops,
         /* Flag validation: currently allow no flags */
         if (op->flags != 0) return GRPC_CALL_ERROR_INVALID_FLAGS;
         req = &reqs[out++];
+	if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG;
         req->op = GRPC_IOREQ_RECV_STATUS;
         req->flags = op->flags;
         req->data.recv_status.set_value = set_cancelled_value;
         req->data.recv_status.user_data =
             op->data.recv_close_on_server.cancelled;
         req = &reqs[out++];
+	if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG;
         req->op = GRPC_IOREQ_RECV_CLOSE;
         finish_func = finish_batch_with_close;
         break;
diff --git a/src/core/surface/secure_channel_create.c b/src/core/surface/secure_channel_create.c
index 1f89353025..c3150250b8 100644
--- a/src/core/surface/secure_channel_create.c
+++ b/src/core/surface/secure_channel_create.c
@@ -88,8 +88,8 @@ static void on_secure_transport_setup_done(void *arg,
         c->args.channel_args, secure_endpoint, c->args.metadata_context, 1);
     grpc_chttp2_transport_start_reading(c->result->transport, NULL, 0);
     c->result->filters = gpr_malloc(sizeof(grpc_channel_filter *) * 2);
-    c->result->filters[0] = &grpc_client_auth_filter;
-    c->result->filters[1] = &grpc_http_client_filter;
+    c->result->filters[0] = &grpc_http_client_filter;
+    c->result->filters[1] = &grpc_client_auth_filter;
     c->result->num_filters = 2;
   }
   notify = c->notify;
diff --git a/src/core/surface/version.c b/src/core/surface/version.c
index 4f5d648371..d7aaba3868 100644
--- a/src/core/surface/version.c
+++ b/src/core/surface/version.c
@@ -37,5 +37,5 @@
 #include <grpc/grpc.h>
 
 const char *grpc_version_string(void) {
-	return "0.10.0.0";
+	return "0.10.1.0";
 }
diff --git a/src/core/transport/chttp2/internal.h b/src/core/transport/chttp2/internal.h
index cb428f8e3c..42cf0ecd5b 100644
--- a/src/core/transport/chttp2/internal.h
+++ b/src/core/transport/chttp2/internal.h
@@ -388,6 +388,8 @@ typedef struct {
   gpr_uint8 in_stream_map;
   /** bitmask of GRPC_CHTTP2_WRITING_xxx above */
   gpr_uint8 writing_now;
+  /** has anything been written to this stream? */
+  gpr_uint8 written_anything;
 
   /** stream state already published to the upper layer */
   grpc_stream_state published_state;
diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c
index 6ba144faa4..a9f91b64d5 100644
--- a/src/core/transport/chttp2_transport.c
+++ b/src/core/transport/chttp2_transport.c
@@ -107,6 +107,11 @@ static void cancel_from_api(grpc_chttp2_transport_global *transport_global,
                             grpc_chttp2_stream_global *stream_global,
                             grpc_status_code status);
 
+static void close_from_api(grpc_chttp2_transport_global *transport_global,
+                           grpc_chttp2_stream_global *stream_global,
+                           grpc_status_code status,
+                           gpr_slice *optional_message);
+
 /** Add endpoint from this transport to pollset */
 static void add_to_pollset_locked(grpc_chttp2_transport *t,
                                   grpc_pollset *pollset);
@@ -602,10 +607,16 @@ static void perform_stream_op_locked(
     cancel_from_api(transport_global, stream_global, op->cancel_with_status);
   }
 
+  if (op->close_with_status != GRPC_STATUS_OK) {
+    close_from_api(transport_global, stream_global, op->close_with_status,
+                   op->optional_close_message);
+  }
+
   if (op->send_ops) {
     GPR_ASSERT(stream_global->outgoing_sopb == NULL);
     stream_global->send_done_closure = op->on_done_send;
     if (!stream_global->cancelled) {
+      stream_global->written_anything = 1;
       stream_global->outgoing_sopb = op->send_ops;
       if (op->is_last_send &&
           stream_global->write_state == GRPC_WRITE_STATE_OPEN) {
@@ -900,6 +911,108 @@ static void cancel_from_api(grpc_chttp2_transport_global *transport_global,
                                                 stream_global);
 }
 
+static void close_from_api(grpc_chttp2_transport_global *transport_global,
+                           grpc_chttp2_stream_global *stream_global,
+                           grpc_status_code status,
+                           gpr_slice *optional_message) {
+  gpr_slice hdr;
+  gpr_slice status_hdr;
+  gpr_slice message_pfx;
+  gpr_uint8 *p;
+  gpr_uint32 len = 0;
+
+  GPR_ASSERT(status >= 0 && (int)status < 100);
+
+  stream_global->cancelled = 1;
+  stream_global->cancelled_status = status;
+  GPR_ASSERT(stream_global->id != 0);
+  GPR_ASSERT(!stream_global->written_anything);
+
+  /* Hand roll a header block.
+     This is unnecessarily ugly - at some point we should find a more elegant
+     solution.
+     It's complicated by the fact that our send machinery would be dead by the
+     time we got around to sending this, so instead we ignore HPACK compression
+     and just write the uncompressed bytes onto the wire. */
+  status_hdr = gpr_slice_malloc(15 + (status >= 10));
+  p = GPR_SLICE_START_PTR(status_hdr);
+  *p++ = 0x40; /* literal header */
+  *p++ = 11;   /* len(grpc-status) */
+  *p++ = 'g';
+  *p++ = 'r';
+  *p++ = 'p';
+  *p++ = 'c';
+  *p++ = '-';
+  *p++ = 's';
+  *p++ = 't';
+  *p++ = 'a';
+  *p++ = 't';
+  *p++ = 'u';
+  *p++ = 's';
+  if (status < 10) {
+    *p++ = 1;
+    *p++ = '0' + status;
+  } else {
+    *p++ = 2;
+    *p++ = '0' + (status / 10);
+    *p++ = '0' + (status % 10);
+  }
+  GPR_ASSERT(p == GPR_SLICE_END_PTR(status_hdr));
+  len += GPR_SLICE_LENGTH(status_hdr);
+
+  if (optional_message) {
+    GPR_ASSERT(GPR_SLICE_LENGTH(*optional_message) < 127);
+    message_pfx = gpr_slice_malloc(15);
+    p = GPR_SLICE_START_PTR(message_pfx);
+    *p++ = 0x40;
+    *p++ = 12; /* len(grpc-message) */
+    *p++ = 'g';
+    *p++ = 'r';
+    *p++ = 'p';
+    *p++ = 'c';
+    *p++ = '-';
+    *p++ = 'm';
+    *p++ = 'e';
+    *p++ = 's';
+    *p++ = 's';
+    *p++ = 'a';
+    *p++ = 'g';
+    *p++ = 'e';
+    *p++ = GPR_SLICE_LENGTH(*optional_message);
+    GPR_ASSERT(p == GPR_SLICE_END_PTR(message_pfx));
+    len += GPR_SLICE_LENGTH(message_pfx);
+    len += GPR_SLICE_LENGTH(*optional_message);
+  }
+
+  hdr = gpr_slice_malloc(9);
+  p = GPR_SLICE_START_PTR(hdr);
+  *p++ = len >> 16;
+  *p++ = len >> 8;
+  *p++ = len;
+  *p++ = GRPC_CHTTP2_FRAME_HEADER;
+  *p++ = GRPC_CHTTP2_DATA_FLAG_END_STREAM | GRPC_CHTTP2_DATA_FLAG_END_HEADERS;
+  *p++ = stream_global->id >> 24;
+  *p++ = stream_global->id >> 16;
+  *p++ = stream_global->id >> 8;
+  *p++ = stream_global->id;
+  GPR_ASSERT(p == GPR_SLICE_END_PTR(hdr));
+
+  gpr_slice_buffer_add(&transport_global->qbuf, hdr);
+  gpr_slice_buffer_add(&transport_global->qbuf, status_hdr);
+  if (optional_message) {
+    gpr_slice_buffer_add(&transport_global->qbuf, message_pfx);
+    gpr_slice_buffer_add(&transport_global->qbuf,
+                         gpr_slice_ref(*optional_message));
+  }
+
+  gpr_slice_buffer_add(
+      &transport_global->qbuf,
+      grpc_chttp2_rst_stream_create(stream_global->id, GRPC_CHTTP2_NO_ERROR));
+
+  grpc_chttp2_list_add_read_write_state_changed(transport_global,
+                                                stream_global);
+}
+
 static void cancel_stream_cb(grpc_chttp2_transport_global *transport_global,
                              void *user_data,
                              grpc_chttp2_stream_global *stream_global) {
diff --git a/src/core/transport/metadata.c b/src/core/transport/metadata.c
index 967fd4898c..44d32b6cb2 100644
--- a/src/core/transport/metadata.c
+++ b/src/core/transport/metadata.c
@@ -135,7 +135,9 @@ static void unlock(grpc_mdctx *ctx) {
   if (ctx->refs == 0) {
     /* uncomment if you're having trouble diagnosing an mdelem leak to make
        things clearer (slows down destruction a lot, however) */
+#ifdef GRPC_METADATA_REFCOUNT_DEBUG
     gc_mdtab(ctx);
+#endif
     if (ctx->mdtab_count && ctx->mdtab_count == ctx->mdtab_free) {
       discard_metadata(ctx);
     }
diff --git a/src/core/transport/stream_op.h b/src/core/transport/stream_op.h
index f27ef1b66b..227320cf2a 100644
--- a/src/core/transport/stream_op.h
+++ b/src/core/transport/stream_op.h
@@ -108,7 +108,7 @@ void grpc_metadata_batch_move(grpc_metadata_batch *dst,
                               grpc_metadata_batch *src);
 
 /** Add \a storage to the beginning of \a batch. storage->md is
-    assumed to be valid. 
+    assumed to be valid.
     \a storage is owned by the caller and must survive for the
     lifetime of batch. This usually means it should be around
     for the lifetime of the call. */
diff --git a/src/core/transport/transport.c b/src/core/transport/transport.c
index 69c00b6a4f..c0d92cf93f 100644
--- a/src/core/transport/transport.c
+++ b/src/core/transport/transport.c
@@ -32,6 +32,8 @@
  */
 
 #include "src/core/transport/transport.h"
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
 #include "src/core/transport/transport_impl.h"
 
 size_t grpc_transport_stream_size(grpc_transport *transport) {
@@ -83,12 +85,54 @@ void grpc_transport_stream_op_finish_with_failure(
 }
 
 void grpc_transport_stream_op_add_cancellation(grpc_transport_stream_op *op,
-                                               grpc_status_code status,
-                                               grpc_mdstr *message) {
+                                               grpc_status_code status) {
+  GPR_ASSERT(status != GRPC_STATUS_OK);
   if (op->cancel_with_status == GRPC_STATUS_OK) {
     op->cancel_with_status = status;
   }
-  if (message) {
-    GRPC_MDSTR_UNREF(message);
+  if (op->close_with_status != GRPC_STATUS_OK) {
+    op->close_with_status = GRPC_STATUS_OK;
+    if (op->optional_close_message != NULL) {
+      gpr_slice_unref(*op->optional_close_message);
+      op->optional_close_message = NULL;
+    }
   }
 }
+
+typedef struct {
+  gpr_slice message;
+  grpc_iomgr_closure *then_call;
+  grpc_iomgr_closure closure;
+} close_message_data;
+
+static void free_message(void *p, int iomgr_success) {
+  close_message_data *cmd = p;
+  gpr_slice_unref(cmd->message);
+  if (cmd->then_call != NULL) {
+    cmd->then_call->cb(cmd->then_call->cb_arg, iomgr_success);
+  }
+  gpr_free(cmd);
+}
+
+void grpc_transport_stream_op_add_close(grpc_transport_stream_op *op,
+                                        grpc_status_code status,
+                                        gpr_slice *optional_message) {
+  close_message_data *cmd;
+  GPR_ASSERT(status != GRPC_STATUS_OK);
+  if (op->cancel_with_status != GRPC_STATUS_OK ||
+      op->close_with_status != GRPC_STATUS_OK) {
+    if (optional_message) {
+      gpr_slice_unref(*optional_message);
+    }
+    return;
+  }
+  if (optional_message) {
+    cmd = gpr_malloc(sizeof(*cmd));
+    cmd->message = *optional_message;
+    cmd->then_call = op->on_consumed;
+    grpc_iomgr_closure_init(&cmd->closure, free_message, cmd);
+    op->on_consumed = &cmd->closure;
+    op->optional_close_message = &cmd->message;
+  }
+  op->close_with_status = status;
+}
diff --git a/src/core/transport/transport.h b/src/core/transport/transport.h
index 7efcfcf970..92c1f38c5e 100644
--- a/src/core/transport/transport.h
+++ b/src/core/transport/transport.h
@@ -80,8 +80,14 @@ typedef struct grpc_transport_stream_op {
 
   grpc_pollset *bind_pollset;
 
+  /** If != GRPC_STATUS_OK, cancel this stream */
   grpc_status_code cancel_with_status;
 
+  /** If != GRPC_STATUS_OK, send grpc-status, grpc-message, and close this
+      stream for both reading and writing */
+  grpc_status_code close_with_status;
+  gpr_slice *optional_close_message;
+
   /* Indexes correspond to grpc_context_index enum values */
   grpc_call_context_element *context;
 } grpc_transport_stream_op;
@@ -148,8 +154,11 @@ void grpc_transport_destroy_stream(grpc_transport *transport,
 void grpc_transport_stream_op_finish_with_failure(grpc_transport_stream_op *op);
 
 void grpc_transport_stream_op_add_cancellation(grpc_transport_stream_op *op,
-                                               grpc_status_code status,
-                                               grpc_mdstr *message);
+                                               grpc_status_code status);
+
+void grpc_transport_stream_op_add_close(grpc_transport_stream_op *op,
+                                        grpc_status_code status,
+                                        gpr_slice *optional_message);
 
 char *grpc_transport_stream_op_string(grpc_transport_stream_op *op);
 
diff --git a/src/core/tsi/transport_security_interface.h b/src/core/tsi/transport_security_interface.h
index 936b0c25b0..e27e6b9fc9 100644
--- a/src/core/tsi/transport_security_interface.h
+++ b/src/core/tsi/transport_security_interface.h
@@ -158,6 +158,8 @@ tsi_result tsi_frame_protector_protect_flush(
      value is expected to be at most max_protected_frame_size minus overhead
      which means that max_protected_frame_size is a safe bet. The output value
      is the number of bytes actually written.
+     If *unprotected_bytes_size is unchanged, there may be more data remaining
+     to unprotect, and the caller should call this function again.
 
    - This method returns TSI_OK in case of success. Success includes cases where
      there is not enough data to output a frame in which case