Struct AppendInput 

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

Input for append operation and AppendSession::submit.

Fields (Non-exhaustive)

Non-exhaustive structs could have additional fields added in future. Therefore, non-exhaustive structs cannot be constructed in external crates using the traditional Struct { .. } syntax; cannot be matched against without a wildcard ..; and struct update syntax will not work.

records: AppendRecordBatch

Batch of records to append atomically.

match_seq_num: Option<u64>

Expected sequence number for the first record in the batch.

If unspecified, no matching is performed. If specified and mismatched, the append fails.

fencing_token: Option<FencingToken>

Fencing token to match against the stream’s current fencing token.

If unspecified, no matching is performed. If specified and mismatched, the append fails. A stream defaults to "" as its fencing token.

Implementations

impl AppendInput

pub fn new(records: AppendRecordBatch) -> Self

Create a new AppendInput with the given batch of records.

Examples found in repository

examples/explicit_trim.rs (lines 26-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 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 tail = stream.check_tail().await?;
    if tail.seq_num == 0 {
        println!("Empty stream");
        return Ok(());
    }

    let input = AppendInput::new(AppendRecordBatch::try_from_iter([CommandRecord::trim(
        tail.seq_num - 1,
    )
    .into()])?);
    stream.append(input).await?;
    println!("Trim requested");

    Ok(())
}

examples/docs_streams.rs (line 47)

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))?;
    let basin = client.basin(basin_name);

    // Create a temporary stream for examples
    let stream_name: StreamName = format!(
        "docs-streams-{}",
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)?
            .as_millis()
    )
    .parse()?;
    basin
        .create_stream(s2_sdk::types::CreateStreamInput::new(stream_name.clone()))
        .await?;

    // ANCHOR: simple-append
    let stream = basin.stream(stream_name.clone());

    let records = AppendRecordBatch::try_from_iter([
        AppendRecord::new("first event")?,
        AppendRecord::new("second event")?,
    ])?;

    let ack = stream.append(AppendInput::new(records)).await?;

    // ack tells us where the records landed
    println!(
        "Wrote records {} through {}",
        ack.start.seq_num,
        ack.end.seq_num - 1
    );
    // ANCHOR_END: simple-append

    // ANCHOR: simple-read
    let batch = stream
        .read(
            ReadInput::new()
                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
                .with_stop(ReadStop::new().with_limits(ReadLimits::new().with_count(100))),
        )
        .await?;

    for record in batch.records {
        println!("[{}] {:?}", record.seq_num, record.body);
    }
    // ANCHOR_END: simple-read

    // ANCHOR: append-session
    let session = stream.append_session(AppendSessionConfig::new());

    // Submit a batch - this enqueues it and returns a ticket
    let records = AppendRecordBatch::try_from_iter([
        AppendRecord::new("event-1")?,
        AppendRecord::new("event-2")?,
    ])?;
    let ticket = session.submit(AppendInput::new(records)).await?;

    // Wait for durability
    let ack = ticket.await?;
    println!("Durable at seqNum {}", ack.start.seq_num);

    session.close().await?;
    // ANCHOR_END: append-session

    // ANCHOR: producer
    let producer = stream.producer(
        ProducerConfig::new()
            .with_batching(BatchingConfig::new().with_linger(Duration::from_millis(5))),
    );

    // Submit individual records
    let ticket = producer.submit(AppendRecord::new("my event")?).await?;

    // Get the exact sequence number
    let ack = ticket.await?;
    println!("Record durable at seqNum {}", ack.seq_num);

    producer.close().await?;
    // ANCHOR_END: producer

    // ANCHOR: check-tail
    let tail = stream.check_tail().await?;
    println!("Stream has {} records", tail.seq_num);
    // ANCHOR_END: check-tail

    // Cleanup
    basin
        .delete_stream(s2_sdk::types::DeleteStreamInput::new(stream_name))
        .await?;

    println!("Streams examples completed");

    // The following read session examples are for documentation snippets only.
    // They are not executed because they would block waiting for new records.
    if std::env::var("RUN_READ_SESSIONS").is_err() {
        return Ok(());
    }

    // ANCHOR: read-session
    let mut session = stream
        .read_session(ReadInput::new().with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0))))
        .await?;

    while let Some(batch) = session.next().await {
        let batch = batch?;
        for record in batch.records {
            println!("[{}] {:?}", record.seq_num, record.body);
        }
    }
    // ANCHOR_END: read-session

    // ANCHOR: read-session-tail-offset
    // Start reading from 10 records before the current tail
    let mut session = stream
        .read_session(
            ReadInput::new().with_start(ReadStart::new().with_from(ReadFrom::TailOffset(10))),
        )
        .await?;

    while let Some(batch) = session.next().await {
        let batch = batch?;
        for record in batch.records {
            println!("[{}] {:?}", record.seq_num, record.body);
        }
    }
    // ANCHOR_END: read-session-tail-offset

    // ANCHOR: read-session-timestamp
    // Start reading from a specific timestamp
    let one_hour_ago = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)?
        .as_millis() as u64
        - 3600 * 1000;
    let mut session = stream
        .read_session(
            ReadInput::new()
                .with_start(ReadStart::new().with_from(ReadFrom::Timestamp(one_hour_ago))),
        )
        .await?;

    while let Some(batch) = session.next().await {
        let batch = batch?;
        for record in batch.records {
            println!("[{}] {:?}", record.seq_num, record.body);
        }
    }
    // ANCHOR_END: read-session-timestamp

    // ANCHOR: read-session-until
    // Read records until a specific timestamp
    let one_hour_ago = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)?
        .as_millis() as u64
        - 3600 * 1000;
    let mut session = stream
        .read_session(
            ReadInput::new()
                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
                .with_stop(ReadStop::new().with_until(..one_hour_ago)),
        )
        .await?;

    while let Some(batch) = session.next().await {
        let batch = batch?;
        for record in batch.records {
            println!("[{}] {:?}", record.seq_num, record.body);
        }
    }
    // ANCHOR_END: read-session-until

    // ANCHOR: read-session-wait
    // Read all available records, and once reaching the current tail, wait an additional 30 seconds
    // for new ones
    let mut session = stream
        .read_session(
            ReadInput::new()
                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
                .with_stop(ReadStop::new().with_wait(30)),
        )
        .await?;

    while let Some(batch) = session.next().await {
        let batch = batch?;
        for record in batch.records {
            println!("[{}] {:?}", record.seq_num, record.body);
        }
    }
    // ANCHOR_END: read-session-wait

    Ok(())
}

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

Set the expected sequence number for the first record in the batch.

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

Set the fencing token to match against the stream’s current 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 From<AppendInput> for AppendInput

fn from(value: AppendInput) -> Self

Converts to this type from the input type.