Struct S2Stream 

pub struct S2Stream { /* private fields */ }

A stream in an S2 basin.

See S2Basin::stream.

Implementations

impl S2Stream

pub async fn check_tail(&self) -> Result<StreamPosition, S2Error>

Check tail position.

Examples found in repository

examples/explicit_trim.rs (line 20)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let access_token =
        std::env::var("S2_ACCESS_TOKEN").map_err(|_| "S2_ACCESS_TOKEN env var not set")?;
    let basin_name: BasinName = std::env::var("S2_BASIN")
        .map_err(|_| "S2_BASIN env var not set")?
        .parse()?;
    let stream_name: StreamName = std::env::var("S2_STREAM")
        .map_err(|_| "S2_STREAM env var not set")?
        .parse()?;

    let s2 = S2::new(S2Config::new(access_token))?;
    let stream = s2.basin(basin_name).stream(stream_name);

    let tail = stream.check_tail().await?;
    if tail.seq_num == 0 {
        println!("Empty stream");
        return Ok(());
    }

    let input = AppendInput::new(AppendRecordBatch::try_from_iter([CommandRecord::trim(
        tail.seq_num - 1,
    )
    .into()])?);
    stream.append(input).await?;
    println!("Trim requested");

    Ok(())
}

pub async fn append(&self, input: AppendInput) -> Result<AppendAck, S2Error>

Append records.

Examples found in repository

examples/explicit_trim.rs (line 30)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let access_token =
        std::env::var("S2_ACCESS_TOKEN").map_err(|_| "S2_ACCESS_TOKEN env var not set")?;
    let basin_name: BasinName = std::env::var("S2_BASIN")
        .map_err(|_| "S2_BASIN env var not set")?
        .parse()?;
    let stream_name: StreamName = std::env::var("S2_STREAM")
        .map_err(|_| "S2_STREAM env var not set")?
        .parse()?;

    let s2 = S2::new(S2Config::new(access_token))?;
    let stream = s2.basin(basin_name).stream(stream_name);

    let tail = stream.check_tail().await?;
    if tail.seq_num == 0 {
        println!("Empty stream");
        return Ok(());
    }

    let input = AppendInput::new(AppendRecordBatch::try_from_iter([CommandRecord::trim(
        tail.seq_num - 1,
    )
    .into()])?);
    stream.append(input).await?;
    println!("Trim requested");

    Ok(())
}

pub async fn read(&self, input: ReadInput) -> Result<ReadBatch, S2Error>

Read records.

Examples found in repository

examples/get_latest_record.rs (line 25)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let access_token =
        std::env::var("S2_ACCESS_TOKEN").map_err(|_| "S2_ACCESS_TOKEN env var not set")?;
    let basin_name: BasinName = std::env::var("S2_BASIN")
        .map_err(|_| "S2_BASIN env var not set")?
        .parse()?;
    let stream_name: StreamName = std::env::var("S2_STREAM")
        .map_err(|_| "S2_STREAM env var not set")?
        .parse()?;

    let s2 = S2::new(S2Config::new(access_token))?;
    let stream = s2.basin(basin_name).stream(stream_name);

    let input = ReadInput::new()
        .with_start(ReadStart::new().with_from(ReadFrom::TailOffset(1)))
        .with_stop(ReadStop::new().with_limits(ReadLimits::new().with_count(1)));
    let batch = stream.read(input).await?;
    println!("{batch:#?}");

    Ok(())
}

pub fn append_session(&self, config: AppendSessionConfig) -> AppendSession

Create an append session for submitting AppendInputs.

pub fn producer(&self, config: ProducerConfig) -> Producer

Create a producer for submitting individual AppendRecords.

Examples found in repository

examples/producer.rs (line 21)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let access_token =
        std::env::var("S2_ACCESS_TOKEN").map_err(|_| "S2_ACCESS_TOKEN env var not set")?;
    let basin_name: BasinName = std::env::var("S2_BASIN")
        .map_err(|_| "S2_BASIN env var not set")?
        .parse()?;
    let stream_name: StreamName = std::env::var("S2_STREAM")
        .map_err(|_| "S2_STREAM env var not set")?
        .parse()?;

    let s2 = S2::new(S2Config::new(access_token))?;
    let stream = s2.basin(basin_name).stream(stream_name);

    let producer = stream.producer(ProducerConfig::new());

    let ticket1 = producer.submit(AppendRecord::new("lorem")?).await?;
    let ticket2 = producer.submit(AppendRecord::new("ipsum")?).await?;

    let ack1 = ticket1.await?;
    let ack2 = ticket2.await?;
    println!("Record 1 seq_num: {}", ack1.seq_num);
    println!("Record 2 seq_num: {}", ack2.seq_num);

    producer.close().await?;

    Ok(())
}

pub async fn read_session( &self, input: ReadInput, ) -> Result<Streaming<ReadBatch>, S2Error>

Create a read session.

Examples found in repository

examples/consumer.rs (line 18)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let access_token = std::env::var("S2_ACCESS_TOKEN")?;
    let basin_name: BasinName = std::env::var("S2_BASIN")?.parse()?;
    let stream_name: StreamName = std::env::var("S2_STREAM")?.parse()?;

    let s2 = S2::new(S2Config::new(access_token))?;
    let stream = s2.basin(basin_name).stream(stream_name);

    let input = ReadInput::new();
    let mut batches = stream.read_session(input).await?;
    loop {
        select! {
            batch = batches.next() => {
                let Some(batch) = batch else { break };
                let batch = batch?;
                println!("{batch:?}");
            }
            _ = tokio::signal::ctrl_c() => break,
        }
    }

    Ok(())
}

Trait Implementations

impl Clone for S2Stream

fn clone(&self) -> S2Stream

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for S2Stream

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.