Update C++ code to set status via the C api.

This prevents mismatches from breaking tests.

2 files changed, 15 insertions, 27 deletions

diff --git a/src/cpp/stream/stream_context.cc b/src/cpp/stream/stream_context.cc
index 7936a30dfd..d46ba6e56a 100644
--- a/src/cpp/stream/stream_context.cc
+++ b/src/cpp/stream/stream_context.cc
@@ -112,9 +112,7 @@ bool StreamContext::Read(google::protobuf::Message* msg) {
   if (read_ev->data.read) {
     if (!DeserializeProto(read_ev->data.read, msg)) {
       ret = false;
-      FinishStream(
-          Status(StatusCode::DATA_LOSS, "Failed to parse incoming proto"),
-          true);
+      grpc_call_cancel_with_status(call(), GRPC_STATUS_DATA_LOSS, "Failed to parse incoming proto");
     }
   } else {
     ret = false;
@@ -132,9 +130,7 @@ bool StreamContext::Write(const google::protobuf::Message* msg, bool is_last) {
   if (msg) {
     grpc_byte_buffer* out_buf = nullptr;
     if (!SerializeProto(*msg, &out_buf)) {
-      FinishStream(Status(StatusCode::INVALID_ARGUMENT,
-                          "Failed to serialize outgoing proto"),
-                   true);
+      grpc_call_cancel_with_status(call(), GRPC_STATUS_INVALID_ARGUMENT, "Failed to serialize outgoing proto");
       return false;
     }
     int flag = is_last ? GRPC_WRITE_BUFFER_HINT : 0;
@@ -172,29 +168,21 @@ const Status& StreamContext::Wait() {
   grpc_event_finish(metadata_ev);
   // TODO(yangg) protect states by a mutex, including other places.
   if (!self_halfclosed_ || !peer_halfclosed_) {
-    FinishStream(Status::Cancelled, true);
-  } else {
-    grpc_event* finish_ev =
-        grpc_completion_queue_pluck(cq(), finished_tag(), gpr_inf_future);
-    GPR_ASSERT(finish_ev->type == GRPC_FINISHED);
-    final_status_ = Status(
-        static_cast<StatusCode>(finish_ev->data.finished.status),
-        finish_ev->data.finished.details ? finish_ev->data.finished.details
-                                         : "");
-    grpc_event_finish(finish_ev);
-  }
-  return final_status_;
-}
-
-void StreamContext::FinishStream(const Status& status, bool send) {
-  if (send) {
-    grpc_call_cancel(call());
-  }
+    Cancel();
+  } 
   grpc_event* finish_ev =
       grpc_completion_queue_pluck(cq(), finished_tag(), gpr_inf_future);
   GPR_ASSERT(finish_ev->type == GRPC_FINISHED);
+  final_status_ = Status(
+      static_cast<StatusCode>(finish_ev->data.finished.status),
+      finish_ev->data.finished.details ? finish_ev->data.finished.details
+                                       : "");
   grpc_event_finish(finish_ev);
-  final_status_ = status;
+  return final_status_;
+}
+
+void StreamContext::Cancel() {
+  grpc_call_cancel(call());
 }
 
 }  // namespace grpc
diff --git a/src/cpp/stream/stream_context.h b/src/cpp/stream/stream_context.h
index f70fe6daa3..4781f27a77 100644
--- a/src/cpp/stream/stream_context.h
+++ b/src/cpp/stream/stream_context.h
@@ -48,7 +48,7 @@ namespace grpc {
 class ClientContext;
 class RpcMethod;
 
-class StreamContext : public StreamContextInterface {
+class StreamContext final : public StreamContextInterface {
  public:
   StreamContext(const RpcMethod& method, ClientContext* context,
                 const google::protobuf::Message* request,
@@ -63,7 +63,7 @@ class StreamContext : public StreamContextInterface {
   bool Read(google::protobuf::Message* msg) override;
   bool Write(const google::protobuf::Message* msg, bool is_last) override;
   const Status& Wait() override;
-  void FinishStream(const Status& status, bool send) override;
+  void Cancel() override;
 
   google::protobuf::Message* request() override { return request_; }
   google::protobuf::Message* response() override { return result_; }