Moved commit "Replication consistency levels support in C++ SDK" from ydb repo

Author: CyberROFL

include/ydb-cpp-sdk/client/draft/ydb_replication.h

@@ -9,6 +9,7 @@
 
 namespace Ydb::Replication {
     class ConnectionParams;
+    class ConsistencyLevelGlobal;
     class DescribeReplicationResult;
     class DescribeReplicationResult_Stats;
 }
@@ -60,6 +61,19 @@ class TConnectionParams: private TCommonClientSettings {
     > Credentials_;
 };
 
+struct TRowConsistency {
+};
+
+class TGlobalConsistency {
+public:
+    explicit TGlobalConsistency(const Ydb::Replication::ConsistencyLevelGlobal& proto);
+
+    const TDuration& GetCommitInterval() const;
+
+private:
+    TDuration CommitInterval_;
+};
+
 class TStats {
 public:
     TStats() = default;
@@ -108,6 +122,11 @@ class TReplicationDescription {
         std::optional<std::string> SrcChangefeedName;
     };
 
+    enum class EConsistencyLevel {
+        Row,
+        Global,
+    };
+
     enum class EState {
         Running,
         Error,
@@ -119,6 +138,9 @@ class TReplicationDescription {
     const TConnectionParams& GetConnectionParams() const;
     const std::vector<TItem> GetItems() const;
 
+    EConsistencyLevel GetConsistencyLevel() const;
+    const TGlobalConsistency& GetGlobalConsistency() const;
+
     EState GetState() const;
     const TRunningState& GetRunningState() const;
     const TErrorState& GetErrorState() const;
@@ -127,6 +149,12 @@ class TReplicationDescription {
 private:
     TConnectionParams ConnectionParams_;
     std::vector<TItem> Items_;
+
+    std::variant<
+        TRowConsistency,
+        TGlobalConsistency
+    > ConsistencyLevel_;
+
     std::variant<
         TRunningState,
         TErrorState,

src/api/protos/draft/ydb_replication.proto

@@ -45,6 +45,13 @@ message ConnectionParams {
     }
 }
 
+message ConsistencyLevelRow {
+}
+
+message ConsistencyLevelGlobal {
+    google.protobuf.Duration commit_interval = 1;
+}
+
 message DescribeReplicationResult {
     message Stats {
         optional google.protobuf.Duration lag = 1;
@@ -72,7 +79,12 @@ message DescribeReplicationResult {
 
     // Description of scheme object.
     Ydb.Scheme.Entry self = 1;
+
     ConnectionParams connection_params = 2;
+    oneof consistency_level {
+        ConsistencyLevelRow row_consistency = 7;
+        ConsistencyLevelGlobal global_consistency = 8;
+    }
     repeated Item items = 3;
     oneof state {
         RunningState running = 4;

src/client/draft/ydb_replication.cpp

@@ -68,6 +68,15 @@ static TDuration DurationToDuration(const google::protobuf::Duration& value) {
     return TDuration::MilliSeconds(google::protobuf::util::TimeUtil::DurationToMilliseconds(value));
 }
 
+TGlobalConsistency::TGlobalConsistency(const Ydb::Replication::ConsistencyLevelGlobal& proto)
+    : CommitInterval_(DurationToDuration(proto.commit_interval()))
+{
+}
+
+const TDuration& TGlobalConsistency::GetCommitInterval() const {
+    return CommitInterval_;
+}
+
 TStats::TStats(const Ydb::Replication::DescribeReplicationResult_Stats& stats)
     : Lag_(stats.has_lag() ? std::make_optional(DurationToDuration(stats.lag())) : std::nullopt)
     , InitialScanProgress_(stats.has_initial_scan_progress() ? std::make_optional(stats.initial_scan_progress()) : std::nullopt)
@@ -131,6 +140,15 @@ TReplicationDescription::TReplicationDescription(const Ydb::Replication::Describ
         });
     }
 
+    switch (desc.consistency_level_case()) {
+    case Ydb::Replication::DescribeReplicationResult::kGlobalConsistency:
+        ConsistencyLevel_ = TGlobalConsistency(desc.global_consistency());
+        break;
+
+    default:
+        break;
+    }
+
     switch (desc.state_case()) {
     case Ydb::Replication::DescribeReplicationResult::kRunning:
         State_ = TRunningState(desc.running().stats());
@@ -157,6 +175,15 @@ const std::vector<TReplicationDescription::TItem> TReplicationDescription::GetIt
     return Items_;
 }
 
+TReplicationDescription::EConsistencyLevel TReplicationDescription::GetConsistencyLevel() const {
+    return static_cast<EConsistencyLevel>(ConsistencyLevel_.index());
+}
+
+const TGlobalConsistency& TReplicationDescription::GetGlobalConsistency() const {
+    return std::get<TGlobalConsistency>(ConsistencyLevel_);
+}
+
+
 TReplicationDescription::EState TReplicationDescription::GetState() const {
     return static_cast<EState>(State_.index());
 }