cleanup

Author: infiniteregrets

src/account.rs

@@ -5,9 +5,9 @@ use s2::{
     client::Client,
     types::{
         AccessTokenId, AccessTokenInfo, BasinConfig, BasinInfo, BasinName, CreateBasinRequest,
-        DeleteBasinRequest, ListAccessTokensRequest, ListAccessTokensResponse, ListBasinsRequest,
-        ListBasinsResponse, Operation, PermittedOperationGroups, ReconfigureBasinRequest,
-        ResourceSet, StreamConfig,
+        DeleteBasinRequest, DeleteOnEmpty, ListAccessTokensRequest, ListAccessTokensResponse,
+        ListBasinsRequest, ListBasinsResponse, Operation, PermittedOperationGroups,
+        ReconfigureBasinRequest, ResourceSet, StreamConfig,
     },
 };
 
@@ -77,21 +77,30 @@ impl AccountService {
     pub async fn create_basin(
         &self,
         basin: BasinName,
-        storage_class: Option<crate::types::StorageClass>,
-        retention_policy: Option<crate::types::RetentionPolicy>,
+        configured_stream_config: crate::types::StreamConfig,
         create_stream_on_append: bool,
         create_stream_on_read: bool,
     ) -> Result<BasinInfo, ServiceError> {
         let mut stream_config = StreamConfig::new();
 
-        if let Some(storage_class) = storage_class {
+        if let Some(storage_class) = configured_stream_config.storage_class {
             stream_config = stream_config.with_storage_class(storage_class.into());
         }
 
-        if let Some(retention_policy) = retention_policy {
+        if let Some(retention_policy) = configured_stream_config.retention_policy {
             stream_config = stream_config.with_retention_policy(retention_policy.into());
         }
 
+        if let Some(timestamping) = configured_stream_config.timestamping {
+            stream_config = stream_config.with_timestamping(timestamping.into());
+        }
+
+        if let Some(delete_on_empty_min_age) = configured_stream_config.delete_on_empty_min_age {
+            stream_config = stream_config.with_delete_on_empty(DeleteOnEmpty {
+                min_age: *delete_on_empty_min_age,
+            });
+        }
+
         let basin_config = BasinConfig {
             default_stream_config: Some(stream_config),
             create_stream_on_append,

src/main.rs

@@ -38,7 +38,7 @@ use tokio_stream::{
 use tracing::trace;
 use tracing_subscriber::{fmt::format::FmtSpan, layer::SubscriberExt, util::SubscriberInitExt};
 use types::{
-    AccessTokenInfo, BasinConfig, DeleteOnEmpty, Operation, PermittedOperationGroups, ResourceSet,
+    AccessTokenInfo, BasinConfig, Operation, PermittedOperationGroups, ResourceSet,
     RetentionPolicy, S2BasinAndMaybeStreamUri, S2BasinAndStreamUri, S2BasinUri, StorageClass,
     StreamConfig, TimestampingConfig, TimestampingMode,
 };
@@ -187,7 +187,7 @@ enum Commands {
 
         /// Delete-on-empty configuration for the default stream config.
         #[arg(long)]
-        delete_on_empty_min_age: Option<DeleteOnEmpty>,
+        delete_on_empty_min_age: Option<humantime::Duration>,
     },
 
     /// Issue an access token.
@@ -329,7 +329,7 @@ enum Commands {
 
         /// Delete-on-empty configuration for the stream.
         #[arg(long)]
-        delete_on_empty_min_age: Option<DeleteOnEmpty>,
+        delete_on_empty_min_age: Option<humantime::Duration>,
     },
 
     /// Get the next sequence number that will be assigned by a stream.
@@ -653,7 +653,7 @@ fn build_basin_reconfig(
     timestamping_uncapped: Option<&bool>,
     create_stream_on_append: Option<&bool>,
     create_stream_on_read: Option<&bool>,
-    delete_on_empty_min_age: Option<&DeleteOnEmpty>,
+    delete_on_empty_min_age: Option<&humantime::Duration>,
 ) -> (Option<StreamConfig>, Vec<String>) {
     let mut mask = Vec::new();
     let has_stream_args = storage_class.is_some()
@@ -712,7 +712,7 @@ fn build_stream_reconfig(
     retention_policy: Option<&RetentionPolicy>,
     timestamping_mode: Option<&TimestampingMode>,
     timestamping_uncapped: Option<&bool>,
-    delete_on_empty_min_age: Option<&DeleteOnEmpty>,
+    delete_on_empty_min_age: Option<&humantime::Duration>,
 ) -> (StreamConfig, Vec<String>) {
     let mut mask = Vec::new();
 
@@ -965,19 +965,11 @@ async fn run() -> Result<(), S2CliError> {
             let cfg = config::load_config(&config_path)?;
             let client_config = client_config(cfg.access_token)?;
             let account_service = AccountService::new(Client::new(client_config));
-            let (storage_class, retention_policy) = match &config.default_stream_config {
-                Some(config) => {
-                    let storage_class = config.storage_class.clone();
-                    let retention_policy = config.retention_policy.clone();
-                    (storage_class, retention_policy)
-                }
-                None => (None, None),
-            };
+
             let BasinInfo { state, .. } = account_service
                 .create_basin(
                     basin.into(),
-                    storage_class,
-                    retention_policy,
+                    config.default_stream_config.unwrap_or_default(),
                     config.create_stream_on_append.unwrap_or_default(),
                     config.create_stream_on_read.unwrap_or_default(),
                 )

src/types.rs

@@ -125,7 +125,7 @@ pub struct BasinConfig {
     pub create_stream_on_read: Option<bool>,
 }
 
-#[derive(Parser, Debug, Clone, Serialize)]
+#[derive(Parser, Debug, Default, Clone, Serialize)]
 pub struct StreamConfig {
     #[arg(long)]
     /// Storage class for a stream.
@@ -138,7 +138,11 @@ pub struct StreamConfig {
     pub timestamping: Option<TimestampingConfig>,
     #[arg(long, help("Example: 1d, 1w, 1y"))]
     /// Delete-on-empty configuration.
-    pub delete_on_empty_min_age: Option<DeleteOnEmpty>,
+    #[serde(
+        skip_serializing_if = "Option::is_none",
+        serialize_with = "serialize_humantime_duration"
+    )]
+    pub delete_on_empty_min_age: Option<humantime::Duration>,
 }
 
 #[derive(ValueEnum, Debug, Clone, Serialize)]
@@ -238,7 +242,7 @@ impl From<StreamConfig> for s2::types::StreamConfig {
 
         let delete_on_empty = config
             .delete_on_empty_min_age
-            .map(s2::types::DeleteOnEmpty::from);
+            .map(|age| s2::types::DeleteOnEmpty { min_age: *age });
 
         let mut stream_config = s2::types::StreamConfig::new();
         if let Some(storage_class) = storage_class {
@@ -345,7 +349,9 @@ impl From<s2::types::StreamConfig> for StreamConfig {
             storage_class: config.storage_class.map(Into::into),
             retention_policy: config.retention_policy.map(Into::into),
             timestamping: config.timestamping.map(Into::into),
-            delete_on_empty_min_age: config.delete_on_empty.map(Into::into),
+            delete_on_empty_min_age: config
+                .delete_on_empty
+                .map(|age| humantime::Duration::from(age.min_age)),
         }
     }
 }
@@ -716,6 +722,19 @@ impl FromStr for Operation {
     }
 }
 
+fn serialize_humantime_duration<S>(
+    duration: &Option<humantime::Duration>,
+    serializer: S,
+) -> Result<S::Ok, S::Error>
+where
+    S: serde::Serializer,
+{
+    match duration {
+        Some(d) => serializer.serialize_str(&format!("{d}")),
+        None => serializer.serialize_none(),
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use crate::{error::S2UriParseError, types::S2BasinAndStreamUri};