s2::types

Struct AppendInput

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

👎Deprecated since 0.21.1: This crate has been renamed to s2-sdk. Please update your Cargo.toml to use s2-sdk instead.

Expand description

Input for append requests.

Fields§

§records: AppendRecordBatch

👎Deprecated since 0.21.1: This crate has been renamed to s2-sdk. Please update your Cargo.toml to use s2-sdk instead.

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>

👎Deprecated since 0.21.1: This crate has been renamed to s2-sdk. Please update your Cargo.toml to use s2-sdk instead.

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

§fencing_token: Option<FencingToken>

👎Deprecated since 0.21.1: This crate has been renamed to s2-sdk. Please update your Cargo.toml to use s2-sdk instead.

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

👎Deprecated since 0.21.1: This crate has been renamed to s2-sdk. Please update your Cargo.toml to use s2-sdk instead.

Create a new append input from record batch.

Examples found in repository ?

examples/explicit_trim.rs (line 25)

7async fn main() -> Result<(), Box<dyn std::error::Error>> {
8    let token = std::env::var("S2_ACCESS_TOKEN")?;
9    let config = ClientConfig::new(token);
10    let basin: BasinName = "my-favorite-basin".parse()?;
11    let stream = "my-favorite-stream";
12    let stream_client = StreamClient::new(config, basin, stream);
13
14    let tail = stream_client.check_tail().await?;
15    if tail.seq_num == 0 {
16        println!("Empty stream");
17        return Ok(());
18    }
19
20    let latest_seq_num = tail.seq_num - 1;
21    let trim_request = CommandRecord::trim(latest_seq_num);
22
23    let append_record_batch = AppendRecordBatch::try_from_iter([trim_request])
24        .expect("valid batch with 1 command record");
25    let append_input = AppendInput::new(append_record_batch);
26    let _ = stream_client.append(append_input).await?;
27
28    println!("Trim requested");
29
30    Ok(())
31}

More examples

Hide additional examples

examples/producer.rs (line 22)

9async fn main() -> Result<(), Box<dyn std::error::Error>> {
10    let token = std::env::var("S2_ACCESS_TOKEN")?;
11    let config = ClientConfig::new(token);
12    let basin: BasinName = "my-favorite-basin".parse()?;
13    let stream = "my-favorite-stream";
14    let stream_client = StreamClient::new(config, basin, stream);
15
16    let fencing_token = FencingToken::generate(32).expect("valid fencing token");
17
18    // Set the fencing token.
19    let fencing_token_record: AppendRecord = CommandRecord::fence(fencing_token.clone()).into();
20    let fencing_token_batch = AppendRecordBatch::try_from_iter([fencing_token_record])
21        .expect("valid batch with 1 append record");
22    let fencing_token_append_input = AppendInput::new(fencing_token_batch);
23    let set_fencing_token = stream_client.append(fencing_token_append_input).await?;
24
25    let match_seq_num = set_fencing_token.tail.seq_num;
26
27    // Stream of records
28    let append_stream = futures::stream::iter([
29        AppendRecord::new("record_1")?,
30        AppendRecord::new("record_2")?,
31    ]);
32
33    let append_records_batching_opts = AppendRecordsBatchingOpts::new()
34        .with_fencing_token(Some(fencing_token))
35        .with_match_seq_num(Some(match_seq_num));
36
37    let append_session_request =
38        AppendRecordsBatchingStream::new(append_stream, append_records_batching_opts);
39
40    let mut append_session_stream = stream_client.append_session(append_session_request).await?;
41
42    while let Some(next) = append_session_stream.next().await {
43        let next = next?;
44        println!("{next:#?}");
45    }
46
47    Ok(())
48}