refactor grpc: refactor server Responder

commit_hash:0317eaa3f81fb9510d403fab69d338bb5004b478

Author: kpavlov00

grpc/include/userver/ugrpc/server/impl/async_methods.hpp

@@ -16,49 +16,47 @@ namespace ugrpc::server::impl {
 extern const grpc::Status kUnimplementedStatus;
 extern const grpc::Status kUnknownErrorStatus;
 
-void CheckInvocationSucceeded(bool ok, std::string_view call_name, std::string_view stage);
-
 template <typename GrpcStream, typename Response>
 [[nodiscard]] bool Finish(GrpcStream& stream, const Response& response, const grpc::Status& status) {
-    ugrpc::impl::AsyncMethodInvocation finish;
-    stream.Finish(response, status, finish.GetCompletionTag());
-    return finish.WaitNonCancellable();
+    ugrpc::impl::AsyncMethodInvocation invocation;
+    stream.Finish(response, status, invocation.GetCompletionTag());
+    return invocation.WaitNonCancellable();
 }
 
 template <typename GrpcStream>
 [[nodiscard]] bool Finish(GrpcStream& stream, const grpc::Status& status) {
-    ugrpc::impl::AsyncMethodInvocation finish;
-    stream.Finish(status, finish.GetCompletionTag());
-    return finish.WaitNonCancellable();
+    ugrpc::impl::AsyncMethodInvocation invocation;
+    stream.Finish(status, invocation.GetCompletionTag());
+    return invocation.WaitNonCancellable();
 }
 
 template <typename GrpcStream>
 [[nodiscard]] bool FinishWithError(GrpcStream& stream, const grpc::Status& status) {
     UASSERT(!status.ok());
-    ugrpc::impl::AsyncMethodInvocation finish;
-    stream.FinishWithError(status, finish.GetCompletionTag());
-    return finish.WaitNonCancellable();
+    ugrpc::impl::AsyncMethodInvocation invocation;
+    stream.FinishWithError(status, invocation.GetCompletionTag());
+    return invocation.WaitNonCancellable();
 }
 
 template <typename GrpcStream>
-void SendInitialMetadata(GrpcStream& stream, std::string_view call_name) {
-    ugrpc::impl::AsyncMethodInvocation metadata;
-    stream.SendInitialMetadata(metadata.GetCompletionTag());
-    CheckInvocationSucceeded(metadata.WaitNonCancellable(), call_name, "SendInitialMetadata");
+[[nodiscard]] bool SendInitialMetadata(GrpcStream& stream) {
+    ugrpc::impl::AsyncMethodInvocation invocation;
+    stream.SendInitialMetadata(invocation.GetCompletionTag());
+    return invocation.WaitNonCancellable();
 }
 
 template <typename GrpcStream, typename Request>
 [[nodiscard]] bool Read(GrpcStream& stream, Request& request) {
-    ugrpc::impl::AsyncMethodInvocation read;
-    stream.Read(&request, read.GetCompletionTag());
-    return read.WaitNonCancellable();
+    ugrpc::impl::AsyncMethodInvocation invocation;
+    stream.Read(&request, invocation.GetCompletionTag());
+    return invocation.WaitNonCancellable();
 }
 
 template <typename GrpcStream, typename Response>
-void Write(GrpcStream& stream, const Response& response, grpc::WriteOptions options, std::string_view call_name) {
-    ugrpc::impl::AsyncMethodInvocation write;
-    stream.Write(response, options, write.GetCompletionTag());
-    CheckInvocationSucceeded(write.WaitNonCancellable(), call_name, "Write");
+[[nodiscard]] bool Write(GrpcStream& stream, const Response& response, grpc::WriteOptions options) {
+    ugrpc::impl::AsyncMethodInvocation invocation;
+    stream.Write(response, options, invocation.GetCompletionTag());
+    return invocation.WaitNonCancellable();
 }
 
 template <typename GrpcStream, typename Response>
@@ -68,9 +66,9 @@ template <typename GrpcStream, typename Response>
     grpc::WriteOptions options,
     const grpc::Status& status
 ) {
-    ugrpc::impl::AsyncMethodInvocation write_and_finish;
-    stream.WriteAndFinish(response, options, status, write_and_finish.GetCompletionTag());
-    return write_and_finish.WaitNonCancellable();
+    ugrpc::impl::AsyncMethodInvocation invocation;
+    stream.WriteAndFinish(response, options, status, invocation.GetCompletionTag());
+    return invocation.WaitNonCancellable();
 }
 
 }  // namespace ugrpc::server::impl

grpc/include/userver/ugrpc/server/impl/call_processor.hpp

@@ -119,9 +119,9 @@ class CallProcessor final {
             impl::UnpackResult(std::move(result), final_response, Status());
         });
 
-        // Streaming handler can detect RPC breakage during a network interaction => IsFinished.
+        // Streaming handler can detect RPC breakage during a network interaction => IsInterrupted().
         // RpcFinishedEvent can signal RPC interruption while in the handler => ShouldCancel.
-        if (responder_.IsFinished() || engine::current_task::ShouldCancel()) {
+        if (responder_.IsInterrupted() || engine::current_task::ShouldCancel()) {
             impl::ReportRpcInterruptedError(state_);
             // Don't run OnCallFinish.
             return;
@@ -223,7 +223,7 @@ class CallProcessor final {
         } catch (const USERVER_NAMESPACE::server::handlers::CustomHandlerException& ex) {
             Status() = impl::ReportCustomError(ex, state_);
         } catch (const RpcInterruptedError& /*ex*/) {
-            UASSERT(responder_.IsFinished());
+            UASSERT(responder_.IsInterrupted());
             // RPC interruption will be reported below.
         } catch (const std::exception& ex) {
             Status() = impl::ReportHandlerError(ex, state_);

grpc/include/userver/ugrpc/server/impl/rpc.hpp

@@ -66,7 +66,7 @@ class Responder final : public ResponderBase, public CallTraits::StreamAdapter {
     Responder& operator=(Responder&&) = delete;
     ~Responder() override;
 
-    bool IsFinished() const { return is_finished_; }
+    bool IsInterrupted() const { return is_interrupted_; }
 
     /// @brief Await and read the next incoming message. Only makes sense for client-streaming RPCs.
     /// @param request where to put the request on success
@@ -113,6 +113,7 @@ class Responder final : public ResponderBase, public CallTraits::StreamAdapter {
     RawResponder& raw_responder_;
     // Separate flags are required to be able to set them in parallel in Read and Write.
     bool are_reads_done_{kCallKind == CallKind::kUnaryCall};
+    bool is_interrupted_{false};
     bool is_finished_{false};
 };
 
@@ -124,7 +125,7 @@ Responder<CallTraits>::Responder(CallState& state, RawResponder& raw_responder)
 
 template <typename CallTraits>
 Responder<CallTraits>::~Responder() {
-    UASSERT(is_finished_ || engine::current_task::ShouldCancel());
+    UASSERT(is_finished_ || is_interrupted_ || engine::current_task::ShouldCancel());
 }
 
 template <typename CallTraits>
@@ -146,7 +147,7 @@ bool Responder<CallTraits>::DoRead(Request& request) {
 template <typename CallTraits>
 void Responder<CallTraits>::DoWrite(Response& response, const grpc::WriteOptions& options) {
     static_assert(impl::IsServerStreaming(kCallKind));
-    UINVARIANT(!is_finished_, "'Write' called on a finished stream");
+    UINVARIANT(!is_interrupted_, "'Write' called on an interrupted stream");
 
     if constexpr (std::is_base_of_v<google::protobuf::Message, Response>) {
         ApplyResponseHook(response);
@@ -156,20 +157,16 @@ void Responder<CallTraits>::DoWrite(Response& response, const grpc::WriteOptions
         // For some reason, gRPC requires explicit 'SendInitialMetadata' in output streams.
         if (!are_reads_done_) {
             are_reads_done_ = true;
-            try {
-                impl::SendInitialMetadata(raw_responder_, GetCallName());
-            } catch (const RpcInterruptedError&) {
-                is_finished_ = true;
-                throw;
+            if (!impl::SendInitialMetadata(raw_responder_)) {
+                is_interrupted_ = true;
+                throw RpcInterruptedError(GetCallName(), "SendInitialMetadata");
             }
         }
     }
 
-    try {
-        impl::Write(raw_responder_, response, options, GetCallName());
-    } catch (const RpcInterruptedError&) {
-        is_finished_ = true;
-        throw;
+    if (!impl::Write(raw_responder_, response, options)) {
+        is_interrupted_ = true;
+        throw RpcInterruptedError(GetCallName(), "Write");
     }
 }
 

grpc/src/ugrpc/server/impl/async_methods.cpp

@@ -13,12 +13,6 @@ const grpc::Status kUnknownErrorStatus{
     "The service method has exited unexpectedly, without providing a status"
 };
 
-void CheckInvocationSucceeded(bool ok, std::string_view call_name, std::string_view stage) {
-    if (!ok) {
-        throw RpcInterruptedError(call_name, stage);
-    }
-}
-
 }  // namespace ugrpc::server::impl
 
 USERVER_NAMESPACE_END