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/docs_overview.rs (line 9)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    // ANCHOR: create-client
    use s2_sdk::{S2, types::S2Config};

    let client = S2::new(S2Config::new(std::env::var("S2_ACCESS_TOKEN")?))?;

    let basin = client.basin("my-basin".parse()?);
    let stream = basin.stream("my-stream".parse()?);
    // ANCHOR_END: create-client

    println!("Created client for stream: {:?}", stream);
    Ok(())
}

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

Additional examples can be found in:

• examples/reconfigure_stream.rs
• 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
• examples/docs_configuration.rs
• examples/docs_account_and_basins.rs
• examples/docs_streams.rs

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

Get an S2Basin.

Examples found in repository

examples/docs_overview.rs (line 11)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    // ANCHOR: create-client
    use s2_sdk::{S2, types::S2Config};

    let client = S2::new(S2Config::new(std::env::var("S2_ACCESS_TOKEN")?))?;

    let basin = client.basin("my-basin".parse()?);
    let stream = basin.stream("my-stream".parse()?);
    // ANCHOR_END: create-client

    println!("Created client for stream: {:?}", stream);
    Ok(())
}

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

Additional examples can be found in:

• examples/explicit_trim.rs
• examples/create_stream.rs
• examples/producer.rs
• examples/docs_account_and_basins.rs
• examples/docs_streams.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.

Examples found in repository

examples/docs_account_and_basins.rs (line 25)

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

    let client = S2::new(S2Config::new(token))?;

    // ANCHOR: basin-operations
    // List basins
    let basins = client.list_basins(ListBasinsInput::new()).await?;

    // Create a basin
    client
        .create_basin(CreateBasinInput::new("my-events".parse()?))
        .await?;

    // Get configuration
    let config = client.get_basin_config("my-events".parse()?).await?;

    // Delete
    client
        .delete_basin(DeleteBasinInput::new("my-events".parse()?))
        .await?;
    // ANCHOR_END: basin-operations
    println!("Basins: {:?}, config: {:?}", basins, config);

    let basin = client.basin(basin_name);

    // ANCHOR: stream-operations
    // List streams
    let streams = basin
        .list_streams(ListStreamsInput::new().with_prefix("user-".parse()?))
        .await?;

    // Create a stream
    basin
        .create_stream(CreateStreamInput::new("user-actions".parse()?))
        .await?;

    // Get configuration
    let config = basin.get_stream_config("user-actions".parse()?).await?;

    // Delete
    basin
        .delete_stream(DeleteStreamInput::new("user-actions".parse()?))
        .await?;
    // ANCHOR_END: stream-operations
    println!("Streams: {:?}, config: {:?}", streams, config);

    // ANCHOR: access-token-basic
    // List tokens (returns metadata, not the secret)
    let tokens = client.list_access_tokens(Default::default()).await?;

    // Issue a token scoped to streams under "users/1234/"
    let result = client
        .issue_access_token(
            IssueAccessTokenInput::new(
                "user-1234-rw-token".parse()?,
                AccessTokenScopeInput::from_op_group_perms(
                    OperationGroupPermissions::new()
                        .with_stream(ReadWritePermissions::read_write()),
                )
                .with_basins(BasinMatcher::Prefix("".parse()?)) // all basins
                .with_streams(StreamMatcher::Prefix("users/1234/".parse()?)),
            )
            .with_expires_at("2027-01-01T00:00:00Z".parse()?),
        )
        .await?;

    // Revoke a token
    client
        .revoke_access_token("user-1234-rw-token".parse()?)
        .await?;
    // ANCHOR_END: access-token-basic
    println!("Tokens: {:?}, issued: {:?}", tokens, result);

    // ANCHOR: access-token-restricted
    client
        .issue_access_token(IssueAccessTokenInput::new(
            "restricted-token".parse()?,
            AccessTokenScopeInput::from_op_group_perms(
                OperationGroupPermissions::new().with_stream(ReadWritePermissions::read_only()),
            )
            .with_basins(BasinMatcher::Exact("production".parse()?))
            .with_streams(StreamMatcher::Prefix("logs/".parse()?)),
        ))
        .await?;
    // ANCHOR_END: access-token-restricted

    // Pagination example - not executed by default
    if false {
        // ANCHOR: pagination
        // Iterate through all streams with automatic pagination
        let mut stream = basin.list_all_streams(ListAllStreamsInput::new());
        while let Some(info) = stream.next().await {
            let info = info?;
            println!("{}", info.name);
        }
        // ANCHOR_END: pagination
    }

    Ok(())
}

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

examples/docs_account_and_basins.rs (line 29)

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

    let client = S2::new(S2Config::new(token))?;

    // ANCHOR: basin-operations
    // List basins
    let basins = client.list_basins(ListBasinsInput::new()).await?;

    // Create a basin
    client
        .create_basin(CreateBasinInput::new("my-events".parse()?))
        .await?;

    // Get configuration
    let config = client.get_basin_config("my-events".parse()?).await?;

    // Delete
    client
        .delete_basin(DeleteBasinInput::new("my-events".parse()?))
        .await?;
    // ANCHOR_END: basin-operations
    println!("Basins: {:?}, config: {:?}", basins, config);

    let basin = client.basin(basin_name);

    // ANCHOR: stream-operations
    // List streams
    let streams = basin
        .list_streams(ListStreamsInput::new().with_prefix("user-".parse()?))
        .await?;

    // Create a stream
    basin
        .create_stream(CreateStreamInput::new("user-actions".parse()?))
        .await?;

    // Get configuration
    let config = basin.get_stream_config("user-actions".parse()?).await?;

    // Delete
    basin
        .delete_stream(DeleteStreamInput::new("user-actions".parse()?))
        .await?;
    // ANCHOR_END: stream-operations
    println!("Streams: {:?}, config: {:?}", streams, config);

    // ANCHOR: access-token-basic
    // List tokens (returns metadata, not the secret)
    let tokens = client.list_access_tokens(Default::default()).await?;

    // Issue a token scoped to streams under "users/1234/"
    let result = client
        .issue_access_token(
            IssueAccessTokenInput::new(
                "user-1234-rw-token".parse()?,
                AccessTokenScopeInput::from_op_group_perms(
                    OperationGroupPermissions::new()
                        .with_stream(ReadWritePermissions::read_write()),
                )
                .with_basins(BasinMatcher::Prefix("".parse()?)) // all basins
                .with_streams(StreamMatcher::Prefix("users/1234/".parse()?)),
            )
            .with_expires_at("2027-01-01T00:00:00Z".parse()?),
        )
        .await?;

    // Revoke a token
    client
        .revoke_access_token("user-1234-rw-token".parse()?)
        .await?;
    // ANCHOR_END: access-token-basic
    println!("Tokens: {:?}, issued: {:?}", tokens, result);

    // ANCHOR: access-token-restricted
    client
        .issue_access_token(IssueAccessTokenInput::new(
            "restricted-token".parse()?,
            AccessTokenScopeInput::from_op_group_perms(
                OperationGroupPermissions::new().with_stream(ReadWritePermissions::read_only()),
            )
            .with_basins(BasinMatcher::Exact("production".parse()?))
            .with_streams(StreamMatcher::Prefix("logs/".parse()?)),
        ))
        .await?;
    // ANCHOR_END: access-token-restricted

    // Pagination example - not executed by default
    if false {
        // ANCHOR: pagination
        // Iterate through all streams with automatic pagination
        let mut stream = basin.list_all_streams(ListAllStreamsInput::new());
        while let Some(info) = stream.next().await {
            let info = info?;
            println!("{}", info.name);
        }
        // ANCHOR_END: pagination
    }

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

examples/docs_account_and_basins.rs (line 33)

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

    let client = S2::new(S2Config::new(token))?;

    // ANCHOR: basin-operations
    // List basins
    let basins = client.list_basins(ListBasinsInput::new()).await?;

    // Create a basin
    client
        .create_basin(CreateBasinInput::new("my-events".parse()?))
        .await?;

    // Get configuration
    let config = client.get_basin_config("my-events".parse()?).await?;

    // Delete
    client
        .delete_basin(DeleteBasinInput::new("my-events".parse()?))
        .await?;
    // ANCHOR_END: basin-operations
    println!("Basins: {:?}, config: {:?}", basins, config);

    let basin = client.basin(basin_name);

    // ANCHOR: stream-operations
    // List streams
    let streams = basin
        .list_streams(ListStreamsInput::new().with_prefix("user-".parse()?))
        .await?;

    // Create a stream
    basin
        .create_stream(CreateStreamInput::new("user-actions".parse()?))
        .await?;

    // Get configuration
    let config = basin.get_stream_config("user-actions".parse()?).await?;

    // Delete
    basin
        .delete_stream(DeleteStreamInput::new("user-actions".parse()?))
        .await?;
    // ANCHOR_END: stream-operations
    println!("Streams: {:?}, config: {:?}", streams, config);

    // ANCHOR: access-token-basic
    // List tokens (returns metadata, not the secret)
    let tokens = client.list_access_tokens(Default::default()).await?;

    // Issue a token scoped to streams under "users/1234/"
    let result = client
        .issue_access_token(
            IssueAccessTokenInput::new(
                "user-1234-rw-token".parse()?,
                AccessTokenScopeInput::from_op_group_perms(
                    OperationGroupPermissions::new()
                        .with_stream(ReadWritePermissions::read_write()),
                )
                .with_basins(BasinMatcher::Prefix("".parse()?)) // all basins
                .with_streams(StreamMatcher::Prefix("users/1234/".parse()?)),
            )
            .with_expires_at("2027-01-01T00:00:00Z".parse()?),
        )
        .await?;

    // Revoke a token
    client
        .revoke_access_token("user-1234-rw-token".parse()?)
        .await?;
    // ANCHOR_END: access-token-basic
    println!("Tokens: {:?}, issued: {:?}", tokens, result);

    // ANCHOR: access-token-restricted
    client
        .issue_access_token(IssueAccessTokenInput::new(
            "restricted-token".parse()?,
            AccessTokenScopeInput::from_op_group_perms(
                OperationGroupPermissions::new().with_stream(ReadWritePermissions::read_only()),
            )
            .with_basins(BasinMatcher::Exact("production".parse()?))
            .with_streams(StreamMatcher::Prefix("logs/".parse()?)),
        ))
        .await?;
    // ANCHOR_END: access-token-restricted

    // Pagination example - not executed by default
    if false {
        // ANCHOR: pagination
        // Iterate through all streams with automatic pagination
        let mut stream = basin.list_all_streams(ListAllStreamsInput::new());
        while let Some(info) = stream.next().await {
            let info = info?;
            println!("{}", info.name);
        }
        // ANCHOR_END: pagination
    }

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

examples/docs_account_and_basins.rs (line 37)

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

    let client = S2::new(S2Config::new(token))?;

    // ANCHOR: basin-operations
    // List basins
    let basins = client.list_basins(ListBasinsInput::new()).await?;

    // Create a basin
    client
        .create_basin(CreateBasinInput::new("my-events".parse()?))
        .await?;

    // Get configuration
    let config = client.get_basin_config("my-events".parse()?).await?;

    // Delete
    client
        .delete_basin(DeleteBasinInput::new("my-events".parse()?))
        .await?;
    // ANCHOR_END: basin-operations
    println!("Basins: {:?}, config: {:?}", basins, config);

    let basin = client.basin(basin_name);

    // ANCHOR: stream-operations
    // List streams
    let streams = basin
        .list_streams(ListStreamsInput::new().with_prefix("user-".parse()?))
        .await?;

    // Create a stream
    basin
        .create_stream(CreateStreamInput::new("user-actions".parse()?))
        .await?;

    // Get configuration
    let config = basin.get_stream_config("user-actions".parse()?).await?;

    // Delete
    basin
        .delete_stream(DeleteStreamInput::new("user-actions".parse()?))
        .await?;
    // ANCHOR_END: stream-operations
    println!("Streams: {:?}, config: {:?}", streams, config);

    // ANCHOR: access-token-basic
    // List tokens (returns metadata, not the secret)
    let tokens = client.list_access_tokens(Default::default()).await?;

    // Issue a token scoped to streams under "users/1234/"
    let result = client
        .issue_access_token(
            IssueAccessTokenInput::new(
                "user-1234-rw-token".parse()?,
                AccessTokenScopeInput::from_op_group_perms(
                    OperationGroupPermissions::new()
                        .with_stream(ReadWritePermissions::read_write()),
                )
                .with_basins(BasinMatcher::Prefix("".parse()?)) // all basins
                .with_streams(StreamMatcher::Prefix("users/1234/".parse()?)),
            )
            .with_expires_at("2027-01-01T00:00:00Z".parse()?),
        )
        .await?;

    // Revoke a token
    client
        .revoke_access_token("user-1234-rw-token".parse()?)
        .await?;
    // ANCHOR_END: access-token-basic
    println!("Tokens: {:?}, issued: {:?}", tokens, result);

    // ANCHOR: access-token-restricted
    client
        .issue_access_token(IssueAccessTokenInput::new(
            "restricted-token".parse()?,
            AccessTokenScopeInput::from_op_group_perms(
                OperationGroupPermissions::new().with_stream(ReadWritePermissions::read_only()),
            )
            .with_basins(BasinMatcher::Exact("production".parse()?))
            .with_streams(StreamMatcher::Prefix("logs/".parse()?)),
        ))
        .await?;
    // ANCHOR_END: access-token-restricted

    // Pagination example - not executed by default
    if false {
        // ANCHOR: pagination
        // Iterate through all streams with automatic pagination
        let mut stream = basin.list_all_streams(ListAllStreamsInput::new());
        while let Some(info) = stream.next().await {
            let info = info?;
            println!("{}", info.name);
        }
        // ANCHOR_END: pagination
    }

    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.

Examples found in repository

examples/docs_account_and_basins.rs (line 67)

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

    let client = S2::new(S2Config::new(token))?;

    // ANCHOR: basin-operations
    // List basins
    let basins = client.list_basins(ListBasinsInput::new()).await?;

    // Create a basin
    client
        .create_basin(CreateBasinInput::new("my-events".parse()?))
        .await?;

    // Get configuration
    let config = client.get_basin_config("my-events".parse()?).await?;

    // Delete
    client
        .delete_basin(DeleteBasinInput::new("my-events".parse()?))
        .await?;
    // ANCHOR_END: basin-operations
    println!("Basins: {:?}, config: {:?}", basins, config);

    let basin = client.basin(basin_name);

    // ANCHOR: stream-operations
    // List streams
    let streams = basin
        .list_streams(ListStreamsInput::new().with_prefix("user-".parse()?))
        .await?;

    // Create a stream
    basin
        .create_stream(CreateStreamInput::new("user-actions".parse()?))
        .await?;

    // Get configuration
    let config = basin.get_stream_config("user-actions".parse()?).await?;

    // Delete
    basin
        .delete_stream(DeleteStreamInput::new("user-actions".parse()?))
        .await?;
    // ANCHOR_END: stream-operations
    println!("Streams: {:?}, config: {:?}", streams, config);

    // ANCHOR: access-token-basic
    // List tokens (returns metadata, not the secret)
    let tokens = client.list_access_tokens(Default::default()).await?;

    // Issue a token scoped to streams under "users/1234/"
    let result = client
        .issue_access_token(
            IssueAccessTokenInput::new(
                "user-1234-rw-token".parse()?,
                AccessTokenScopeInput::from_op_group_perms(
                    OperationGroupPermissions::new()
                        .with_stream(ReadWritePermissions::read_write()),
                )
                .with_basins(BasinMatcher::Prefix("".parse()?)) // all basins
                .with_streams(StreamMatcher::Prefix("users/1234/".parse()?)),
            )
            .with_expires_at("2027-01-01T00:00:00Z".parse()?),
        )
        .await?;

    // Revoke a token
    client
        .revoke_access_token("user-1234-rw-token".parse()?)
        .await?;
    // ANCHOR_END: access-token-basic
    println!("Tokens: {:?}, issued: {:?}", tokens, result);

    // ANCHOR: access-token-restricted
    client
        .issue_access_token(IssueAccessTokenInput::new(
            "restricted-token".parse()?,
            AccessTokenScopeInput::from_op_group_perms(
                OperationGroupPermissions::new().with_stream(ReadWritePermissions::read_only()),
            )
            .with_basins(BasinMatcher::Exact("production".parse()?))
            .with_streams(StreamMatcher::Prefix("logs/".parse()?)),
        ))
        .await?;
    // ANCHOR_END: access-token-restricted

    // Pagination example - not executed by default
    if false {
        // ANCHOR: pagination
        // Iterate through all streams with automatic pagination
        let mut stream = basin.list_all_streams(ListAllStreamsInput::new());
        while let Some(info) = stream.next().await {
            let info = info?;
            println!("{}", info.name);
        }
        // ANCHOR_END: pagination
    }

    Ok(())
}

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

examples/docs_account_and_basins.rs (lines 71-82)

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

    let client = S2::new(S2Config::new(token))?;

    // ANCHOR: basin-operations
    // List basins
    let basins = client.list_basins(ListBasinsInput::new()).await?;

    // Create a basin
    client
        .create_basin(CreateBasinInput::new("my-events".parse()?))
        .await?;

    // Get configuration
    let config = client.get_basin_config("my-events".parse()?).await?;

    // Delete
    client
        .delete_basin(DeleteBasinInput::new("my-events".parse()?))
        .await?;
    // ANCHOR_END: basin-operations
    println!("Basins: {:?}, config: {:?}", basins, config);

    let basin = client.basin(basin_name);

    // ANCHOR: stream-operations
    // List streams
    let streams = basin
        .list_streams(ListStreamsInput::new().with_prefix("user-".parse()?))
        .await?;

    // Create a stream
    basin
        .create_stream(CreateStreamInput::new("user-actions".parse()?))
        .await?;

    // Get configuration
    let config = basin.get_stream_config("user-actions".parse()?).await?;

    // Delete
    basin
        .delete_stream(DeleteStreamInput::new("user-actions".parse()?))
        .await?;
    // ANCHOR_END: stream-operations
    println!("Streams: {:?}, config: {:?}", streams, config);

    // ANCHOR: access-token-basic
    // List tokens (returns metadata, not the secret)
    let tokens = client.list_access_tokens(Default::default()).await?;

    // Issue a token scoped to streams under "users/1234/"
    let result = client
        .issue_access_token(
            IssueAccessTokenInput::new(
                "user-1234-rw-token".parse()?,
                AccessTokenScopeInput::from_op_group_perms(
                    OperationGroupPermissions::new()
                        .with_stream(ReadWritePermissions::read_write()),
                )
                .with_basins(BasinMatcher::Prefix("".parse()?)) // all basins
                .with_streams(StreamMatcher::Prefix("users/1234/".parse()?)),
            )
            .with_expires_at("2027-01-01T00:00:00Z".parse()?),
        )
        .await?;

    // Revoke a token
    client
        .revoke_access_token("user-1234-rw-token".parse()?)
        .await?;
    // ANCHOR_END: access-token-basic
    println!("Tokens: {:?}, issued: {:?}", tokens, result);

    // ANCHOR: access-token-restricted
    client
        .issue_access_token(IssueAccessTokenInput::new(
            "restricted-token".parse()?,
            AccessTokenScopeInput::from_op_group_perms(
                OperationGroupPermissions::new().with_stream(ReadWritePermissions::read_only()),
            )
            .with_basins(BasinMatcher::Exact("production".parse()?))
            .with_streams(StreamMatcher::Prefix("logs/".parse()?)),
        ))
        .await?;
    // ANCHOR_END: access-token-restricted

    // Pagination example - not executed by default
    if false {
        // ANCHOR: pagination
        // Iterate through all streams with automatic pagination
        let mut stream = basin.list_all_streams(ListAllStreamsInput::new());
        while let Some(info) = stream.next().await {
            let info = info?;
            println!("{}", info.name);
        }
        // ANCHOR_END: pagination
    }

    Ok(())
}

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

Revoke an access token.

Examples found in repository

examples/docs_account_and_basins.rs (line 87)

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

    let client = S2::new(S2Config::new(token))?;

    // ANCHOR: basin-operations
    // List basins
    let basins = client.list_basins(ListBasinsInput::new()).await?;

    // Create a basin
    client
        .create_basin(CreateBasinInput::new("my-events".parse()?))
        .await?;

    // Get configuration
    let config = client.get_basin_config("my-events".parse()?).await?;

    // Delete
    client
        .delete_basin(DeleteBasinInput::new("my-events".parse()?))
        .await?;
    // ANCHOR_END: basin-operations
    println!("Basins: {:?}, config: {:?}", basins, config);

    let basin = client.basin(basin_name);

    // ANCHOR: stream-operations
    // List streams
    let streams = basin
        .list_streams(ListStreamsInput::new().with_prefix("user-".parse()?))
        .await?;

    // Create a stream
    basin
        .create_stream(CreateStreamInput::new("user-actions".parse()?))
        .await?;

    // Get configuration
    let config = basin.get_stream_config("user-actions".parse()?).await?;

    // Delete
    basin
        .delete_stream(DeleteStreamInput::new("user-actions".parse()?))
        .await?;
    // ANCHOR_END: stream-operations
    println!("Streams: {:?}, config: {:?}", streams, config);

    // ANCHOR: access-token-basic
    // List tokens (returns metadata, not the secret)
    let tokens = client.list_access_tokens(Default::default()).await?;

    // Issue a token scoped to streams under "users/1234/"
    let result = client
        .issue_access_token(
            IssueAccessTokenInput::new(
                "user-1234-rw-token".parse()?,
                AccessTokenScopeInput::from_op_group_perms(
                    OperationGroupPermissions::new()
                        .with_stream(ReadWritePermissions::read_write()),
                )
                .with_basins(BasinMatcher::Prefix("".parse()?)) // all basins
                .with_streams(StreamMatcher::Prefix("users/1234/".parse()?)),
            )
            .with_expires_at("2027-01-01T00:00:00Z".parse()?),
        )
        .await?;

    // Revoke a token
    client
        .revoke_access_token("user-1234-rw-token".parse()?)
        .await?;
    // ANCHOR_END: access-token-basic
    println!("Tokens: {:?}, issued: {:?}", tokens, result);

    // ANCHOR: access-token-restricted
    client
        .issue_access_token(IssueAccessTokenInput::new(
            "restricted-token".parse()?,
            AccessTokenScopeInput::from_op_group_perms(
                OperationGroupPermissions::new().with_stream(ReadWritePermissions::read_only()),
            )
            .with_basins(BasinMatcher::Exact("production".parse()?))
            .with_streams(StreamMatcher::Prefix("logs/".parse()?)),
        ))
        .await?;
    // ANCHOR_END: access-token-restricted

    // Pagination example - not executed by default
    if false {
        // ANCHOR: pagination
        // Iterate through all streams with automatic pagination
        let mut stream = basin.list_all_streams(ListAllStreamsInput::new());
        while let Some(info) = stream.next().await {
            let info = info?;
            println!("{}", info.name);
        }
        // ANCHOR_END: pagination
    }

    Ok(())
}

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.