Struct AppendInput 

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

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

impl AppendInput

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_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 tail = stream_client.check_tail().await?;
    if tail.seq_num == 0 {
        println!("Empty stream");
        return Ok(());
    }

    let latest_seq_num = tail.seq_num - 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(())
}

examples/producer.rs (line 22)

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_match_seq_num(self, match_seq_num: impl Into<u64>) -> Self

Overwrite match sequence number.

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

Overwrite fencing token.

Trait Implementations

impl Clone for AppendInput

fn clone(&self) -> AppendInput

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for AppendInput

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

Formats the value using the given formatter.

impl Default for AppendInput

fn default() -> AppendInput

Returns the “default value” for a type.

impl MeteredBytes for AppendInput

fn metered_bytes(&self) -> u64

Return the metered bytes of the object.