1 files changed, 84 insertions, 15 deletions

diff --git a/src/core/ext/filters/http/server/http_server_filter.cc b/src/core/ext/filters/http/server/http_server_filter.cc
index 3919447f26..ce1be8370c 100644
--- a/src/core/ext/filters/http/server/http_server_filter.cc
+++ b/src/core/ext/filters/http/server/http_server_filter.cc
@@ -23,6 +23,7 @@
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
 #include <string.h>
+#include "src/core/lib/channel/channel_args.h"
 #include "src/core/lib/gprpp/manual_constructor.h"
 #include "src/core/lib/profiling/timers.h"
 #include "src/core/lib/slice/b64.h"
@@ -34,9 +35,32 @@
 #define EXPECTED_CONTENT_TYPE "application/grpc"
 #define EXPECTED_CONTENT_TYPE_LENGTH sizeof(EXPECTED_CONTENT_TYPE) - 1
 
+static void hs_recv_initial_metadata_ready(void* user_data, grpc_error* err);
+static void hs_recv_trailing_metadata_ready(void* user_data, grpc_error* err);
+static void hs_recv_message_ready(void* user_data, grpc_error* err);
+
 namespace {
 
 struct call_data {
+  call_data(grpc_call_element* elem, const grpc_call_element_args& args)
+      : call_combiner(args.call_combiner) {
+    GRPC_CLOSURE_INIT(&recv_initial_metadata_ready,
+                      hs_recv_initial_metadata_ready, elem,
+                      grpc_schedule_on_exec_ctx);
+    GRPC_CLOSURE_INIT(&recv_message_ready, hs_recv_message_ready, elem,
+                      grpc_schedule_on_exec_ctx);
+    GRPC_CLOSURE_INIT(&recv_trailing_metadata_ready,
+                      hs_recv_trailing_metadata_ready, elem,
+                      grpc_schedule_on_exec_ctx);
+  }
+
+  ~call_data() {
+    GRPC_ERROR_UNREF(recv_initial_metadata_ready_error);
+    if (have_read_stream) {
+      read_stream->Orphan();
+    }
+  }
+
   grpc_call_combiner* call_combiner;
 
   // Outgoing headers to add to send_initial_metadata.
@@ -46,20 +70,31 @@ struct call_data {
   // If we see the recv_message contents in the GET query string, we
   // store it here.
   grpc_core::ManualConstructor<grpc_core::SliceBufferByteStream> read_stream;
-  bool have_read_stream;
+  bool have_read_stream = false;
 
   // State for intercepting recv_initial_metadata.
   grpc_closure recv_initial_metadata_ready;
+  grpc_error* recv_initial_metadata_ready_error = GRPC_ERROR_NONE;
   grpc_closure* original_recv_initial_metadata_ready;
-  grpc_metadata_batch* recv_initial_metadata;
+  grpc_metadata_batch* recv_initial_metadata = nullptr;
   uint32_t* recv_initial_metadata_flags;
-  bool seen_recv_initial_metadata_ready;
+  bool seen_recv_initial_metadata_ready = false;
 
   // State for intercepting recv_message.
   grpc_closure* original_recv_message_ready;
   grpc_closure recv_message_ready;
   grpc_core::OrphanablePtr<grpc_core::ByteStream>* recv_message;
-  bool seen_recv_message_ready;
+  bool seen_recv_message_ready = false;
+
+  // State for intercepting recv_trailing_metadata
+  grpc_closure recv_trailing_metadata_ready;
+  grpc_closure* original_recv_trailing_metadata_ready;
+  grpc_error* recv_trailing_metadata_ready_error;
+  bool seen_recv_trailing_metadata_ready = false;
+};
+
+struct channel_data {
+  bool surface_user_agent;
 };
 
 }  // namespace
@@ -258,6 +293,11 @@ static grpc_error* hs_filter_incoming_metadata(grpc_call_element* elem,
             GRPC_ERROR_STR_KEY, grpc_slice_from_static_string(":authority")));
   }
 
+  channel_data* chand = static_cast<channel_data*>(elem->channel_data);
+  if (!chand->surface_user_agent && b->idx.named.user_agent != nullptr) {
+    grpc_metadata_batch_remove(b, b->idx.named.user_agent);
+  }
+
   return error;
 }
 
@@ -267,6 +307,7 @@ static void hs_recv_initial_metadata_ready(void* user_data, grpc_error* err) {
   calld->seen_recv_initial_metadata_ready = true;
   if (err == GRPC_ERROR_NONE) {
     err = hs_filter_incoming_metadata(elem, calld->recv_initial_metadata);
+    calld->recv_initial_metadata_ready_error = GRPC_ERROR_REF(err);
     if (calld->seen_recv_message_ready) {
       // We've already seen the recv_message callback, but we previously
       // deferred it, so we need to return it here.
@@ -286,6 +327,13 @@ static void hs_recv_initial_metadata_ready(void* user_data, grpc_error* err) {
   } else {
     GRPC_ERROR_REF(err);
   }
+  if (calld->seen_recv_trailing_metadata_ready) {
+    GRPC_CALL_COMBINER_START(calld->call_combiner,
+                             &calld->recv_trailing_metadata_ready,
+                             calld->recv_trailing_metadata_ready_error,
+                             "resuming hs_recv_trailing_metadata_ready from "
+                             "hs_recv_initial_metadata_ready");
+  }
   GRPC_CLOSURE_RUN(calld->original_recv_initial_metadata_ready, err);
 }
 
@@ -313,6 +361,23 @@ static void hs_recv_message_ready(void* user_data, grpc_error* err) {
   }
 }
 
+static void hs_recv_trailing_metadata_ready(void* user_data, grpc_error* err) {
+  grpc_call_element* elem = static_cast<grpc_call_element*>(user_data);
+  call_data* calld = static_cast<call_data*>(elem->call_data);
+  if (!calld->seen_recv_initial_metadata_ready) {
+    calld->recv_trailing_metadata_ready_error = GRPC_ERROR_REF(err);
+    calld->seen_recv_trailing_metadata_ready = true;
+    GRPC_CALL_COMBINER_STOP(calld->call_combiner,
+                            "deferring hs_recv_trailing_metadata_ready until "
+                            "ater hs_recv_initial_metadata_ready");
+    return;
+  }
+  err = grpc_error_add_child(
+      GRPC_ERROR_REF(err),
+      GRPC_ERROR_REF(calld->recv_initial_metadata_ready_error));
+  GRPC_CLOSURE_RUN(calld->original_recv_trailing_metadata_ready, err);
+}
+
 static grpc_error* hs_mutate_op(grpc_call_element* elem,
                                 grpc_transport_stream_op_batch* op) {
   /* grab pointers to our data from the call element */
@@ -357,6 +422,13 @@ static grpc_error* hs_mutate_op(grpc_call_element* elem,
     op->payload->recv_message.recv_message_ready = &calld->recv_message_ready;
   }
 
+  if (op->recv_trailing_metadata) {
+    calld->original_recv_trailing_metadata_ready =
+        op->payload->recv_trailing_metadata.recv_trailing_metadata_ready;
+    op->payload->recv_trailing_metadata.recv_trailing_metadata_ready =
+        &calld->recv_trailing_metadata_ready;
+  }
+
   if (op->send_trailing_metadata) {
     grpc_error* error = hs_filter_outgoing_metadata(
         elem, op->payload->send_trailing_metadata.send_trailing_metadata);
@@ -382,13 +454,7 @@ static void hs_start_transport_stream_op_batch(
 /* Constructor for call_data */
 static grpc_error* hs_init_call_elem(grpc_call_element* elem,
                                      const grpc_call_element_args* args) {
-  call_data* calld = static_cast<call_data*>(elem->call_data);
-  calld->call_combiner = args->call_combiner;
-  GRPC_CLOSURE_INIT(&calld->recv_initial_metadata_ready,
-                    hs_recv_initial_metadata_ready, elem,
-                    grpc_schedule_on_exec_ctx);
-  GRPC_CLOSURE_INIT(&calld->recv_message_ready, hs_recv_message_ready, elem,
-                    grpc_schedule_on_exec_ctx);
+  new (elem->call_data) call_data(elem, *args);
   return GRPC_ERROR_NONE;
 }
 
@@ -397,15 +463,18 @@ static void hs_destroy_call_elem(grpc_call_element* elem,
                                  const grpc_call_final_info* final_info,
                                  grpc_closure* ignored) {
   call_data* calld = static_cast<call_data*>(elem->call_data);
-  if (calld->have_read_stream) {
-    calld->read_stream->Orphan();
-  }
+  calld->~call_data();
 }
 
 /* Constructor for channel_data */
 static grpc_error* hs_init_channel_elem(grpc_channel_element* elem,
                                         grpc_channel_element_args* args) {
+  channel_data* chand = static_cast<channel_data*>(elem->channel_data);
   GPR_ASSERT(!args->is_last);
+  chand->surface_user_agent = grpc_channel_arg_get_bool(
+      grpc_channel_args_find(args->channel_args,
+                             const_cast<char*>(GRPC_ARG_SURFACE_USER_AGENT)),
+      true);
   return GRPC_ERROR_NONE;
 }
 
@@ -419,7 +488,7 @@ const grpc_channel_filter grpc_http_server_filter = {
     hs_init_call_elem,
     grpc_call_stack_ignore_set_pollset_or_pollset_set,
     hs_destroy_call_elem,
-    0,
+    sizeof(channel_data),
     hs_init_channel_elem,
     hs_destroy_channel_elem,
     grpc_channel_next_get_info,