RS-311: add each_n and each_s replication settings (#13)

* add each_n and each_s replication settings * set settings * update CHANGELOG * fix test for 1.9 api * fix token mask
reductstore · Jun 4, 2024 · afdd1a8 · afdd1a8
1 parent 0cdc848
commit afdd1a8
Show file tree

Hide file tree

Showing 4 changed files with 73 additions and 19 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -9,7 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
-- RS-261: Add `each_n` and `each_s` query parameters, [PR-11](https://github.com/reductstore/reduct-rs/pull/11)
+- RS-261: add `each_n` and `each_s` query parameters, [PR-11](https://github.com/reductstore/reduct-rs/pull/11)
+- RS-311: add `each_n` and `each_s` replication settings, [PR-13](https://github.com/reductstore/reduct-rs/pull/13)
 
 ### Changed
 

diff --git a/Cargo.toml b/Cargo.toml
@@ -1,7 +1,7 @@
 [package]
 name = "reduct-rs"
 
-version = "1.9.5"
+version = "1.10.0"
 authors =["Alexey Timin <[email protected]>"]
 edition = "2021"
 rust-version = "1.75.0"
@@ -23,13 +23,13 @@ test-api-110 = []   # Test API 1.10
 crate-type = ["lib"]
 
 [dependencies]
-reduct-base = "1.9.4"
+reduct-base = { git = "https://github.com/reductstore/reductstore.git", branch = "main", package = "reduct-base" }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls", "stream"] }
 http = "1.0.0"
 rustls = "0.23"
-chrono = { version = "0.4.11", features = ["serde"] }
+chrono = { version = "0.4.38", features = ["serde"] }
 bytes = "1.4.0"
 futures = "0.3.17"
 futures-util = "0.3.28"

diff --git a/src/client.rs b/src/client.rs
@@ -565,6 +565,8 @@ pub(crate) mod tests {
                 .entries(settings.entries.clone())
                 .include(settings.include.clone())
                 .exclude(settings.exclude.clone())
+                .each_s(settings.each_s.unwrap())
+                .each_n(settings.each_n.unwrap())
                 .send()
                 .await
                 .unwrap();
@@ -595,14 +597,25 @@ pub(crate) mod tests {
                     pending_records: 0,
                 }
             );
-
-            assert_eq!(
-                replication.settings,
-                ReplicationSettings {
-                    dst_token: "***".to_string(),
-                    ..settings
-                }
-            );
+            if cfg!(feature = "test-api-110") {
+                assert_eq!(
+                    replication.settings,
+                    ReplicationSettings {
+                        dst_token: "***".to_string(),
+                        ..settings
+                    }
+                );
+            } else {
+                assert_eq!(
+                    replication.settings,
+                    ReplicationSettings {
+                        dst_token: "***".to_string(),
+                        each_n: None,
+                        each_s: None,
+                        ..settings
+                    }
+                );
+            }
             assert_eq!(replication.diagnostics, Diagnostics::default());
         }
 
@@ -621,13 +634,25 @@ pub(crate) mod tests {
                 .unwrap();
             let replication = client.get_replication("test-replication").await.unwrap();
 
-            assert_eq!(
-                replication.settings,
-                ReplicationSettings {
-                    dst_token: "***".to_string(),
-                    ..settings
-                }
-            );
+            if cfg!(feature = "test-api-110") {
+                assert_eq!(
+                    replication.settings,
+                    ReplicationSettings {
+                        dst_token: "***".to_string(),
+                        ..settings
+                    }
+                );
+            } else {
+                assert_eq!(
+                    replication.settings,
+                    ReplicationSettings {
+                        dst_token: "***".to_string(),
+                        each_n: None,
+                        each_s: None,
+                        ..settings
+                    }
+                );
+            }
         }
 
         #[rstest]
@@ -658,6 +683,8 @@ pub(crate) mod tests {
                 entries: vec![],
                 include: Labels::default(),
                 exclude: Labels::default(),
+                each_s: Some(1.0),
+                each_n: Some(1),
             }
         }
     }

diff --git a/src/replication.rs b/src/replication.rs
@@ -31,6 +31,8 @@ impl ReplicationBuilder {
                 entries: vec![],
                 include: Default::default(),
                 exclude: Default::default(),
+                each_s: None,
+                each_n: None,
             },
             http_client,
         }
@@ -107,6 +109,30 @@ impl ReplicationBuilder {
         self
     }
 
+    /// Set the replication each_s setting.
+    ///
+    /// Replicate a record every S seconds if set.
+    ///
+    /// # Arguments
+    ///
+    /// * `each_s` - Replicate a record every S seconds.
+    pub fn each_s(mut self, each_s: f64) -> Self {
+        self.settings.each_s = Some(each_s);
+        self
+    }
+
+    /// Set the replication each_n setting.
+    ///
+    /// Replicate every Nth record if set.
+    ///
+    /// # Arguments
+    ///
+    /// * `each_n` - Replicate every Nth record.
+    pub fn each_n(mut self, each_n: u64) -> Self {
+        self.settings.each_n = Some(each_n);
+        self
+    }
+
     /// Override all the replication settings.
     ///
     /// # Arguments