diff options
author | Craig Tiller <ctiller@google.com> | 2015-01-14 11:49:12 -0800 |
---|---|---|
committer | Craig Tiller <ctiller@google.com> | 2015-01-14 11:49:12 -0800 |
commit | 2ba0dacbde45ed3a491d78134244b3175c556494 (patch) | |
tree | 6cda9f2e4f49713dd409edb3fab752a83126c175 /src/cpp | |
parent | e4fcd6aa0e3f1d1c759f55cb72966693d8fa774b (diff) |
Update C++ code to set status via the C api.
This prevents mismatches from breaking tests.
Diffstat (limited to 'src/cpp')
-rw-r--r-- | src/cpp/stream/stream_context.cc | 38 | ||||
-rw-r--r-- | src/cpp/stream/stream_context.h | 4 |
2 files changed, 15 insertions, 27 deletions
diff --git a/src/cpp/stream/stream_context.cc b/src/cpp/stream/stream_context.cc index 720f3e27fe..aec226b23c 100644 --- a/src/cpp/stream/stream_context.cc +++ b/src/cpp/stream/stream_context.cc @@ -105,9 +105,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; @@ -125,9 +123,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; @@ -165,29 +161,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 c1a8d8ae52..d6bd7a2370 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_; } |