@@ -27,14 +27,15 @@ void SetErrorAndResetSpan(CallState& state, std::string_view error_message) noex
2727 state.ResetSpan ();
2828}
2929
30- void HandleCallStatistics (CallState& state, const grpc::Status& status) noexcept {
31- auto & stats = state.GetStatsScope ();
32- if (grpc::StatusCode::DEADLINE_EXCEEDED == status.error_code () && state.IsDeadlinePropagated ()) {
33- stats.OnCancelledByDeadlinePropagation ();
34- } else {
35- stats.OnExplicitFinish (status.error_code ());
30+ void TryRunMiddlewarePipeline (CallState& state, bool throw_on_error, const MiddlewareHooks& hooks) {
31+ try {
32+ RunMiddlewarePipeline (state, hooks);
33+ } catch (std::exception& ex) {
34+ LOG_WARNING () << " There is a caught exception in 'Finish': "
35+ if (throw_on_error) {
36+ throw ;
37+ }
3638 }
37- stats.Flush ();
3839}
3940
4041} // namespace
@@ -66,55 +67,92 @@ void CheckOk(
6667) {
6768 if (wait_status == ugrpc::impl::AsyncMethodInvocation::WaitStatus::kError ) {
6869 state.SetFinished ();
69- ThrowIfDeadlineIsExceeded (state.GetClientContext (), state.GetCallName ());
70- ProcessNetworkError (state, stage);
71- throw RpcInterruptedError (state.GetCallName (), stage);
70+ ProcessNetworkError (state, true , stage);
7271 } else if (wait_status == ugrpc::impl::AsyncMethodInvocation::WaitStatus::kCancelled ) {
7372 state.SetFinished ();
74- ProcessCancelled (state, stage);
75- throw RpcCancelledError (state.GetCallName (), stage);
73+ if (impl::IsTaskCancelledByDeadlinePropagation ()) {
74+ ProcessTimeoutDeadlinePropagated (state, true , stage);
75+ } else {
76+ ProcessCancelled (state, true , stage);
77+ }
7678 }
7779}
7880
7981void ProcessFinish (
8082 StreamingCallState& state,
81- const CompletionStatus& completion_status,
83+ bool throw_on_error,
84+ CompletionStatus&& completion_status,
8285 const google::protobuf::Message* response
8386) {
8487 UINVARIANT (completion_status.has_value (), " ProcessFinish must be called only with grpc::Status completions"
85- const auto & status = completion_status.value ();
88+ auto & status = completion_status.value ();
89+
90+ TryRunMiddlewarePipeline (
91+ state,
92+ throw_on_error,
93+ MiddlewareHooks::FinishHooks (status, status.ok () ? response : nullptr )
94+ );
8695
87- RunMiddlewarePipeline ( state, MiddlewareHooks::FinishHooks ( status, status. ok () ? response : nullptr ));
88- impl::HandleCallStatistics ( state, status );
96+ state. GetStatsScope (). OnExplicitFinish ( status. error_code ( ));
97+ state. GetStatsScope (). Flush ( );
8998 SetStatusAndResetSpan (state, status);
99+
100+ if (throw_on_error && !status.ok ()) {
101+ ThrowErrorWithStatus (state.GetCallName (), std::move (status));
102+ }
90103}
91104
92- void ProcessFinishAbandoned (StreamingCallState& state, const grpc::Status& status) noexcept {
93- RunMiddlewarePipeline (
105+ void ProcessTimeoutDeadlinePropagated (StreamingCallState& state, bool throw_on_error, std::string_view stage) {
106+ TryRunMiddlewarePipeline (
94107 state,
95- MiddlewareHooks::FinishHooks (utils::unexpected{SpecialCaseCompletionType::kAbandoned }, nullptr )
108+ throw_on_error,
109+ MiddlewareHooks::FinishHooks (utils::unexpected{SpecialCaseCompletionType::kTimeoutDeadlinePropagated }, nullptr )
96110 );
97111
98- // Nothing to do with statistics, `RpcStatisticsScope` automatically accounts "abandoned-error"
99- SetStatusAndResetSpan (state, status);
112+ state.GetStatsScope ().OnCancelledByDeadlinePropagation ();
113+ state.GetStatsScope ().Flush ();
114+ SetErrorAndResetSpan (state, fmt::format (" Deadline propagated at '{}'"
115+
116+ if (throw_on_error) {
117+ throw RpcCancelledError (state.GetCallName (), fmt::format (" {} (Deadline Propagation)"
118+ }
100119}
101120
102- void ProcessCancelled (StreamingCallState& state, std::string_view stage) noexcept {
121+ void ProcessCancelled (StreamingCallState& state, bool throw_on_error, std::string_view stage) {
103122 CompletionStatus result{utils::unexpected{SpecialCaseCompletionType::kCancelled }};
104- RunMiddlewarePipeline (state, MiddlewareHooks::FinishHooks (result, nullptr ));
123+ TryRunMiddlewarePipeline (state, throw_on_error , MiddlewareHooks::FinishHooks (result, nullptr ));
105124
106125 state.GetStatsScope ().OnCancelled ();
107126 state.GetStatsScope ().Flush ();
108127 SetErrorAndResetSpan (state, fmt::format (" Task cancellation at '{}'"
128+
129+ if (throw_on_error) {
130+ throw RpcCancelledError (state.GetCallName (), stage);
131+ }
109132}
110133
111- void ProcessNetworkError (StreamingCallState& state, std::string_view stage) noexcept {
134+ void ProcessNetworkError (StreamingCallState& state, bool throw_on_error, std::string_view stage) {
112135 CompletionStatus result{utils::unexpected{SpecialCaseCompletionType::kNetworkError }};
113- RunMiddlewarePipeline (state, MiddlewareHooks::FinishHooks (result, nullptr ));
136+ TryRunMiddlewarePipeline (state, throw_on_error , MiddlewareHooks::FinishHooks (result, nullptr ));
114137
115138 state.GetStatsScope ().OnNetworkError ();
116139 state.GetStatsScope ().Flush ();
117140 SetErrorAndResetSpan (state, fmt::format (" Network error at '{}'"
141+
142+ if (throw_on_error) {
143+ throw RpcInterruptedError (state.GetCallName (), stage);
144+ }
145+ }
146+
147+ void ProcessFinishAbandoned (StreamingCallState& state, const grpc::Status& status) noexcept {
148+ TryRunMiddlewarePipeline (
149+ state,
150+ false ,
151+ MiddlewareHooks::FinishHooks (utils::unexpected{SpecialCaseCompletionType::kAbandoned }, nullptr )
152+ );
153+
154+ // Nothing to do with statistics, `RpcStatisticsScope` automatically accounts "abandoned-error"
155+ SetStatusAndResetSpan (state, status);
118156}
119157
120158} // namespace ugrpc::client::impl
0 commit comments