stream_op cleanup: channel, client_config changes

1 files changed, 20 insertions, 9 deletions

diff --git a/src/core/channel/connected_channel.c b/src/core/channel/connected_channel.c
index 6d4d7be632..0e1efd965a 100644
--- a/src/core/channel/connected_channel.c
+++ b/src/core/channel/connected_channel.c
@@ -83,8 +83,7 @@ static void con_start_transport_op(grpc_exec_ctx *exec_ctx,
 
 /* Constructor for call_data */
 static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
-                           const void *server_transport_data,
-                           grpc_transport_stream_op *initial_op) {
+                           grpc_call_element_args *args) {
   call_data *calld = elem->call_data;
   channel_data *chand = elem->channel_data;
   int r;
@@ -92,10 +91,18 @@ static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
   GPR_ASSERT(elem->filter == &grpc_connected_channel_filter);
   r = grpc_transport_init_stream(exec_ctx, chand->transport,
                                  TRANSPORT_STREAM_FROM_CALL_DATA(calld),
-                                 server_transport_data, initial_op);
+                                 args->refcount, args->server_transport_data);
   GPR_ASSERT(r == 0);
 }
 
+static void set_pollset(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
+                        grpc_pollset *pollset) {
+  call_data *calld = elem->call_data;
+  channel_data *chand = elem->channel_data;
+  grpc_transport_set_pollset(exec_ctx, chand->transport,
+                             TRANSPORT_STREAM_FROM_CALL_DATA(calld), pollset);
+}
+
 /* Destructor for call_data */
 static void destroy_call_elem(grpc_exec_ctx *exec_ctx,
                               grpc_call_element *elem) {
@@ -108,11 +115,10 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx,
 
 /* Constructor for channel_data */
 static void init_channel_elem(grpc_exec_ctx *exec_ctx,
-                              grpc_channel_element *elem, grpc_channel *master,
-                              const grpc_channel_args *args, grpc_mdctx *mdctx,
-                              int is_first, int is_last) {
+                              grpc_channel_element *elem,
+                              grpc_channel_element_args *args) {
   channel_data *cd = (channel_data *)elem->channel_data;
-  GPR_ASSERT(is_last);
+  GPR_ASSERT(args->is_last);
   GPR_ASSERT(elem->filter == &grpc_connected_channel_filter);
   cd->transport = NULL;
 }
@@ -132,8 +138,8 @@ static char *con_get_peer(grpc_exec_ctx *exec_ctx, grpc_call_element *elem) {
 
 const grpc_channel_filter grpc_connected_channel_filter = {
     con_start_transport_stream_op, con_start_transport_op, sizeof(call_data),
-    init_call_elem, destroy_call_elem, sizeof(channel_data), init_channel_elem,
-    destroy_channel_elem, con_get_peer, "connected",
+    init_call_elem, set_pollset, destroy_call_elem, sizeof(channel_data),
+    init_channel_elem, destroy_channel_elem, con_get_peer, "connected",
 };
 
 void grpc_connected_channel_bind_transport(grpc_channel_stack *channel_stack,
@@ -154,3 +160,8 @@ void grpc_connected_channel_bind_transport(grpc_channel_stack *channel_stack,
      channel. */
   channel_stack->call_stack_size += grpc_transport_stream_size(transport);
 }
+
+grpc_stream *grpc_connected_channel_get_stream(grpc_call_element *elem) {
+  call_data *calld = elem->call_data;
+  return TRANSPORT_STREAM_FROM_CALL_DATA(calld);
+}