Struct S2Basin 

pub struct S2Basin { /* private fields */ }

A basin in an S2 account.

See S2::basin.

Implementations

impl S2Basin

pub fn stream(&self, name: StreamName) -> S2Stream

Get an S2Stream.

Examples found in repository

examples/consumer.rs (line 15)

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(())
}

examples/get_latest_record.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 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(())
}

examples/explicit_trim.rs (line 18)

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(())
}

examples/producer.rs (line 19)

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 list_streams( &self, input: ListStreamsInput, ) -> Result<Page<StreamInfo>, S2Error>

List a page of streams.

See list_all_streams for automatic pagination.

Examples found in repository

examples/list_streams.rs (line 18)

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 s2 = S2::new(S2Config::new(access_token))?;
    let basin = s2.basin(basin_name);

    let input = ListStreamsInput::new().with_prefix("my-".parse()?);
    let page = basin.list_streams(input).await?;
    println!("{page:#?}");

    Ok(())
}

pub fn list_all_streams( &self, input: ListAllStreamsInput, ) -> Streaming<StreamInfo>

List all streams, paginating automatically.

pub async fn create_stream( &self, input: CreateStreamInput, ) -> Result<StreamInfo, S2Error>

Create a stream.

Examples found in repository

examples/create_stream.rs (line 28)

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 basin = s2.basin(basin_name);

    let input = CreateStreamInput::new(stream_name.clone()).with_config(
        StreamConfig::new().with_timestamping(
            TimestampingConfig::new().with_mode(TimestampingMode::ClientRequire),
        ),
    );
    let stream_info = basin.create_stream(input).await?;
    println!("{stream_info:#?}");

    let stream_config = basin.get_stream_config(stream_name).await?;
    println!("{stream_config:#?}");

    Ok(())
}

pub async fn get_stream_config( &self, name: StreamName, ) -> Result<StreamConfig, S2Error>

Get stream configuration.

Examples found in repository

examples/create_stream.rs (line 31)

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 basin = s2.basin(basin_name);

    let input = CreateStreamInput::new(stream_name.clone()).with_config(
        StreamConfig::new().with_timestamping(
            TimestampingConfig::new().with_mode(TimestampingMode::ClientRequire),
        ),
    );
    let stream_info = basin.create_stream(input).await?;
    println!("{stream_info:#?}");

    let stream_config = basin.get_stream_config(stream_name).await?;
    println!("{stream_config:#?}");

    Ok(())
}

pub async fn delete_stream( &self, input: DeleteStreamInput, ) -> Result<(), S2Error>

Delete a stream.

Examples found in repository

examples/delete_stream.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 basin = s2.basin(basin_name);

    let input = DeleteStreamInput::new(stream_name);
    basin.delete_stream(input).await?;
    println!("Deletion requested");

    Ok(())
}

pub async fn reconfigure_stream( &self, input: ReconfigureStreamInput, ) -> Result<StreamConfig, S2Error>

Reconfigure a stream.

Examples found in repository

examples/reconfigure_stream.rs (line 27)

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 basin = s2.basin(basin_name);

    let input = ReconfigureStreamInput::new(
        stream_name,
        StreamReconfiguration::new().with_retention_policy(RetentionPolicy::Age(10 * 24 * 60 * 60)),
    );
    let config = basin.reconfigure_stream(input).await?;
    println!("{config:#?}");

    Ok(())
}

Trait Implementations

impl Clone for S2Basin

fn clone(&self) -> S2Basin

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for S2Basin

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

Formats the value using the given formatter.