Struct StreamConfig

pub struct StreamConfig {
    pub storage_class: StorageClass,
    pub retention_policy: Option<RetentionPolicy>,
    pub require_client_timestamps: Option<bool>,
    pub uncapped_client_timestamps: Option<bool>,
}

Stream configuration.

Fields

storage_class: StorageClass

Storage class for recent writes. This is the main cost:performance knob in S2.

retention_policy: Option<RetentionPolicy>

Retention policy for the stream. If unspecified, the default is to retain records for 7 days.

require_client_timestamps: Option<bool>

Controls how to handle timestamps when they are not provided by the client. If this is false (or not set), the record’s arrival time in milliseconds since Unix epoch will be assigned as its timestamp. If this is true, then any append without a client-specified timestamp will be rejected as invalid.

uncapped_client_timestamps: Option<bool>

Allow client timestamps to exceed the arrival time in milliseconds since Unix epoch. If this is false (or not set), client timestamps will be capped at the arrival time.

Implementations

impl StreamConfig

pub fn new() -> Self

Create a new stream config.

Examples found in repository

examples/create_stream.rs (line 17)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let token = std::env::var("S2_ACCESS_TOKEN")?;
    let config = ClientConfig::new(token);
    let client = Client::new(config);

    let basin: BasinName = "my-favorite-basin".parse()?;
    let basin_client = client.basin_client(basin);

    let stream = "my-favorite-stream";

    let stream_config = StreamConfig::new().with_storage_class(StorageClass::Express);

    let create_stream_request = CreateStreamRequest::new(stream).with_config(stream_config);

    let created_stream = basin_client.create_stream(create_stream_request).await?;
    println!("{created_stream:#?}");

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

    Ok(())
}

examples/reconfigure_stream.rs (line 17)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let token = std::env::var("S2_ACCESS_TOKEN")?;
    let config = ClientConfig::new(token);
    let basin: BasinName = "my-favorite-basin".parse()?;
    let basin_client = BasinClient::new(config, basin);

    let stream = "my-favorite-stream";

    let stream_config_updates = StreamConfig::new().with_retention_policy(RetentionPolicy::Age(
        // Change to retention policy to 1 day
        Duration::from_secs(24 * 60 * 60),
    ));

    let reconfigure_stream_request = ReconfigureStreamRequest::new(stream)
        .with_config(stream_config_updates)
        // Field mask specifies which fields to update.
        .with_mask(vec!["retention_policy".to_string()]);

    let updated_stream_config = basin_client
        .reconfigure_stream(reconfigure_stream_request)
        .await?;

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

    Ok(())
}

examples/create_basin.rs (line 16)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let token = std::env::var("S2_ACCESS_TOKEN")?;
    let config = ClientConfig::new(token);
    let client = Client::new(config);

    let basin: BasinName = "my-favorite-basin".parse()?;

    let default_stream_config = StreamConfig::new().with_retention_policy(RetentionPolicy::Age(
        // Set the default retention age to 10 days.
        Duration::from_secs(10 * 24 * 60 * 60),
    ));

    let basin_config = BasinConfig::new()
        .with_default_stream_config(default_stream_config)
        .with_create_stream_on_append(false)
        .with_create_stream_on_read(false);

    let create_basin_request = CreateBasinRequest::new(basin.clone()).with_config(basin_config);

    let created_basin = client.create_basin(create_basin_request).await?;
    println!("{created_basin:#?}");

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

    Ok(())
}

examples/reconfigure_basin.rs (line 15)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let token = std::env::var("S2_ACCESS_TOKEN")?;
    let config = ClientConfig::new(token);
    let client = Client::new(config);

    let basin: BasinName = "my-favorite-basin".parse()?;

    let default_stream_config_updates =
        StreamConfig::new().with_storage_class(StorageClass::Standard);
    let basin_config_updates = BasinConfig::new()
        .with_default_stream_config(default_stream_config_updates)
        .with_create_stream_on_append(true)
        .with_create_stream_on_read(true);

    let reconfigure_basin_request = ReconfigureBasinRequest::new(basin)
        .with_config(basin_config_updates)
        // Field mask specifies which fields to update.
        .with_mask(vec!["default_stream_config.retention_policy".to_string()]);

    let updated_basin_config = client.reconfigure_basin(reconfigure_basin_request).await?;

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

    Ok(())
}

pub fn with_storage_class(self, storage_class: impl Into<StorageClass>) -> Self

Overwrite storage class.

Examples found in repository

examples/create_stream.rs (line 17)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let token = std::env::var("S2_ACCESS_TOKEN")?;
    let config = ClientConfig::new(token);
    let client = Client::new(config);

    let basin: BasinName = "my-favorite-basin".parse()?;
    let basin_client = client.basin_client(basin);

    let stream = "my-favorite-stream";

    let stream_config = StreamConfig::new().with_storage_class(StorageClass::Express);

    let create_stream_request = CreateStreamRequest::new(stream).with_config(stream_config);

    let created_stream = basin_client.create_stream(create_stream_request).await?;
    println!("{created_stream:#?}");

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

    Ok(())
}

examples/reconfigure_basin.rs (line 15)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let token = std::env::var("S2_ACCESS_TOKEN")?;
    let config = ClientConfig::new(token);
    let client = Client::new(config);

    let basin: BasinName = "my-favorite-basin".parse()?;

    let default_stream_config_updates =
        StreamConfig::new().with_storage_class(StorageClass::Standard);
    let basin_config_updates = BasinConfig::new()
        .with_default_stream_config(default_stream_config_updates)
        .with_create_stream_on_append(true)
        .with_create_stream_on_read(true);

    let reconfigure_basin_request = ReconfigureBasinRequest::new(basin)
        .with_config(basin_config_updates)
        // Field mask specifies which fields to update.
        .with_mask(vec!["default_stream_config.retention_policy".to_string()]);

    let updated_basin_config = client.reconfigure_basin(reconfigure_basin_request).await?;

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

    Ok(())
}

pub fn with_retention_policy(self, retention_policy: RetentionPolicy) -> Self

Overwrite retention policy.

Examples found in repository

examples/reconfigure_stream.rs (lines 17-20)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let token = std::env::var("S2_ACCESS_TOKEN")?;
    let config = ClientConfig::new(token);
    let basin: BasinName = "my-favorite-basin".parse()?;
    let basin_client = BasinClient::new(config, basin);

    let stream = "my-favorite-stream";

    let stream_config_updates = StreamConfig::new().with_retention_policy(RetentionPolicy::Age(
        // Change to retention policy to 1 day
        Duration::from_secs(24 * 60 * 60),
    ));

    let reconfigure_stream_request = ReconfigureStreamRequest::new(stream)
        .with_config(stream_config_updates)
        // Field mask specifies which fields to update.
        .with_mask(vec!["retention_policy".to_string()]);

    let updated_stream_config = basin_client
        .reconfigure_stream(reconfigure_stream_request)
        .await?;

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

    Ok(())
}

examples/create_basin.rs (lines 16-19)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let token = std::env::var("S2_ACCESS_TOKEN")?;
    let config = ClientConfig::new(token);
    let client = Client::new(config);

    let basin: BasinName = "my-favorite-basin".parse()?;

    let default_stream_config = StreamConfig::new().with_retention_policy(RetentionPolicy::Age(
        // Set the default retention age to 10 days.
        Duration::from_secs(10 * 24 * 60 * 60),
    ));

    let basin_config = BasinConfig::new()
        .with_default_stream_config(default_stream_config)
        .with_create_stream_on_append(false)
        .with_create_stream_on_read(false);

    let create_basin_request = CreateBasinRequest::new(basin.clone()).with_config(basin_config);

    let created_basin = client.create_basin(create_basin_request).await?;
    println!("{created_basin:#?}");

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

    Ok(())
}

pub fn with_require_client_timestamps( self, require_client_timestamps: bool, ) -> Self

Overwrite require_client_timestamps.

pub fn with_uncapped_client_timestamps( self, uncapped_client_timestamps: bool, ) -> Self

Overwrite uncapped_client_timestamps.

Trait Implementations

impl Clone for StreamConfig

fn clone(&self) -> StreamConfig

Returns a copy of the value.

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

Performs copy-assignment from source.

impl Debug for StreamConfig

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

Formats the value using the given formatter.

impl Default for StreamConfig

fn default() -> StreamConfig

Returns the “default value” for a type.