Implement more missing pieces

5 files changed, 23 insertions, 5 deletions

diff --git a/src/core/transport/chttp2/internal.h b/src/core/transport/chttp2/internal.h
index 74385e8ad9..b60811d71e 100644
--- a/src/core/transport/chttp2/internal.h
+++ b/src/core/transport/chttp2/internal.h
@@ -430,7 +430,8 @@ typedef struct {
   /** address to publish next stream state to */
   grpc_stream_state *publish_state;
   /** pointer to sop buffer to fill in with new stream ops */
-  grpc_stream_op_buffer *incoming_sopb;
+  grpc_stream_op_buffer *publish_sopb;
+  grpc_stream_op_buffer incoming_sopb;
 
   /** incoming metadata */
   grpc_chttp2_incoming_metadata_buffer incoming_metadata;
diff --git a/src/core/transport/chttp2/parsing.c b/src/core/transport/chttp2/parsing.c
index 4c681ec9d3..91b509dc9a 100644
--- a/src/core/transport/chttp2/parsing.c
+++ b/src/core/transport/chttp2/parsing.c
@@ -183,6 +183,8 @@ void grpc_chttp2_publish_reads(
       grpc_chttp2_list_add_read_write_state_changed(transport_global,
                                                     stream_global);
     }
+
+    /* publish incoming stream ops */
   }
 }
 
diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c
index 14e408a53f..6bb9c72035 100644
--- a/src/core/transport/chttp2_transport.c
+++ b/src/core/transport/chttp2_transport.c
@@ -384,7 +384,7 @@ static void destroy_stream(grpc_transport *gt, grpc_stream *gs) {
   gpr_mu_unlock(&t->mu);
 
   GPR_ASSERT(s->global.outgoing_sopb == NULL);
-  GPR_ASSERT(s->global.incoming_sopb == NULL);
+  GPR_ASSERT(s->global.publish_sopb == NULL);
   grpc_sopb_destroy(&s->writing.sopb);
   grpc_chttp2_data_parser_destroy(&s->parsing.data_parser);
   grpc_chttp2_incoming_metadata_buffer_destroy(&s->parsing.incoming_metadata);
@@ -604,11 +604,11 @@ static void perform_op_locked(grpc_chttp2_transport_global *transport_global,
   }
 
   if (op->recv_ops) {
-    GPR_ASSERT(stream_global->incoming_sopb == NULL);
+    GPR_ASSERT(stream_global->publish_sopb == NULL);
     GPR_ASSERT(stream_global->published_state != GRPC_STREAM_CLOSED);
     stream_global->recv_done_closure = op->on_done_recv;
-    stream_global->incoming_sopb = op->recv_ops;
-    stream_global->incoming_sopb->nops = 0;
+    stream_global->publish_sopb = op->recv_ops;
+    stream_global->publish_sopb->nops = 0;
     grpc_chttp2_incoming_metadata_live_op_buffer_end(
         &stream_global->outstanding_metadata);
     grpc_chttp2_list_add_read_write_state_changed(transport_global,
diff --git a/src/core/transport/stream_op.c b/src/core/transport/stream_op.c
index 8996ecac35..8b5549cd4c 100644
--- a/src/core/transport/stream_op.c
+++ b/src/core/transport/stream_op.c
@@ -163,6 +163,19 @@ void grpc_sopb_append(grpc_stream_op_buffer *sopb, grpc_stream_op *ops,
   sopb->nops = new_nops;
 }
 
+void grpc_sopb_move_to(grpc_stream_op_buffer *src, grpc_stream_op_buffer *dst) {
+  size_t i;
+  if (src->nops == 0) {
+    return;
+  }
+  if (dst->nops == 0) {
+    grpc_sopb_swap(src, dst);
+    return;
+  }
+  grpc_sopb_append(dst, src->ops, src->nops);
+  src->ops = 0;
+}
+
 static void assert_valid_list(grpc_mdelem_list *list) {
 #ifndef NDEBUG
   grpc_linked_mdelem *l;
diff --git a/src/core/transport/stream_op.h b/src/core/transport/stream_op.h
index 5215cc87b1..644be567f3 100644
--- a/src/core/transport/stream_op.h
+++ b/src/core/transport/stream_op.h
@@ -159,6 +159,8 @@ void grpc_sopb_add_slice(grpc_stream_op_buffer *sopb, gpr_slice slice);
 void grpc_sopb_append(grpc_stream_op_buffer *sopb, grpc_stream_op *ops,
                       size_t nops);
 
+void grpc_sopb_move_to(grpc_stream_op_buffer *src, grpc_stream_op_buffer *dst);
+
 char *grpc_sopb_string(grpc_stream_op_buffer *sopb);
 
 #endif /* GRPC_INTERNAL_CORE_TRANSPORT_STREAM_OP_H */