Make the things compile again

src/core/security/auth.c

@@ -51,7 +51,9 @@ typedef struct {
   grpc_credentials *creds;
   grpc_mdstr *host;
   grpc_mdstr *method;
-  grpc_call_op op;
+  grpc_transport_op op;
+  size_t op_md_idx;
+  int sent_initial_metadata;
   grpc_linked_mdelem md_links[MAX_CREDENTIALS_METADATA_COUNT];
 } call_data;
 
@@ -75,14 +77,17 @@ static void on_credentials_metadata(void *user_data, grpc_mdelem **md_elems,
                                     grpc_credentials_status status) {
   grpc_call_element *elem = (grpc_call_element *)user_data;
   call_data *calld = elem->call_data;
-  grpc_call_op op = calld->op;
+  grpc_transport_op *op = &calld->op;
+  grpc_metadata_batch *mdb;
   size_t i;
   GPR_ASSERT(num_md <= MAX_CREDENTIALS_METADATA_COUNT);
+  GPR_ASSERT(op->send_ops && op->send_ops->nops > calld->op_md_idx && op->send_ops->ops[calld->op_md_idx].type == GRPC_OP_METADATA);
+  mdb = &op->send_ops->ops[calld->op_md_idx].data.metadata;
   for (i = 0; i < num_md; i++) {
-    grpc_metadata_batch_add_tail(&op.data.metadata, &calld->md_links[i],
+    grpc_metadata_batch_add_tail(mdb, &calld->md_links[i],
                                  grpc_mdelem_ref(md_elems[i]));
   }
-  grpc_call_next_op(elem, &op);
+  grpc_call_next_op(elem, op);
 }
 
 static char *build_service_url(const char *url_scheme, call_data *calld) {
@@ -105,7 +110,7 @@ static char *build_service_url(const char *url_scheme, call_data *calld) {
   return service_url;
 }
 
-static void send_security_metadata(grpc_call_element *elem, grpc_call_op *op) {
+static void send_security_metadata(grpc_call_element *elem, grpc_transport_op *op) {
   /* grab pointers to our data from the call element */
   call_data *calld = elem->call_data;
   channel_data *channeld = elem->channel_data;
@@ -144,9 +149,8 @@ static void on_host_checked(void *user_data, grpc_security_status status) {
     gpr_asprintf(&error_msg, "Invalid host %s set in :authority metadata.",
                  grpc_mdstr_as_c_string(calld->host));
     bubbleup_error(elem, error_msg);
-    grpc_metadata_batch_destroy(&calld->op.data.metadata);
     gpr_free(error_msg);
-    calld->op.done_cb(calld->op.user_data, GRPC_OP_ERROR);
+    grpc_transport_op_finish_with_failure(&calld->op);
   }
 }
 
@@ -155,16 +159,22 @@ static void on_host_checked(void *user_data, grpc_security_status status) {
      - a network event (or similar) from below, to receive something
    op contains type and call direction information, in addition to the data
    that is being sent or received. */
-static void call_op(grpc_call_element *elem, grpc_call_element *from_elem,
-                    grpc_call_op *op) {
+static void auth_start_transport_op(grpc_call_element *elem, 
+                    grpc_transport_op *op) {
   /* grab pointers to our data from the call element */
   call_data *calld = elem->call_data;
   channel_data *channeld = elem->channel_data;
   grpc_linked_mdelem *l;
+  size_t i;
 
-  switch (op->type) {
-    case GRPC_SEND_METADATA:
-      for (l = op->data.metadata.list.head; l != NULL; l = l->next) {
+  if (op->send_ops && !calld->sent_initial_metadata) {
+    size_t nops = op->send_ops->nops;
+    grpc_stream_op *ops = op->send_ops->ops;
+    for (i = 0; i < nops; i++) {
+      grpc_stream_op *sop = &ops[i];
+      if (sop->type != GRPC_OP_METADATA) continue;
+      calld->sent_initial_metadata = 1;
+      for (l = sop->data.metadata.list.head; l != NULL; l = l->next) {
         grpc_mdelem *md = l->md;
         /* Pointer comparison is OK for md_elems created from the same context.
          */
@@ -189,20 +199,19 @@ static void call_op(grpc_call_element *elem, grpc_call_element *from_elem,
                          "Invalid host %s set in :authority metadata.",
                          call_host);
             bubbleup_error(elem, error_msg);
-            grpc_metadata_batch_destroy(&calld->op.data.metadata);
             gpr_free(error_msg);
-            op->done_cb(op->user_data, GRPC_OP_ERROR);
+            grpc_transport_op_finish_with_failure(&calld->op);
           }
-          break;
+          return; /* early exit */
         }
       }
       send_security_metadata(elem, op);
-      break;
-    default:
-      /* pass control up or down the stack depending on op->dir */
-      grpc_call_next_op(elem, op);
-      break;
+      return; /* early exit */
+    }
   }
+
+  /* pass control up or down the stack */
+  grpc_call_next_op(elem, op);
 }
 
 /* Called on special channel events, such as disconnection or new incoming
@@ -214,13 +223,15 @@ static void channel_op(grpc_channel_element *elem,
 
 /* Constructor for call_data */
 static void init_call_elem(grpc_call_element *elem,
-                           const void *server_transport_data) {
+                           const void *server_transport_data, grpc_transport_op *initial_op) {
   /* TODO(jboeuf):
      Find a way to pass-in the credentials from the caller here.  */
   call_data *calld = elem->call_data;
   calld->creds = NULL;
   calld->host = NULL;
   calld->method = NULL;
+
+  GPR_ASSERT(!initial_op || !initial_op->send_ops);
 }
 
 /* Destructor for call_data */
@@ -288,5 +299,5 @@ static void destroy_channel_elem(grpc_channel_element *elem) {
 }
 
 const grpc_channel_filter grpc_client_auth_filter = {
-    call_op, channel_op, sizeof(call_data), init_call_elem, destroy_call_elem,
+    auth_start_transport_op, channel_op, sizeof(call_data), init_call_elem, destroy_call_elem,
     sizeof(channel_data), init_channel_elem, destroy_channel_elem, "auth"};

src/core/security/server_secure_chttp2.c

@@ -35,7 +35,6 @@
 
 #include <string.h>
 
-#include "src/core/channel/http_filter.h"
 #include "src/core/channel/http_server_filter.h"
 #include "src/core/iomgr/endpoint.h"
 #include "src/core/iomgr/resolve_address.h"
@@ -73,8 +72,7 @@ static void state_unref(grpc_server_secure_state *state) {
 static grpc_transport_setup_result setup_transport(void *server,
                                                    grpc_transport *transport,
                                                    grpc_mdctx *mdctx) {
-  static grpc_channel_filter const *extra_filters[] = {&grpc_http_server_filter,
-                                                       &grpc_http_filter};
+  static grpc_channel_filter const *extra_filters[] = {&grpc_http_server_filter};
   return grpc_server_setup_transport(server, transport, extra_filters,
                                      GPR_ARRAY_SIZE(extra_filters), mdctx);
 }

src/core/surface/call.c

@@ -598,6 +598,8 @@ static void call_on_done_send(void *pc, int success) {
     finish_ioreq_op(call, GRPC_IOREQ_SEND_MESSAGE, error);
   }
   if (call->last_send_contains & (1 << GRPC_IOREQ_SEND_CLOSE)) {
+    finish_ioreq_op(call, GRPC_IOREQ_SEND_TRAILING_METADATA, error);
+    finish_ioreq_op(call, GRPC_IOREQ_SEND_STATUS, error);
     finish_ioreq_op(call, GRPC_IOREQ_SEND_CLOSE, error);
   }
   call->sending = 0;
@@ -684,6 +686,7 @@ static void call_on_done_recv(void *pc, int success) {
   size_t i;
   int unref = 0;
   lock(call);
+  call->receiving = 0;
   for (i = 0; success && i < call->recv_ops.nops; i++) {
     grpc_stream_op *op = &call->recv_ops.ops[i];
     switch (op->type) {

src/core/surface/secure_channel_create.c

@@ -44,7 +44,6 @@
 #include "src/core/channel/client_setup.h"
 #include "src/core/channel/connected_channel.h"
 #include "src/core/channel/http_client_filter.h"
-#include "src/core/channel/http_filter.h"
 #include "src/core/iomgr/resolve_address.h"
 #include "src/core/iomgr/tcp_client.h"
 #include "src/core/security/auth.h"
@@ -193,7 +192,7 @@ static grpc_transport_setup_result complete_setup(void *channel_stack,
                                                   grpc_transport *transport,
                                                   grpc_mdctx *mdctx) {
   static grpc_channel_filter const *extra_filters[] = {
-      &grpc_client_auth_filter, &grpc_http_client_filter, &grpc_http_filter};
+      &grpc_client_auth_filter, &grpc_http_client_filter};
   return grpc_client_channel_transport_setup_complete(
       channel_stack, transport, extra_filters, GPR_ARRAY_SIZE(extra_filters),
       mdctx);

src/core/transport/transport.c

@@ -85,3 +85,12 @@ void grpc_transport_setup_cancel(grpc_transport_setup *setup) {
 void grpc_transport_setup_initiate(grpc_transport_setup *setup) {
   setup->vtable->initiate(setup);
 }
+
+void grpc_transport_op_finish_with_failure(grpc_transport_op *op) {
+  if (op->send_ops) {
+    op->on_done_send(op->send_user_data, 0);
+  }
+  if (op->recv_ops) {
+    op->on_done_recv(op->recv_user_data, 0);
+  }
+}

src/core/transport/transport_op_string.c

@@ -42,7 +42,7 @@
 #include <grpc/support/useful.h>
 
 static void put_metadata(gpr_strvec *b, grpc_mdelem *md) {
-  gpr_strvec_add(b, gpr_strdup(" key="));
+  gpr_strvec_add(b, gpr_strdup("key="));
   gpr_strvec_add(
       b, gpr_hexdump((char *)GPR_SLICE_START_PTR(md->key->slice),
                      GPR_SLICE_LENGTH(md->key->slice), GPR_HEXDUMP_PLAINTEXT));
@@ -56,6 +56,7 @@ static void put_metadata(gpr_strvec *b, grpc_mdelem *md) {
 static void put_metadata_list(gpr_strvec *b, grpc_metadata_batch md) {
   grpc_linked_mdelem *m;
   for (m = md.list.head; m != NULL; m = m->next) {
+    if (m != md.list.head) gpr_strvec_add(b, gpr_strdup(", "));
     put_metadata(b, m->md);
   }
   if (gpr_time_cmp(md.deadline, gpr_inf_future) != 0) {
@@ -75,7 +76,7 @@ char *grpc_sopb_string(grpc_stream_op_buffer *sopb) {
 
   for (i = 0; i < sopb->nops; i++) {
     grpc_stream_op *op = &sopb->ops[i];
-    if (i) gpr_strvec_add(&b, gpr_strdup(", "));
+    if (i > 0) gpr_strvec_add(&b, gpr_strdup(", "));
     switch (op->type) {
       case GRPC_NO_OP:
         gpr_strvec_add(&b, gpr_strdup("NO_OP"));
@@ -88,7 +89,9 @@ char *grpc_sopb_string(grpc_stream_op_buffer *sopb) {
         gpr_asprintf(&tmp, "SLICE:%d", GPR_SLICE_LENGTH(op->data.slice));
         break;
       case GRPC_OP_METADATA:
+        gpr_strvec_add(&b, gpr_strdup("METADATA{"));
         put_metadata_list(&b, op->data.metadata);
+        gpr_strvec_add(&b, gpr_strdup("}"));
         break;
     }
   }

test/core/channel/channel_stack_test.c

@@ -55,7 +55,7 @@ static void channel_init_func(grpc_channel_element *elem,
 }
 
 static void call_init_func(grpc_call_element *elem,
-                           const void *server_transport_data) {
+                           const void *server_transport_data, grpc_transport_op *initial_op) {
   ++*(int *)(elem->channel_data);
   *(int *)(elem->call_data) = 0;
 }
@@ -66,8 +66,8 @@ static void call_destroy_func(grpc_call_element *elem) {
   ++*(int *)(elem->channel_data);
 }
 
-static void call_func(grpc_call_element *elem, grpc_call_element *from_elem,
-                      grpc_call_op *op) {
+static void call_func(grpc_call_element *elem, 
+                      grpc_transport_op *op) {
   ++*(int *)(elem->call_data);
 }
 
@@ -112,7 +112,7 @@ static void test_create_channel_stack(void) {
   GPR_ASSERT(*channel_data == 0);
 
   call_stack = gpr_malloc(channel_stack->call_stack_size);
-  grpc_call_stack_init(channel_stack, NULL, call_stack);
+  grpc_call_stack_init(channel_stack, NULL, NULL, call_stack);
   GPR_ASSERT(call_stack->count == 1);
   call_elem = grpc_call_stack_element(call_stack, 0);
   GPR_ASSERT(call_elem->filter == channel_elem->filter);

test/core/end2end/fixtures/chttp2_fullstack.c

@@ -37,7 +37,6 @@
 
 #include "src/core/channel/client_channel.h"
 #include "src/core/channel/connected_channel.h"
-#include "src/core/channel/http_filter.h"
 #include "src/core/channel/http_server_filter.h"
 #include "src/core/surface/channel.h"
 #include "src/core/surface/client.h"

test/core/end2end/fixtures/chttp2_fullstack_uds.c

@@ -39,7 +39,6 @@
 
 #include "src/core/channel/client_channel.h"
 #include "src/core/channel/connected_channel.h"
-#include "src/core/channel/http_filter.h"
 #include "src/core/channel/http_server_filter.h"
 #include "src/core/support/string.h"
 #include "src/core/surface/channel.h"

test/core/end2end/fixtures/chttp2_socket_pair_one_byte_at_a_time.c

@@ -37,7 +37,6 @@
 
 #include "src/core/channel/client_channel.h"
 #include "src/core/channel/connected_channel.h"
-#include "src/core/channel/http_filter.h"
 #include "src/core/channel/http_client_filter.h"
 #include "src/core/channel/http_server_filter.h"
 #include "src/core/iomgr/endpoint_pair.h"
@@ -60,8 +59,7 @@
 static grpc_transport_setup_result server_setup_transport(
     void *ts, grpc_transport *transport, grpc_mdctx *mdctx) {
   grpc_end2end_test_fixture *f = ts;
-  static grpc_channel_filter const *extra_filters[] = {&grpc_http_server_filter,
-                                                       &grpc_http_filter};
+  static grpc_channel_filter const *extra_filters[] = {&grpc_http_server_filter};
   return grpc_server_setup_transport(f->server, transport, extra_filters,
                                      GPR_ARRAY_SIZE(extra_filters), mdctx);
 }
@@ -76,7 +74,7 @@ static grpc_transport_setup_result client_setup_transport(
   sp_client_setup *cs = ts;
 
   const grpc_channel_filter *filters[] = {
-      &grpc_client_surface_filter, &grpc_http_client_filter, &grpc_http_filter,
+      &grpc_client_surface_filter, &grpc_http_client_filter, 
       &grpc_connected_channel_filter};
   size_t nfilters = sizeof(filters) / sizeof(*filters);
   grpc_channel *channel = grpc_channel_create_from_filters(

test/core/transport/chttp2/stream_encoder_test.c

@@ -102,15 +102,10 @@ static void verify_sopb(size_t window_available, int eof,
   gpr_slice_unref(expect);
 }
 
-static void assert_result_ok(void *user_data, grpc_op_error error) {
-  GPR_ASSERT(error == GRPC_OP_OK);
-}
-
 static void test_small_data_framing(void) {
   grpc_sopb_add_no_op(&g_sopb);
   verify_sopb(10, 0, 0, "");
 
-  grpc_sopb_add_flow_ctl_cb(&g_sopb, assert_result_ok, NULL);
   grpc_sopb_add_slice(&g_sopb, create_test_slice(3));
   verify_sopb(10, 0, 3, "000003 0000 deadbeef 000102");
 

test/core/transport/stream_op_test.c

@@ -38,10 +38,6 @@
 #include <grpc/support/log.h>
 #include "test/core/util/test_config.h"
 
-static void flow_ctl_cb_fails(void *ignored, grpc_op_error error) {
-  GPR_ASSERT(error == GRPC_OP_ERROR);
-}
-
 static void assert_slices_equal(gpr_slice a, gpr_slice b) {
   GPR_ASSERT(a.refcount == b.refcount);
   if (a.refcount) {
@@ -60,7 +56,6 @@ int main(int argc, char **argv) {
   gpr_slice test_slice_2 = gpr_slice_malloc(2);
   gpr_slice test_slice_3 = gpr_slice_malloc(3);
   gpr_slice test_slice_4 = gpr_slice_malloc(4);
-  char x;
   unsigned i;
 
   grpc_stream_op_buffer buf;
@@ -78,7 +73,6 @@ int main(int argc, char **argv) {
   grpc_sopb_add_slice(&buf, test_slice_2);
   grpc_sopb_add_slice(&buf, test_slice_3);
   grpc_sopb_add_slice(&buf, test_slice_4);
-  grpc_sopb_add_flow_ctl_cb(&buf, flow_ctl_cb_fails, &x);
   grpc_sopb_add_no_op(&buf);
 
   /* verify that the data went in ok */
@@ -94,10 +88,7 @@ int main(int argc, char **argv) {
   assert_slices_equal(buf.ops[3].data.slice, test_slice_3);
   GPR_ASSERT(buf.ops[4].type == GRPC_OP_SLICE);
   assert_slices_equal(buf.ops[4].data.slice, test_slice_4);
-  GPR_ASSERT(buf.ops[5].type == GRPC_OP_FLOW_CTL_CB);
-  GPR_ASSERT(buf.ops[5].data.flow_ctl_cb.cb == flow_ctl_cb_fails);
-  GPR_ASSERT(buf.ops[5].data.flow_ctl_cb.arg == &x);
-  GPR_ASSERT(buf.ops[6].type == GRPC_NO_OP);
+  GPR_ASSERT(buf.ops[5].type == GRPC_NO_OP);
 
   /* initialize the second buffer */
   grpc_sopb_init(&buf2);