Added Ru return in GetConsumedRu method with streaming calls in C++ SDK (#15095)

Author: Gazizonoki

src/client/impl/ydb_internal/grpc_connections/grpc_connections.h

@@ -459,7 +459,6 @@ class TGRpcConnectionsImpl
                                 }
                             };
                             processor->AddFinishedCallback(std::move(finishedCallback));
-                            // TODO: Add headers for streaming calls.
                             TPlainStatus status(std::move(grpcStatus), endpoint.GetEndpoint(), {});
                             responseCb(std::move(status), std::move(processor));
                         } else {
@@ -468,7 +467,6 @@ class TGRpcConnectionsImpl
                             if (grpcStatus.GRpcStatusCode != grpc::StatusCode::CANCELLED) {
                                 dbState->EndpointPool.BanEndpoint(endpoint.GetEndpoint());
                             }
-                            // TODO: Add headers for streaming calls.
                             TPlainStatus status(std::move(grpcStatus), endpoint.GetEndpoint(), {});
                             responseCb(std::move(status), nullptr);
                         }
@@ -560,7 +558,6 @@ class TGRpcConnectionsImpl
                                 }
                             };
                             processor->AddFinishedCallback(std::move(finishedCallback));
-                            // TODO: Add headers for streaming calls.
                             TPlainStatus status(std::move(grpcStatus), endpoint.GetEndpoint(), {});
                             connectedCallback(std::move(status), std::move(processor));
                         } else {
@@ -569,7 +566,6 @@ class TGRpcConnectionsImpl
                             if (grpcStatus.GRpcStatusCode != grpc::StatusCode::CANCELLED) {
                                 dbState->EndpointPool.BanEndpoint(endpoint.GetEndpoint());
                             }
-                            // TODO: Add headers for streaming calls.
                             TPlainStatus status(std::move(grpcStatus), endpoint.GetEndpoint(), {});
                             connectedCallback(std::move(status), nullptr);
                         }

src/client/impl/ydb_internal/plain_status/status.cpp

@@ -1,6 +1,8 @@
 #define INCLUDE_YDB_INTERNAL_H
 #include "status.h"
 
+#include <ydb-cpp-sdk/client/resources/ydb_resources.h>
+
 #include <util/string/builder.h>
 
 namespace NYdb::inline V3 {
@@ -62,10 +64,26 @@ TPlainStatus::TPlainStatus(
             std::string(value.begin(), value.end())
         );
     }
+
+    InitCostInfo();
 }
 
 TPlainStatus TPlainStatus::Internal(const std::string& message) {
     return { EStatus::CLIENT_INTERNAL_ERROR, "Internal client error: " + message };
 }
 
+void TPlainStatus::InitCostInfo() {
+    if (auto metaIt = Metadata.find(YDB_CONSUMED_UNITS_HEADER); metaIt != Metadata.end()) {
+        try {
+            CostInfo.set_consumed_units(std::stod(metaIt->second));
+        } catch (std::exception& e) {
+            if (Ok()) {
+                Status = EStatus::CLIENT_INTERNAL_ERROR;
+            }
+
+            Issues.AddIssue(NIssue::TIssue{"Failed to parse CostInfo from Metadata: " + std::string{e.what()}});
+        }
+    }
+}
+
 } // namespace NYdb

src/client/impl/ydb_internal/plain_status/status.h

@@ -18,7 +18,7 @@ struct TPlainStatus {
     NYdb::NIssue::TIssues Issues;
     std::string Endpoint;
     std::multimap<std::string, std::string> Metadata;
-    Ydb::CostInfo ConstInfo;
+    Ydb::CostInfo CostInfo;
 
     TPlainStatus()
         : Status(EStatus::SUCCESS)
@@ -35,7 +35,9 @@ struct TPlainStatus {
         , Issues(std::move(issues))
         , Endpoint(endpoint)
         , Metadata(std::move(metadata))
-    { }
+    {
+        InitCostInfo();
+    }
 
     TPlainStatus(EStatus status, const std::string& message)
         : Status(status)
@@ -51,7 +53,7 @@ struct TPlainStatus {
 
     template<class T>
     void SetCostInfo(T&& costInfo) {
-        ConstInfo = std::forward<T>(costInfo);
+        CostInfo = std::forward<T>(costInfo);
     }
 
     bool Ok() const {
@@ -73,7 +75,8 @@ struct TPlainStatus {
         return ret;
     }
 
-
+private:
+    void InitCostInfo();
 };
 
 } // namespace NYdb

src/client/table/impl/readers.cpp

@@ -78,7 +78,6 @@ TAsyncScanQueryPart TScanQueryPartIterator::TReaderImpl::ReadNext(std::shared_pt
             NYdb::NIssue::TIssues issues;
             NYdb::NIssue::IssuesFromMessage(self->Response_.issues(), issues);
             EStatus clientStatus = static_cast<EStatus>(self->Response_.status());
-            // TODO: Add headers for streaming calls.
             TPlainStatus plainStatus{clientStatus, std::move(issues), self->Endpoint_, {}};
             TStatus status{std::move(plainStatus)};
             std::optional<TQueryStats> queryStats;

src/client/types/status/status.cpp

@@ -71,7 +71,7 @@ const std::multimap<std::string, std::string>& TStatus::GetResponseMetadata() co
 }
 
 float TStatus::GetConsumedRu() const {
-    return Impl_->Status.ConstInfo.consumed_units();
+    return Impl_->Status.CostInfo.consumed_units();
 }
 
 void TStatus::Out(IOutputStream& out) const {