Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

chore: Add examples for API #113

Merged
merged 13 commits into from
Dec 16, 2024
32 changes: 32 additions & 0 deletions examples/consumer.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,32 @@
use futures::StreamExt;
use streamstore::{
client::{ClientConfig, StreamClient},
types::{BasinName, ReadSessionRequest},
};
use tokio::select;

#[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 stream = "my-basin";
let stream_client = StreamClient::new(config, basin, stream);

let start_seq_num = 0;
let read_session_request = ReadSessionRequest::new(start_seq_num);
let mut read_stream = stream_client.read_session(read_session_request).await?;

loop {
select! {
next_batch = read_stream.next() => {
let Some(next_batch) = next_batch else { break };
let next_batch = next_batch?;
println!("{next_batch:?}");
}
_ = tokio::signal::ctrl_c() => break,
}
}

Ok(())
}
24 changes: 24 additions & 0 deletions examples/get_latest_record.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,24 @@
use streamstore::{
client::{ClientConfig, StreamClient},
types::{BasinName, ReadLimit, ReadRequest},
};

#[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 stream = "my-basin";
let stream_client = StreamClient::new(config, basin, stream);

let tail = stream_client.check_tail().await?;
let latest_seq_num = tail - 1;
vrongmeal marked this conversation as resolved.
Show resolved Hide resolved

let read_limit = ReadLimit { count: 1, bytes: 0 };
let read_request = ReadRequest::new(latest_seq_num).with_limit(read_limit);
let latest_record = stream_client.read(read_request).await?;

println!("{latest_record:#?}");

Ok(())
}
Loading