Merge branch 'master' into uv_core_transport

1 files changed, 44 insertions, 36 deletions

diff --git a/src/core/lib/surface/server.c b/src/core/lib/surface/server.c
index 64afcecc07..56fb80e92e 100644
--- a/src/core/lib/surface/server.c
+++ b/src/core/lib/surface/server.c
@@ -149,6 +149,7 @@ struct call_data {
 
   grpc_metadata_batch *recv_initial_metadata;
   bool recv_idempotent_request;
+  bool recv_cacheable_request;
   grpc_metadata_array initial_metadata;
 
   request_matcher *request_matcher;
@@ -272,23 +273,20 @@ static void shutdown_cleanup(grpc_exec_ctx *exec_ctx, void *arg,
 }
 
 static void send_shutdown(grpc_exec_ctx *exec_ctx, grpc_channel *channel,
-                          bool send_goaway, grpc_error *send_disconnect) {
-  grpc_transport_op op;
-  struct shutdown_cleanup_args *sc;
+                          int send_goaway, grpc_error *send_disconnect) {
+  struct shutdown_cleanup_args *sc = gpr_malloc(sizeof(*sc));
+  grpc_closure_init(&sc->closure, shutdown_cleanup, sc);
+  grpc_transport_op *op = grpc_make_transport_op(&sc->closure);
   grpc_channel_element *elem;
 
-  memset(&op, 0, sizeof(op));
-  op.send_goaway = send_goaway;
-  sc = gpr_malloc(sizeof(*sc));
+  op->send_goaway = send_goaway;
   sc->slice = gpr_slice_from_copied_string("Server shutdown");
-  op.goaway_message = &sc->slice;
-  op.goaway_status = GRPC_STATUS_OK;
-  op.disconnect_with_error = send_disconnect;
-  grpc_closure_init(&sc->closure, shutdown_cleanup, sc);
-  op.on_consumed = &sc->closure;
+  op->goaway_message = &sc->slice;
+  op->goaway_status = GRPC_STATUS_OK;
+  op->disconnect_with_error = send_disconnect;
 
   elem = grpc_channel_stack_element(grpc_channel_get_channel_stack(channel), 0);
-  elem->filter->start_transport_op(exec_ctx, elem, &op);
+  elem->filter->start_transport_op(exec_ctx, elem, op);
 }
 
 static void channel_broadcaster_shutdown(grpc_exec_ctx *exec_ctx,
@@ -431,7 +429,8 @@ static void finish_destroy_channel(grpc_exec_ctx *exec_ctx, void *cd,
   server_unref(exec_ctx, server);
 }
 
-static void destroy_channel(grpc_exec_ctx *exec_ctx, channel_data *chand) {
+static void destroy_channel(grpc_exec_ctx *exec_ctx, channel_data *chand,
+                            grpc_error *error) {
   if (is_channel_orphaned(chand)) return;
   GPR_ASSERT(chand->server != NULL);
   orphan_channel(chand);
@@ -440,14 +439,20 @@ static void destroy_channel(grpc_exec_ctx *exec_ctx, channel_data *chand) {
   chand->finish_destroy_channel_closure.cb = finish_destroy_channel;
   chand->finish_destroy_channel_closure.cb_arg = chand;
 
-  grpc_transport_op op;
-  memset(&op, 0, sizeof(op));
-  op.set_accept_stream = true;
-  op.on_consumed = &chand->finish_destroy_channel_closure;
+  grpc_transport_op *op =
+      grpc_make_transport_op(&chand->finish_destroy_channel_closure);
+  op->set_accept_stream = true;
   grpc_channel_next_op(exec_ctx,
                        grpc_channel_stack_element(
                            grpc_channel_get_channel_stack(chand->channel), 0),
-                       &op);
+                       op);
+
+  if (error != GRPC_ERROR_NONE) {
+    const char *msg = grpc_error_string(error);
+    gpr_log(GPR_INFO, "Disconnected client: %s", msg);
+    grpc_error_free_string(msg);
+  }
+  GRPC_ERROR_UNREF(error);
 }
 
 static void cpstr(char **dest, size_t *capacity, grpc_mdstr *value) {
@@ -497,9 +502,12 @@ static void publish_call(grpc_exec_ctx *exec_ctx, grpc_server *server,
             &rc->data.batch.details->method_capacity, calld->path);
       rc->data.batch.details->deadline = calld->deadline;
       rc->data.batch.details->flags =
-          0 | (calld->recv_idempotent_request
-                   ? GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST
-                   : 0);
+          (calld->recv_idempotent_request
+               ? GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST
+               : 0) |
+          (calld->recv_cacheable_request
+               ? GRPC_INITIAL_METADATA_CACHEABLE_REQUEST
+               : 0);
       break;
     case REGISTERED_CALL:
       *rc->data.registered.deadline = calld->deadline;
@@ -779,6 +787,7 @@ static void server_mutate_op(grpc_call_element *elem,
     calld->on_done_recv_initial_metadata = op->recv_initial_metadata_ready;
     op->recv_initial_metadata_ready = &calld->server_on_recv_initial_metadata;
     op->recv_idempotent_request = &calld->recv_idempotent_request;
+    op->recv_cacheable_request = &calld->recv_cacheable_request;
   }
 }
 
@@ -840,17 +849,16 @@ static void channel_connectivity_changed(grpc_exec_ctx *exec_ctx, void *cd,
   channel_data *chand = cd;
   grpc_server *server = chand->server;
   if (chand->connectivity_state != GRPC_CHANNEL_SHUTDOWN) {
-    grpc_transport_op op;
-    memset(&op, 0, sizeof(op));
-    op.on_connectivity_state_change = &chand->channel_connectivity_changed,
-    op.connectivity_state = &chand->connectivity_state;
+    grpc_transport_op *op = grpc_make_transport_op(NULL);
+    op->on_connectivity_state_change = &chand->channel_connectivity_changed,
+    op->connectivity_state = &chand->connectivity_state;
     grpc_channel_next_op(exec_ctx,
                          grpc_channel_stack_element(
                              grpc_channel_get_channel_stack(chand->channel), 0),
-                         &op);
+                         op);
   } else {
     gpr_mu_lock(&server->mu_global);
-    destroy_channel(exec_ctx, chand);
+    destroy_channel(exec_ctx, chand, GRPC_ERROR_REF(error));
     gpr_mu_unlock(&server->mu_global);
     GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, chand->channel, "connectivity");
   }
@@ -1114,7 +1122,7 @@ void grpc_server_setup_transport(grpc_exec_ctx *exec_ctx, grpc_server *s,
   size_t slots;
   uint32_t probes;
   uint32_t max_probes = 0;
-  grpc_transport_op op;
+  grpc_transport_op *op = NULL;
 
   channel =
       grpc_channel_create(exec_ctx, NULL, args, GRPC_SERVER_CHANNEL, transport);
@@ -1174,16 +1182,16 @@ void grpc_server_setup_transport(grpc_exec_ctx *exec_ctx, grpc_server *s,
   gpr_mu_unlock(&s->mu_global);
 
   GRPC_CHANNEL_INTERNAL_REF(channel, "connectivity");
-  memset(&op, 0, sizeof(op));
-  op.set_accept_stream = true;
-  op.set_accept_stream_fn = accept_stream;
-  op.set_accept_stream_user_data = chand;
-  op.on_connectivity_state_change = &chand->channel_connectivity_changed;
-  op.connectivity_state = &chand->connectivity_state;
+  op = grpc_make_transport_op(NULL);
+  op->set_accept_stream = true;
+  op->set_accept_stream_fn = accept_stream;
+  op->set_accept_stream_user_data = chand;
+  op->on_connectivity_state_change = &chand->channel_connectivity_changed;
+  op->connectivity_state = &chand->connectivity_state;
   if (gpr_atm_acq_load(&s->shutdown_flag) != 0) {
-    op.disconnect_with_error = GRPC_ERROR_CREATE("Server shutdown");
+    op->disconnect_with_error = GRPC_ERROR_CREATE("Server shutdown");
   }
-  grpc_transport_perform_op(exec_ctx, transport, &op);
+  grpc_transport_perform_op(exec_ctx, transport, op);
 }
 
 void done_published_shutdown(grpc_exec_ctx *exec_ctx, void *done_arg,