chore: Upgrade SDK (#41)

Signed-off-by: Vaibhav Rabber <[email protected]>
s2-streamstore · Dec 2, 2024 · 0431ef4 · 0431ef4
1 parent 217c1f5
commit 0431ef4
Show file tree

Hide file tree

Showing 4 changed files with 23 additions and 21 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -28,4 +28,4 @@ tracing-subscriber = { version = "0.3.18", features = ["env-filter"] }
 
 [dependencies.streamstore]
 git = "https://github.com/s2-streamstore/s2-sdk-rust.git"
-rev = "88d6539cbec2165a79f07de4d3a13a9539b547d6"
+rev = "d40de3df1c76ba71361f9e9f5327aee1619763cb"
diff --git a/src/main.rs b/src/main.rs
@@ -13,9 +13,8 @@ use signal_hook::consts::{SIGINT, SIGTERM, SIGTSTP};
 use signal_hook_tokio::Signals;
 use stream::{RecordStream, StreamService};
 use streamstore::{
-    bytesize::ByteSize,
     client::{BasinClient, Client, ClientConfig, S2Endpoints, StreamClient},
-    types::{BasinInfo, BasinName, FencingToken, MeteredSize as _, ReadOutput, StreamInfo},
+    types::{BasinInfo, BasinName, FencingToken, MeteredBytes as _, ReadOutput, StreamInfo},
     HeaderValue,
 };
 use tokio::{
@@ -245,7 +244,7 @@ enum StreamActions {
 
         /// Limit the number of bytes returned.
         #[arg(short = 'b', long)]
-        limit_bytes: Option<ByteSize>,
+        limit_bytes: Option<u64>,
     },
 }
 
@@ -600,7 +599,7 @@ async fn run() -> Result<(), S2CliError> {
                     let mut writer = output.into_writer().await.unwrap();
 
                     let mut start = None;
-                    let mut total_data_len = ByteSize::b(0);
+                    let mut total_data_len = 0;
 
                     loop {
                         select! {
@@ -613,7 +612,7 @@ async fn run() -> Result<(), S2CliError> {
                                         match read_result {
                                             Ok(ReadOutput::Batch(sequenced_record_batch)) => {
                                                 let num_records = sequenced_record_batch.records.len();
-                                                let mut batch_len = ByteSize::b(0);
+                                                let mut batch_len = 0;
 
                                                 let seq_range = match (
                                                     sequenced_record_batch.records.first(),
@@ -624,7 +623,7 @@ async fn run() -> Result<(), S2CliError> {
                                                 };
                                                 for sequenced_record in sequenced_record_batch.records {
                                                     let data = &sequenced_record.body;
-                                                    batch_len += sequenced_record.metered_size();
+                                                    batch_len += sequenced_record.metered_bytes();
 
                                                     writer
                                                         .write_all(data)
@@ -637,7 +636,7 @@ async fn run() -> Result<(), S2CliError> {
                                                 }
                                                 total_data_len += batch_len;
 
-                                                let throughput_mibps = (total_data_len.0 as f64
+                                                let throughput_mibps = (total_data_len as f64
                                                     / start.unwrap().elapsed().as_secs_f64())
                                                     / 1024.0
                                                     / 1024.0;
@@ -683,7 +682,7 @@ async fn run() -> Result<(), S2CliError> {
                         let total_elapsed_time = start.unwrap().elapsed().as_secs_f64();
 
                         let total_throughput_mibps =
-                            (total_data_len.0 as f64 / total_elapsed_time) / 1024.0 / 1024.0;
+                            (total_data_len as f64 / total_elapsed_time) / 1024.0 / 1024.0;
 
                         eprintln!(
                             "{}",

diff --git a/src/stream.rs b/src/stream.rs
@@ -14,7 +14,6 @@ use tokio::io::Lines;
 use tokio_stream::Stream;
 
 use crate::error::{ServiceError, ServiceErrorContext};
-use crate::ByteSize;
 
 pin_project! {
     #[derive(Debug)]
@@ -91,11 +90,11 @@ impl StreamService {
         &self,
         start_seq_num: u64,
         limit_count: Option<u64>,
-        limit_bytes: Option<ByteSize>,
+        limit_bytes: Option<u64>,
     ) -> Result<Streaming<ReadOutput>, ServiceError> {
         let read_session_req = ReadSessionRequest {
             start_seq_num: Some(start_seq_num),
-            limit: match (limit_count, limit_bytes.map(|b| b.as_u64())) {
+            limit: match (limit_count, limit_bytes) {
                 (Some(count), Some(bytes)) => Some(ReadLimit { count, bytes }),
                 (Some(count), None) => Some(ReadLimit { count, bytes: 0 }),
                 (None, Some(bytes)) => Some(ReadLimit { count: 0, bytes }),