Struct AppendRecordsBatchingOpts 

pub struct AppendRecordsBatchingOpts { /* private fields */ }

Options to configure batching scheme for AppendRecordsBatchingStream.

Implementations

impl AppendRecordsBatchingOpts

pub fn new() -> Self

Construct an options struct with defaults.

Examples found in repository

examples/producer.rs (line 33)

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 stream = "my-favorite-stream";
    let stream_client = StreamClient::new(config, basin, stream);

    let fencing_token = FencingToken::generate(32).expect("valid fencing token");

    // Set the fencing token.
    let fencing_token_record: AppendRecord = CommandRecord::fence(fencing_token.clone()).into();
    let fencing_token_batch = AppendRecordBatch::try_from_iter([fencing_token_record])
        .expect("valid batch with 1 append record");
    let fencing_token_append_input = AppendInput::new(fencing_token_batch);
    let set_fencing_token = stream_client.append(fencing_token_append_input).await?;

    let match_seq_num = set_fencing_token.tail.seq_num;

    // Stream of records
    let append_stream = futures::stream::iter([
        AppendRecord::new("record_1")?,
        AppendRecord::new("record_2")?,
    ]);

    let append_records_batching_opts = AppendRecordsBatchingOpts::new()
        .with_fencing_token(Some(fencing_token))
        .with_match_seq_num(Some(match_seq_num));

    let append_session_request =
        AppendRecordsBatchingStream::new(append_stream, append_records_batching_opts);

    let mut append_session_stream = stream_client.append_session(append_session_request).await?;

    while let Some(next) = append_session_stream.next().await {
        let next = next?;
        println!("{next:#?}");
    }

    Ok(())
}

pub fn with_max_batch_records(self, max_batch_records: usize) -> Self

Maximum number of records in a batch.

pub fn with_match_seq_num(self, match_seq_num: Option<u64>) -> Self

Enforce that the sequence number issued to the first record matches.

This is incremented automatically for each batch.

Examples found in repository

examples/producer.rs (line 35)

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 stream = "my-favorite-stream";
    let stream_client = StreamClient::new(config, basin, stream);

    let fencing_token = FencingToken::generate(32).expect("valid fencing token");

    // Set the fencing token.
    let fencing_token_record: AppendRecord = CommandRecord::fence(fencing_token.clone()).into();
    let fencing_token_batch = AppendRecordBatch::try_from_iter([fencing_token_record])
        .expect("valid batch with 1 append record");
    let fencing_token_append_input = AppendInput::new(fencing_token_batch);
    let set_fencing_token = stream_client.append(fencing_token_append_input).await?;

    let match_seq_num = set_fencing_token.tail.seq_num;

    // Stream of records
    let append_stream = futures::stream::iter([
        AppendRecord::new("record_1")?,
        AppendRecord::new("record_2")?,
    ]);

    let append_records_batching_opts = AppendRecordsBatchingOpts::new()
        .with_fencing_token(Some(fencing_token))
        .with_match_seq_num(Some(match_seq_num));

    let append_session_request =
        AppendRecordsBatchingStream::new(append_stream, append_records_batching_opts);

    let mut append_session_stream = stream_client.append_session(append_session_request).await?;

    while let Some(next) = append_session_stream.next().await {
        let next = next?;
        println!("{next:#?}");
    }

    Ok(())
}

pub fn with_fencing_token(self, fencing_token: Option<FencingToken>) -> Self

Enforce a fencing token.

Examples found in repository

examples/producer.rs (line 34)

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 stream = "my-favorite-stream";
    let stream_client = StreamClient::new(config, basin, stream);

    let fencing_token = FencingToken::generate(32).expect("valid fencing token");

    // Set the fencing token.
    let fencing_token_record: AppendRecord = CommandRecord::fence(fencing_token.clone()).into();
    let fencing_token_batch = AppendRecordBatch::try_from_iter([fencing_token_record])
        .expect("valid batch with 1 append record");
    let fencing_token_append_input = AppendInput::new(fencing_token_batch);
    let set_fencing_token = stream_client.append(fencing_token_append_input).await?;

    let match_seq_num = set_fencing_token.tail.seq_num;

    // Stream of records
    let append_stream = futures::stream::iter([
        AppendRecord::new("record_1")?,
        AppendRecord::new("record_2")?,
    ]);

    let append_records_batching_opts = AppendRecordsBatchingOpts::new()
        .with_fencing_token(Some(fencing_token))
        .with_match_seq_num(Some(match_seq_num));

    let append_session_request =
        AppendRecordsBatchingStream::new(append_stream, append_records_batching_opts);

    let mut append_session_stream = stream_client.append_session(append_session_request).await?;

    while let Some(next) = append_session_stream.next().await {
        let next = next?;
        println!("{next:#?}");
    }

    Ok(())
}

pub fn with_linger(self, linger_duration: impl Into<Duration>) -> Self

Linger duration for records before flushing.

A linger duration of 5ms is set by default. Set to Duration::ZERO to disable.

Trait Implementations

impl Clone for AppendRecordsBatchingOpts

fn clone(&self) -> AppendRecordsBatchingOpts

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for AppendRecordsBatchingOpts

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

Formats the value using the given formatter.

impl Default for AppendRecordsBatchingOpts

fn default() -> Self

Returns the “default value” for a type.