s2::types

Struct AppendInput

pub struct AppendInput {
    pub records: AppendRecordBatch,
    pub match_seq_num: Option<u64>,
    pub fencing_token: Option<FencingToken>,
}

Expand description

Input for append requests.

Fields§

§records: AppendRecordBatch

Batch of records to append atomically, which must contain at least one record, and no more than 1000. The total size of a batch of records may not exceed 1MiB of metered bytes.

§match_seq_num: Option<u64>

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

§fencing_token: Option<FencingToken>

Enforce a fencing token which must have been previously set by a fence command record.

Implementations§

Source §

impl AppendInput

Source

pub fn new(records: impl Into<AppendRecordBatch>) -> Self

Create a new append input from record batch.

Examples found in repository ?

examples/explicit_trim.rs (line 25)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let token = std::env::var("S2_AUTH_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 tail = stream_client.check_tail().await?;
    if tail == 0 {
        println!("Empty stream");
        return Ok(());
    }

    let latest_seq_num = tail - 1;
    let trim_request = CommandRecord::trim(latest_seq_num);

    let append_record_batch = AppendRecordBatch::try_from_iter([trim_request])
        .expect("valid batch with 1 command record");
    let append_input = AppendInput::new(append_record_batch);
    let _ = stream_client.append(append_input).await?;

    println!("Trim requested");

    Ok(())
}

More examples

Hide additional examples

examples/producer.rs (line 22)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let token = std::env::var("S2_AUTH_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(16).expect("valid fencing token with 16 bytes");

    // 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.next_seq_num; // Tail

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