1 files changed, 36 insertions, 25 deletions

diff --git a/src/core/surface/call.c b/src/core/surface/call.c
index 79295ae0ff..b999ecd37c 100644
--- a/src/core/surface/call.c
+++ b/src/core/surface/call.c
@@ -336,26 +336,19 @@ void grpc_call_set_completion_queue(grpc_exec_ctx *exec_ctx, grpc_call *call,
                               grpc_cq_pollset(cq));
 }
 
-grpc_completion_queue *grpc_call_get_completion_queue(grpc_call *call) {
-  return call->cq;
-}
-
 #ifdef GRPC_STREAM_REFCOUNT_DEBUG
-void grpc_call_internal_ref(grpc_call *c, const char *reason) {
-  grpc_call_stack_ref(CALL_STACK_FROM_CALL(c), reason);
-}
-void grpc_call_internal_unref(grpc_exec_ctx *exec_ctx, grpc_call *c,
-                              const char *reason) {
-  grpc_call_stack_unref(exec_ctx, CALL_STACK_FROM_CALL(c), reason);
-}
+#define REF_REASON reason
+#define REF_ARG , const char *reason
 #else
-void grpc_call_internal_ref(grpc_call *c) {
-  grpc_call_stack_ref(CALL_STACK_FROM_CALL(c));
+#define REF_REASON ""
+#define REF_ARG
+#endif
+void grpc_call_internal_ref(grpc_call *c REF_ARG) {
+  GRPC_CALL_STACK_REF(CALL_STACK_FROM_CALL(c), REF_REASON);
 }
-void grpc_call_internal_unref(grpc_exec_ctx *exec_ctx, grpc_call *c) {
-  grpc_call_stack_unref(exec_ctx, CALL_STACK_FROM_CALL(c));
+void grpc_call_internal_unref(grpc_exec_ctx *exec_ctx, grpc_call *c REF_ARG) {
+  GRPC_CALL_STACK_UNREF(exec_ctx, CALL_STACK_FROM_CALL(c), REF_REASON);
 }
-#endif
 
 static void destroy_call(grpc_exec_ctx *exec_ctx, void *call, int success) {
   size_t i;
@@ -742,8 +735,15 @@ static void execute_op(grpc_exec_ctx *exec_ctx, grpc_call *call,
 char *grpc_call_get_peer(grpc_call *call) {
   grpc_call_element *elem = CALL_ELEM_FROM_CALL(call, 0);
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
-  char *result = elem->filter->get_peer(&exec_ctx, elem);
+  char *result;
   GRPC_API_TRACE("grpc_call_get_peer(%p)", 1, (call));
+  result = elem->filter->get_peer(&exec_ctx, elem);
+  if (result == NULL) {
+    result = grpc_channel_get_target(call->channel);
+  }
+  if (result == NULL) {
+    result = gpr_strdup("unknown");
+  }
   grpc_exec_ctx_finish(&exec_ctx);
   return result;
 }
@@ -974,11 +974,19 @@ static void receiving_slice_ready(grpc_exec_ctx *exec_ctx, void *bctlp,
   batch_control *bctl = bctlp;
   grpc_call *call = bctl->call;
 
-  GPR_ASSERT(success);
-  gpr_slice_buffer_add(&(*call->receiving_buffer)->data.raw.slice_buffer,
-                       call->receiving_slice);
-
-  continue_receiving_slices(exec_ctx, bctl);
+  if (success) {
+    gpr_slice_buffer_add(&(*call->receiving_buffer)->data.raw.slice_buffer,
+                         call->receiving_slice);
+    continue_receiving_slices(exec_ctx, bctl);
+  } else {
+    grpc_byte_stream_destroy(call->receiving_stream);
+    call->receiving_stream = NULL;
+    grpc_byte_buffer_destroy(*call->receiving_buffer);
+    *call->receiving_buffer = NULL;
+    if (gpr_unref(&bctl->steps_to_complete)) {
+      post_batch_completion(exec_ctx, bctl);
+    }
+  }
 }
 
 static void finish_batch(grpc_exec_ctx *exec_ctx, void *bctlp, int success) {
@@ -1060,6 +1068,7 @@ static void receiving_stream_ready(grpc_exec_ctx *exec_ctx, void *bctlp,
 
   if (call->receiving_stream == NULL) {
     *call->receiving_buffer = NULL;
+    call->receiving_message = 0; 
     if (gpr_unref(&bctl->steps_to_complete)) {
       post_batch_completion(exec_ctx, bctl);
     }
@@ -1070,6 +1079,7 @@ static void receiving_stream_ready(grpc_exec_ctx *exec_ctx, void *bctlp,
     grpc_byte_stream_destroy(exec_ctx, call->receiving_stream);
     call->receiving_stream = NULL;
     *call->receiving_buffer = NULL;
+    call->receiving_message = 0; 
     if (gpr_unref(&bctl->steps_to_complete)) {
       post_batch_completion(exec_ctx, bctl);
     }
@@ -1119,11 +1129,12 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx,
     GRPC_CALL_INTERNAL_REF(call, "completion");
     bctl->success = 1;
     if (!is_notify_tag_closure) {
-      grpc_cq_begin_op(call->cq);
+      grpc_cq_begin_op(call->cq, notify_tag);
     }
     gpr_mu_unlock(&call->mu);
     post_batch_completion(exec_ctx, bctl);
-    return GRPC_CALL_OK;
+    error = GRPC_CALL_OK;
+    goto done;
   }
 
   /* rewrite batch ops into a transport op */
@@ -1333,7 +1344,7 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx,
 
   GRPC_CALL_INTERNAL_REF(call, "completion");
   if (!is_notify_tag_closure) {
-    grpc_cq_begin_op(call->cq);
+    grpc_cq_begin_op(call->cq, notify_tag);
   }
   gpr_ref_init(&bctl->steps_to_complete, num_completion_callbacks_needed);