reconfigure stream

Signed-off-by: Vaibhav Rabber <[email protected]>

examples/create_basin.rs

@@ -22,11 +22,13 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         default_stream_config: Some(default_stream_config),
     };
 
-    let create_basin_request = CreateBasinRequest::new(basin).with_config(basin_config);
+    let create_basin_request = CreateBasinRequest::new(basin.clone()).with_config(basin_config);
 
     let created_basin = client.create_basin(create_basin_request).await?;
-
     println!("{created_basin:#?}");
 
+    let basin_config = client.get_basin_config(basin).await?;
+    println!("{basin_config:#?}");
+
     Ok(())
 }

examples/create_stream.rs

@@ -19,8 +19,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let create_stream_request = CreateStreamRequest::new(stream).with_config(stream_config);
 
     let created_stream = basin_client.create_stream(create_stream_request).await?;
-
     println!("{created_stream:#?}");
 
+    let stream_config = basin_client.get_stream_config(stream).await?;
+    println!("{stream_config:#?}");
+
     Ok(())
 }

examples/reconfigure_stream.rs

@@ -0,0 +1,34 @@
+use std::time::Duration;
+
+use streamstore::{
+    client::{BasinClient, ClientConfig},
+    types::{BasinName, ReconfigureStreamRequest, RetentionPolicy, StreamConfig},
+};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let token = std::env::var("S2_AUTH_TOKEN")?;
+    let config = ClientConfig::new(token);
+    let basin: BasinName = "my-basin".parse()?;
+    let basin_client = BasinClient::new(config, basin);
+
+    let stream = "my-stream";
+
+    let stream_config_updates = StreamConfig::new().with_retention_policy(RetentionPolicy::Age(
+        // Change to retention policy to 1 day
+        Duration::from_secs(24 * 60 * 60),
+    ));
+
+    let reconfigure_stream_request = ReconfigureStreamRequest::new(stream)
+        .with_config(stream_config_updates)
+        // Field mask specifies which fields to update.
+        .with_mask(vec!["retention_policy".to_string()]);
+
+    let updated_stream = basin_client
+        .reconfigure_stream(reconfigure_stream_request)
+        .await?;
+
+    println!("{updated_stream:#?}");
+
+    Ok(())
+}