feat: display throughput for read session

Cargo.toml

@@ -16,7 +16,7 @@ pin-project-lite = "0.2"
 pin-utils = "0.1.0"
 serde = { version = "1.0.214", features = ["derive"] }
 serde_json = "1.0.132"
-streamstore = { git = "https://github.com/s2-streamstore/s2-sdk-rust.git", rev = "54e36154a1aaf9ead2512bb446677cb7e94a4dc2" }
+streamstore = { git = "https://github.com/s2-streamstore/s2-sdk-rust.git", rev = "ead36a9bf8177dab744ac158b625ffec9ea6c50a" }
 thiserror = "1.0.67"
 tokio = { version = "*", features = ["full"] }
 tokio-stream = { version = "0.1.16", features = ["io-util"] }

src/account.rs

@@ -1,15 +1,13 @@
 use streamstore::{
     client::Client,
-    service_error::{
-        CreateBasinError, DeleteBasinError, GetBasinConfigError, ReconfigureBasinError,
-        ServiceError,
-    },
     types::{
         BasinConfig, BasinMetadata, CreateBasinRequest, DeleteBasinRequest, ListBasinsRequest,
         ListBasinsResponse, ReconfigureBasinRequest, StreamConfig,
     },
 };
 
+use crate::error::s2_status;
+
 pub struct AccountService {
     client: Client,
 }
@@ -19,17 +17,17 @@ pub enum AccountServiceError {
     #[error("Failed to list basins: {0}")]
     ListBasins(String),
 
-    #[error("Failed to create basin")]
-    CreateBasin(#[from] ServiceError<CreateBasinError>),
+    #[error("Failed to create basin: {0}")]
+    CreateBasin(String),
 
-    #[error("Failed to delete basin")]
-    DeleteBasin(#[from] ServiceError<DeleteBasinError>),
+    #[error("Failed to delete basin: {0}")]
+    DeleteBasin(String),
 
-    #[error("Failed to get basin config")]
-    GetBasinConfig(#[from] ServiceError<GetBasinConfigError>),
+    #[error("Failed to get basin config: {0}")]
+    GetBasinConfig(String),
 
-    #[error("Failed to reconfigure basin")]
-    ReconfigureBasin(#[from] ServiceError<ReconfigureBasinError>),
+    #[error("Failed to reconfigure basin: {0}")]
+    ReconfigureBasin(String),
 }
 
 impl AccountService {
@@ -51,7 +49,7 @@ impl AccountService {
         self.client
             .list_basins(list_basins_req)
             .await
-            .map_err(|e| AccountServiceError::ListBasins(e.to_string()))
+            .map_err(|e| AccountServiceError::ListBasins(s2_status(&e)))
     }
 
     pub async fn create_basin(
@@ -76,20 +74,27 @@ impl AccountService {
         self.client
             .create_basin(create_basin_req)
             .await
-            .map_err(AccountServiceError::CreateBasin)
+            .map_err(|e| AccountServiceError::CreateBasin(s2_status(&e)))
     }
 
     pub async fn delete_basin(&self, basin: String) -> Result<(), AccountServiceError> {
         let delete_basin_req = DeleteBasinRequest::new(basin);
-        self.client.delete_basin(delete_basin_req).await?;
+        self.client
+            .delete_basin(delete_basin_req)
+            .await
+            .map_err(|e| AccountServiceError::DeleteBasin(s2_status(&e)))?;
         Ok(())
     }
 
     pub async fn get_basin_config(
         &self,
         basin: String,
     ) -> Result<BasinConfig, AccountServiceError> {
-        Ok(self.client.get_basin_config(basin).await?)
+        Ok(self
+            .client
+            .get_basin_config(basin)
+            .await
+            .map_err(|e| AccountServiceError::GetBasinConfig(s2_status(&e)))?)
     }
 
     pub async fn reconfigure_basin(
@@ -101,7 +106,10 @@ impl AccountService {
         let reconfigure_basin_req = ReconfigureBasinRequest::new(basin)
             .with_config(basin_config)
             .with_mask(mask);
-        self.client.reconfigure_basin(reconfigure_basin_req).await?;
+        self.client
+            .reconfigure_basin(reconfigure_basin_req)
+            .await
+            .map_err(|e| AccountServiceError::ReconfigureBasin(s2_status(&e)))?;
         Ok(())
     }
 }