Struct S2 

pub struct S2 { /* private fields */ }

An S2 account.

Implementations

impl S2

pub fn new(config: S2Config) -> Result<Self, S2Error>

Create a new S2.

Examples found in repository

examples/list_all_basins.rs (line 13)

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

    let input = ListAllBasinsInput::new();

    let basins: Vec<_> = s2.list_all_basins(input).take(10).try_collect().await?;

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

    Ok(())
}

examples/delete_basin.rs (line 15)

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

    let input = DeleteBasinInput::new(basin_name).with_ignore_not_found(true);
    s2.delete_basin(input).await?;
    println!("Deletion requested");

    Ok(())
}

examples/list_streams.rs (line 14)

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

examples/delete_stream.rs (line 17)

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

examples/consumer.rs (line 14)

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

Additional examples can be found in:

• examples/create_basin.rs
• examples/get_latest_record.rs
• examples/reconfigure_basin.rs
• examples/explicit_trim.rs
• examples/issue_access_token.rs
• examples/create_stream.rs
• examples/producer.rs

pub fn basin(&self, name: BasinName) -> S2Basin

Get an S2Basin.

Examples found in repository

examples/list_streams.rs (line 15)

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

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

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

    Ok(())
}

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/reconfigure_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 = 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(())
}

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

Additional examples can be found in:

• examples/create_stream.rs
• examples/producer.rs

pub async fn list_basins( &self, input: ListBasinsInput, ) -> Result<Page<BasinInfo>, S2Error>

List a page of basins.

See list_all_basins for automatic pagination.

pub fn list_all_basins(&self, input: ListAllBasinsInput) -> Streaming<BasinInfo>

List all basins, paginating automatically.

Examples found in repository

examples/list_all_basins.rs (line 17)

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

    let input = ListAllBasinsInput::new();

    let basins: Vec<_> = s2.list_all_basins(input).take(10).try_collect().await?;

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

    Ok(())
}

pub async fn create_basin( &self, input: CreateBasinInput, ) -> Result<BasinInfo, S2Error>

Create a basin.

Examples found in repository

examples/create_basin.rs (line 22)

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

    let input = CreateBasinInput::new(basin_name.clone()).with_config(
        BasinConfig::new().with_default_stream_config(
            StreamConfig::new().with_retention_policy(RetentionPolicy::Age(10 * 24 * 60 * 60)),
        ),
    );
    let basin_info = s2.create_basin(input).await?;
    println!("{basin_info:#?}");

    let basin_config = s2.get_basin_config(basin_name).await?;
    println!("{basin_config:#?}");

    Ok(())
}

pub async fn get_basin_config( &self, name: BasinName, ) -> Result<BasinConfig, S2Error>

Get basin configuration.

Examples found in repository

examples/create_basin.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 config = S2Config::new(access_token);
    let s2 = S2::new(config)?;

    let input = CreateBasinInput::new(basin_name.clone()).with_config(
        BasinConfig::new().with_default_stream_config(
            StreamConfig::new().with_retention_policy(RetentionPolicy::Age(10 * 24 * 60 * 60)),
        ),
    );
    let basin_info = s2.create_basin(input).await?;
    println!("{basin_info:#?}");

    let basin_config = s2.get_basin_config(basin_name).await?;
    println!("{basin_config:#?}");

    Ok(())
}

pub async fn delete_basin(&self, input: DeleteBasinInput) -> Result<(), S2Error>

Delete a basin.

Examples found in repository

examples/delete_basin.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 config = S2Config::new(access_token);
    let s2 = S2::new(config)?;

    let input = DeleteBasinInput::new(basin_name).with_ignore_not_found(true);
    s2.delete_basin(input).await?;
    println!("Deletion requested");

    Ok(())
}

pub async fn reconfigure_basin( &self, input: ReconfigureBasinInput, ) -> Result<BasinConfig, S2Error>

Reconfigure a basin.

Examples found in repository

examples/reconfigure_basin.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 config = S2Config::new(access_token);
    let s2 = S2::new(config)?;

    let input = ReconfigureBasinInput::new(
        basin_name,
        BasinReconfiguration::new()
            .with_default_stream_config(
                StreamReconfiguration::new()
                    .with_storage_class(StorageClass::Standard)
                    .with_timestamping(TimestampingReconfiguration::new().with_uncapped(true)),
            )
            .with_create_stream_on_read(true),
    );
    let config = s2.reconfigure_basin(input).await?;
    println!("{config:#?}");

    Ok(())
}

pub async fn list_access_tokens( &self, input: ListAccessTokensInput, ) -> Result<Page<AccessTokenInfo>, S2Error>

List a page of access tokens.

See list_all_access_tokens for automatic pagination.

pub fn list_all_access_tokens( &self, input: ListAllAccessTokensInput, ) -> Streaming<AccessTokenInfo>

List all access tokens, paginating automatically.

pub async fn issue_access_token( &self, input: IssueAccessTokenInput, ) -> Result<String, S2Error>

Issue an access token.

Examples found in repository

examples/issue_access_token.rs (line 29)

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

    let input = IssueAccessTokenInput::new(
        "ro-token".parse()?,
        AccessTokenScopeInput::from_op_group_perms(
            OperationGroupPermissions::new().with_account(ReadWritePermissions::read_only()),
        )
        .with_ops([Operation::CreateStream])
        .with_streams(StreamMatcher::Prefix("audit".parse()?))
        .with_basins(BasinMatcher::Exact(basin_name)),
    );
    let issued_token = s2.issue_access_token(input).await?;
    println!("Issued access token: {issued_token}");

    Ok(())
}

pub async fn revoke_access_token( &self, id: AccessTokenId, ) -> Result<(), S2Error>

Revoke an access token.

pub async fn get_account_metrics( &self, input: GetAccountMetricsInput, ) -> Result<Vec<Metric>, S2Error>

Get account metrics.

pub async fn get_basin_metrics( &self, input: GetBasinMetricsInput, ) -> Result<Vec<Metric>, S2Error>

Get basin metrics.

pub async fn get_stream_metrics( &self, input: GetStreamMetricsInput, ) -> Result<Vec<Metric>, S2Error>

Get stream metrics.

Trait Implementations

impl Clone for S2

fn clone(&self) -> S2

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for S2

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

Formats the value using the given formatter.