feat grpc: refactor StreamReadFuture, add Reader type erasure

Relates:(https://nda.ya.ru/t/Fbv9fMDT7R3izG
commit_hash:3e77b684dfdae7769dd1af4a5440d53cd14414cf

Author: kpavlov00

.mapping.json

@@ -2354,6 +2354,7 @@
   "grpc/include/userver/ugrpc/client/impl/retry_backoff.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/impl/retry_backoff.hpp",
   "grpc/include/userver/ugrpc/client/impl/retry_policy.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/impl/retry_policy.hpp",
   "grpc/include/userver/ugrpc/client/impl/rpc.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/impl/rpc.hpp",
+  "grpc/include/userver/ugrpc/client/impl/stream_read_future_impl_base.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/impl/stream_read_future_impl_base.hpp",
   "grpc/include/userver/ugrpc/client/impl/stub_any.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/impl/stub_any.hpp",
   "grpc/include/userver/ugrpc/client/impl/stub_handle.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/impl/stub_handle.hpp",
   "grpc/include/userver/ugrpc/client/impl/stub_pool.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/impl/stub_pool.hpp",

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

@@ -1,8 +1,5 @@
 #pragma once
 
-/// @file userver/ugrpc/client/impl/rpc.hpp
-/// @brief Classes representing an outgoing RPC
-
 #include <utility>
 
 #include <grpcpp/impl/codegen/proto_utils.h>
@@ -23,17 +20,29 @@ USERVER_NAMESPACE_BEGIN
 
 namespace ugrpc::client::impl {
 
-/// @brief Controls a single request -> response stream RPC
-///
-/// This class is not thread-safe except for `GetContext`.
-///
-/// The RPC is cancelled on destruction unless the stream is closed (`Read` has
-/// returned `false`). In that case the connection is not closed (it will be
-/// reused for new RPCs), and the server receives `RpcInterruptedError`
-/// immediately. gRPC provides no way to early-close a server-streaming RPC
-/// gracefully.
+template <typename Reader, typename Response>
+class StreamReadFutureImpl : public StreamReadFutureImplBase<Response> {
+public:
+    StreamReadFutureImpl(impl::StreamingCallState& state, Reader& reader, const Response& response) noexcept;
+
+    StreamReadFutureImpl(StreamReadFutureImpl&&) = delete;
+    StreamReadFutureImpl& operator=(StreamReadFutureImpl&&) = delete;
+
+    ~StreamReadFutureImpl() override;
+
+    bool IsReady() const noexcept override;
+
+    bool Get() override;
+
+private:
+    impl::StreamingCallState& state_;
+    Reader& reader_;
+    const Response& response_{};
+    bool valid_{};
+};
+
 template <typename Response>
-class [[nodiscard]] InputStream final {
+class InputStream final {
 public:
     template <typename Stub, typename Request>
     InputStream(
@@ -42,22 +51,14 @@ class [[nodiscard]] InputStream final {
         const Request& request
     );
 
-    ~InputStream();
-
     InputStream(InputStream&&) = delete;
     InputStream& operator=(InputStream&&) = delete;
 
+    ~InputStream();
+
     CallContext& GetContext() noexcept { return context_; }
     const CallContext& GetContext() const noexcept { return context_; }
 
-    /// @brief Await and read the next incoming message
-    ///
-    /// On end-of-input, `Finish` is called automatically.
-    ///
-    /// @param response where to put response on success
-    /// @returns `true` on success, `false` on end-of-input, task cancellation,
-    //           or if the stream is already closed for reads
-    /// @throws ugrpc::client::RpcError on an RPC error
     [[nodiscard]] bool Read(Response& response);
 
 private:
@@ -66,66 +67,24 @@ class [[nodiscard]] InputStream final {
     std::unique_ptr<grpc::ClientAsyncReader<Response>> stream_;
 };
 
-/// @brief Controls a request stream -> single response RPC
-///
-/// This class is not thread-safe except for `GetContext`.
-///
-/// The RPC is cancelled on destruction unless `Finish` has been called. In that
-/// case the connection is not closed (it will be reused for new RPCs), and the
-/// server receives `RpcInterruptedError` immediately.
 template <typename Request, typename Response>
-class [[nodiscard]] OutputStream final {
+class OutputStream final {
 public:
     template <typename Stub>
     OutputStream(CallParams&& params, PrepareClientStreamingCall<Stub, Request, Response> prepare_async_method);
 
-    ~OutputStream();
-
     OutputStream(OutputStream&&) = delete;
     OutputStream& operator=(OutputStream&&) = delete;
 
+    ~OutputStream();
+
     CallContext& GetContext() noexcept { return context_; }
     const CallContext& GetContext() const noexcept { return context_; }
 
-    /// @brief Write the next outgoing message
-    ///
-    /// `Write` doesn't store any references to `request`, so it can be
-    /// deallocated right after the call.
-    ///
-    /// @param request the next message to write
-    /// @return true if the data is going to the wire; false if the write
-    ///         operation failed (including due to task cancellation,
-    //          or if the stream is already closed for writes),
-    ///         in which case no more writes will be accepted,
-    ///         and the error details can be fetched from Finish
     [[nodiscard]] bool Write(const Request& request);
 
-    /// @brief Write the next outgoing message and check result
-    ///
-    /// `WriteAndCheck` doesn't store any references to `request`, so it can be
-    /// deallocated right after the call.
-    ///
-    /// `WriteAndCheck` verifies result of the write and generates exception
-    /// in case of issues.
-    ///
-    /// @param request the next message to write
-    /// @throws ugrpc::client::RpcError on an RPC error
-    /// @throws ugrpc::client::RpcCancelledError on task cancellation
-    /// @throws ugrpc::client::RpcError if the stream is already closed for writes
     void WriteAndCheck(const Request& request);
 
-    /// @brief Complete the RPC successfully
-    ///
-    /// Should be called once all the data is written. The server will then
-    /// send a single `Response`.
-    ///
-    /// `Finish` should not be called multiple times.
-    ///
-    /// The connection is not closed, it will be reused for new RPCs.
-    ///
-    /// @returns the single `Response` received after finishing the writes
-    /// @throws ugrpc::client::RpcError on an RPC error
-    /// @throws ugrpc::client::RpcCancelledError on task cancellation
     Response Finish();
 
 private:
@@ -135,110 +94,28 @@ class [[nodiscard]] OutputStream final {
     std::unique_ptr<grpc::ClientAsyncWriter<Request>> stream_;
 };
 
-/// @brief Controls a request stream -> response stream RPC
-///
-/// It is safe to call the following methods from different coroutines:
-///
-///   - `GetContext`;
-///   - one of (`Read`, `ReadAsync`);
-///   - one of (`Write`, `WritesDone`).
-///
-/// `WriteAndCheck` is NOT thread-safe.
-///
-/// The RPC is cancelled on destruction unless the stream is closed (`Read` has
-/// returned `false`). In that case the connection is not closed (it will be
-/// reused for new RPCs), and the server receives `RpcInterruptedError`
-/// immediately. gRPC provides no way to early-close a server-streaming RPC
-/// gracefully.
-///
-/// `Read` and `AsyncRead` can throw if error status is received from server.
-/// User MUST NOT call `Read` or `AsyncRead` again after failure of any of these
-/// operations.
-///
-/// `Write` and `WritesDone` methods do not throw, but indicate issues with
-/// the RPC by returning `false`.
-///
-/// `WriteAndCheck` is intended for ping-pong scenarios, when after write
-/// operation the user calls `Read` and vice versa.
-///
-/// If `Write` or `WritesDone` returns negative result, the user MUST NOT call
-/// any of these methods anymore.
-/// Instead the user SHOULD call `Read` method until the end of input. If
-/// `Write` or `WritesDone` finishes with negative result, finally `Read`
-/// will throw an exception.
-/// ## Usage example:
-///
-/// @snippet grpc/tests/stream_test.cpp concurrent bidirectional stream
-///
 template <typename Request, typename Response>
-class [[nodiscard]] BidirectionalStream final {
+class BidirectionalStream final {
 public:
-    using RawStream = grpc::ClientAsyncReaderWriter<Request, Response>;
-    using StreamReadFuture = ugrpc::client::StreamReadFuture<RawStream>;
-
     template <typename Stub>
     BidirectionalStream(CallParams&& params, PrepareBidiStreamingCall<Stub, Request, Response> prepare_async_method);
 
-    ~BidirectionalStream();
-
     BidirectionalStream(BidirectionalStream&&) = delete;
     BidirectionalStream& operator=(BidirectionalStream&&) = delete;
 
+    ~BidirectionalStream();
+
     CallContext& GetContext() noexcept { return context_; }
     const CallContext& GetContext() const noexcept { return context_; }
 
-    /// @brief Await and read the next incoming message
-    ///
-    /// On end-of-input, `Finish` is called automatically.
-    ///
-    /// @param response where to put response on success
-    /// @returns `true` on success, `false` on end-of-input, task cancellation,
-    ///              or if the stream is already closed for reads
-    /// @throws ugrpc::client::RpcError on an RPC error
     [[nodiscard]] bool Read(Response& response);
 
-    /// @brief Return future to read next incoming result
-    ///
-    /// @param response where to put response on success
-    /// @return StreamReadFuture future
-    /// @throws ugrpc::client::RpcError on an RPC error
-    /// @throws ugrpc::client::RpcError if the stream is already closed for reads
-    StreamReadFuture ReadAsync(Response& response);
-
-    /// @brief Write the next outgoing message
-    ///
-    /// RPC will be performed immediately. No references to `request` are
-    /// saved, so it can be deallocated right after the call.
-    ///
-    /// @param request the next message to write
-    /// @return true if the data is going to the wire; false if the write
-    ///         operation failed (including due to task cancellation,
-    //          or if the stream is already closed for writes),
-    ///         in which case no more writes will be accepted,
-    ///         but Read may still have some data and status code available
+    StreamReadFuture<Response> ReadAsync(Response& response);
+
     [[nodiscard]] bool Write(const Request& request);
 
-    /// @brief Write the next outgoing message and check result
-    ///
-    /// `WriteAndCheck` doesn't store any references to `request`, so it can be
-    /// deallocated right after the call.
-    ///
-    /// `WriteAndCheck` verifies result of the write and generates exception
-    /// in case of issues.
-    ///
-    /// @param request the next message to write
-    /// @throws ugrpc::client::RpcError on an RPC error
-    /// @throws ugrpc::client::RpcCancelledError on task cancellation
-    /// @throws ugrpc::client::RpcError if the stream is already closed for writes
     void WriteAndCheck(const Request& request);
 
-    /// @brief Announce end-of-output to the server
-    ///
-    /// Should be called to notify the server and receive the final response(s).
-    ///
-    /// @return true if the data is going to the wire; false if the operation
-    ///         failed (including if the stream is already closed for writes),
-    ///         but Read may still have some data and status code available
     [[nodiscard]] bool WritesDone();
 
 private:
@@ -247,6 +124,50 @@ class [[nodiscard]] BidirectionalStream final {
     std::unique_ptr<grpc::ClientAsyncReaderWriter<Request, Response>> stream_;
 };
 
+template <typename Reader, typename Response>
+StreamReadFutureImpl<Reader, Response>::StreamReadFutureImpl(
+    impl::StreamingCallState& state,
+    Reader& reader,
+    const Response& response
+) noexcept
+        : state_{state}, reader_{reader}, response_{response}, valid_{true} {}
+
+template <typename Reader, typename Response>
+StreamReadFutureImpl<Reader, Response>::~StreamReadFutureImpl() {
+    if (valid_) {
+        // StreamReadFuture::Get wasn't called => finish RPC.
+        impl::FinishAbandoned(reader_, state_);
+    }
+}
+
+template <typename Reader, typename Response>
+bool StreamReadFutureImpl<Reader, Response>::IsReady() const noexcept {
+    UINVARIANT(valid_, "IsReady should be called only before 'Get'");
+    return state_.GetAsyncMethodInvocation().IsReady();
+}
+
+template <typename Reader, typename Response>
+bool StreamReadFutureImpl<Reader, Response>::Get() {
+    UINVARIANT(valid_, "'Get' must be called only once");
+    valid_ = false;
+    const impl::StreamingCallState::AsyncMethodInvocationGuard guard(state_);
+    const auto
+        wait_status = impl::WaitAndTryCancelIfNeeded(state_.GetAsyncMethodInvocation(), state_.GetClientContext());
+    if (wait_status == ugrpc::impl::AsyncMethodInvocation::WaitStatus::kCancelled) {
+        state_.GetStatsScope().OnCancelled();
+        state_.GetStatsScope().Flush();
+    } else if (wait_status == ugrpc::impl::AsyncMethodInvocation::WaitStatus::kError) {
+        // Finish can only be called once all the data is read, otherwise the
+        // underlying gRPC driver hangs.
+        impl::Finish(reader_, state_, /*final_response=*/nullptr, /*throw_on_error=*/true);
+    } else {
+        if constexpr (std::is_base_of_v<google::protobuf::Message, Response>) {
+            RunMiddlewarePipeline(state_, impl::MiddlewareHooks::RecvMessageHooks(response_));
+        }
+    }
+    return wait_status == ugrpc::impl::AsyncMethodInvocation::WaitStatus::kOk;
+}
+
 template <typename Response>
 template <typename Stub, typename Request>
 InputStream<Response>::InputStream(
@@ -394,17 +315,18 @@ BidirectionalStream<Request, Response>::~BidirectionalStream() {
 }
 
 template <typename Request, typename Response>
-typename BidirectionalStream<Request, Response>::StreamReadFuture BidirectionalStream<
-    Request,
-    Response>::ReadAsync(Response& response) {
+StreamReadFuture<Response> BidirectionalStream<Request, Response>::ReadAsync(Response& response) {
     if (!IsReadAvailable(state_)) {
         // If the stream is already finished, we must exit immediately.
         // If not, even the middlewares may access something that is already dead.
         throw RpcError(state_.GetCallName(), "'ReadAsync' called on a finished call");
     }
 
     impl::ReadAsync(*stream_, response, state_);
-    return StreamReadFuture{state_, *stream_, ToBaseMessage(&response)};
+    using BidirectionalStreamReadFutureImpl = StreamReadFutureImpl<grpc::ClientAsyncReaderWriter<Request, Response>, Response>;
+    return StreamReadFuture<Response>{
+        std::make_unique<BidirectionalStreamReadFutureImpl>(state_, *stream_, response)
+    };
 }
 
 template <typename Request, typename Response>

grpc/include/userver/ugrpc/client/impl/stream_read_future_impl_base.hpp

@@ -0,0 +1,19 @@
+#pragma once
+
+USERVER_NAMESPACE_BEGIN
+
+namespace ugrpc::client::impl {
+
+template <typename Response>
+class StreamReadFutureImplBase {
+public:
+    virtual ~StreamReadFutureImplBase() = default;
+
+    virtual bool IsReady() const = 0;
+
+    virtual bool Get() = 0;
+};
+
+}  // namespace ugrpc::client::impl
+
+USERVER_NAMESPACE_END

grpc/include/userver/ugrpc/client/stream.hpp

@@ -4,8 +4,10 @@
 /// @brief Client streaming interfaces
 
 #include <memory>
+#include <utility>
 
 #include <userver/ugrpc/client/impl/rpc.hpp>
+#include <userver/ugrpc/client/stream_read_future.hpp>
 
 USERVER_NAMESPACE_BEGIN
 
@@ -168,9 +170,6 @@ class [[nodiscard]] Writer final {
 template <typename Request, typename Response>
 class [[nodiscard]] ReaderWriter final {
 public:
-    using StreamReadFuture = ugrpc::client::StreamReadFuture<
-        typename impl::BidirectionalStream<Request, Response>::RawStream>;
-
     /// @cond
     // For internal use only
     template <typename Stub>
@@ -202,7 +201,7 @@ class [[nodiscard]] ReaderWriter final {
     /// @return StreamReadFuture future
     /// @throws ugrpc::client::RpcError on an RPC error
     /// @throws ugrpc::client::RpcError if the stream is already closed for reads
-    StreamReadFuture ReadAsync(Response& response) { return stream_->ReadAsync(response); }
+    StreamReadFuture<Response> ReadAsync(Response& response) { return stream_->ReadAsync(response); }
 
     /// @brief Write the next outgoing message
     ///