Struct S2Stream

Source

pub struct S2Stream { /* private fields */ }

Expand description

A stream in an S2 basin.

See S2Basin::stream.

Implementations§

Source §

impl S2Stream

Source

pub fn with_encryption_key(self, encryption: EncryptionKey) -> Self

Set the encryption key for this stream handle.

Examples found in repository ?

examples/docs_encryption.rs (line 52)

15async fn main() -> Result<(), Box<dyn std::error::Error>> {
16    let access_token = std::env::var("S2_ACCESS_TOKEN")?;
17    let basin_name: BasinName = std::env::var("S2_BASIN")?.parse()?;
18    let stream_name: StreamName = format!(
19        "docs-encryption-{}",
20        std::time::SystemTime::now()
21            .duration_since(std::time::UNIX_EPOCH)?
22            .as_millis()
23    )
24    .parse()?;
25
26    let client = S2::new(S2Config::new(access_token))?;
27
28    // ANCHOR: basin-cipher
29    client
30        .create_basin(
31            CreateBasinInput::new(basin_name.clone())
32                .with_config(BasinConfig::new().with_stream_cipher(EncryptionAlgorithm::Aegis256)),
33        )
34        .await?;
35
36    client
37        .reconfigure_basin(ReconfigureBasinInput::new(
38            basin_name.clone(),
39            BasinReconfiguration::new().with_stream_cipher(EncryptionAlgorithm::Aes256Gcm),
40        ))
41        .await?;
42    // ANCHOR_END: basin-cipher
43
44    let basin = client.basin(basin_name.clone());
45    basin
46        .create_stream(CreateStreamInput::new(stream_name.clone()))
47        .await?;
48
49    // ANCHOR: append-read
50    let stream = basin
51        .stream(stream_name.clone())
52        .with_encryption_key(std::env::var("S2_ENCRYPTION_KEY")?.parse()?);
53
54    stream
55        .append(AppendInput::new(AppendRecordBatch::try_from_iter([
56            AppendRecord::new("top secret")?,
57        ])?))
58        .await?;
59
60    let batch = stream
61        .read(
62            ReadInput::new()
63                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
64                .with_stop(ReadStop::new().with_limits(ReadLimits::new().with_count(10))),
65        )
66        .await?;
67    // ANCHOR_END: append-read
68
69    println!("Read {} encrypted record(s)", batch.records.len());
70
71    basin
72        .delete_stream(DeleteStreamInput::new(stream_name))
73        .await?;
74
75    Ok(())
76}

Source

pub async fn check_tail(&self) -> Result<StreamPosition, ReadError>

Check tail position.

Examples found in repository ?

examples/explicit_trim.rs (line 20)

7async fn main() -> Result<(), Box<dyn std::error::Error>> {
8    let access_token =
9        std::env::var("S2_ACCESS_TOKEN").map_err(|_| "S2_ACCESS_TOKEN env var not set")?;
10    let basin_name: BasinName = std::env::var("S2_BASIN")
11        .map_err(|_| "S2_BASIN env var not set")?
12        .parse()?;
13    let stream_name: StreamName = std::env::var("S2_STREAM")
14        .map_err(|_| "S2_STREAM env var not set")?
15        .parse()?;
16
17    let s2 = S2::new(S2Config::new(access_token))?;
18    let stream = s2.basin(basin_name).stream(stream_name);
19
20    let tail = stream.check_tail().await?;
21    if tail.seq_num == 0 {
22        println!("Empty stream");
23        return Ok(());
24    }
25
26    let input = AppendInput::new(AppendRecordBatch::try_from_iter([CommandRecord::trim(
27        tail.seq_num - 1,
28    )
29    .into()])?);
30    stream.append(input).await?;
31    println!("Trim requested");
32
33    Ok(())
34}

More examples

Hide additional examples

examples/docs_streams.rs (line 105)

20async fn main() -> Result<(), Box<dyn std::error::Error>> {
21    let access_token = std::env::var("S2_ACCESS_TOKEN")?;
22    let basin_name: BasinName = std::env::var("S2_BASIN")?.parse()?;
23
24    let client = S2::new(S2Config::new(access_token))?;
25    let basin = client.basin(basin_name);
26
27    // Create a temporary stream for examples
28    let stream_name: StreamName = format!(
29        "docs-streams-{}",
30        std::time::SystemTime::now()
31            .duration_since(std::time::UNIX_EPOCH)?
32            .as_millis()
33    )
34    .parse()?;
35    basin
36        .create_stream(s2_sdk::types::CreateStreamInput::new(stream_name.clone()))
37        .await?;
38
39    // ANCHOR: simple-append
40    let stream = basin.stream(stream_name.clone());
41
42    let ack = stream
43        .append(AppendInput::new(AppendRecordBatch::try_from_iter([
44            AppendRecord::new("first event")?,
45            AppendRecord::new("second event")?,
46        ])?))
47        .await?;
48
49    // ack tells us where the records landed
50    println!(
51        "Wrote records {} through {}",
52        ack.start.seq_num,
53        ack.end.seq_num - 1
54    );
55    // ANCHOR_END: simple-append
56
57    // ANCHOR: simple-read
58    let batch = stream
59        .read(
60            ReadInput::new()
61                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
62                .with_stop(ReadStop::new().with_limits(ReadLimits::new().with_count(100))),
63        )
64        .await?;
65
66    for record in batch.records {
67        println!("[{}] {:?}", record.seq_num, record.body);
68    }
69    // ANCHOR_END: simple-read
70
71    // ANCHOR: append-session
72    let session = stream.append_session(AppendSessionConfig::new());
73
74    // Submit a batch - this enqueues it and returns a ticket
75    let records = AppendRecordBatch::try_from_iter([
76        AppendRecord::new("event-1")?,
77        AppendRecord::new("event-2")?,
78    ])?;
79    let ticket = session.submit(AppendInput::new(records)).await?;
80
81    // Wait for durability
82    let ack = ticket.await?;
83    println!("Durable at seqNum {}", ack.start.seq_num);
84
85    session.close().await?;
86    // ANCHOR_END: append-session
87
88    // ANCHOR: producer
89    let producer = stream.producer(
90        ProducerConfig::new()
91            .with_batching(BatchingConfig::new().with_linger(Duration::from_millis(5))),
92    );
93
94    // Submit individual records
95    let ticket = producer.submit(AppendRecord::new("my event")?).await?;
96
97    // Get the exact sequence number
98    let ack = ticket.await?;
99    println!("Record durable at seqNum {}", ack.seq_num);
100
101    producer.close().await?;
102    // ANCHOR_END: producer
103
104    // ANCHOR: check-tail
105    let tail = stream.check_tail().await?;
106    println!("Stream has {} records", tail.seq_num);
107    // ANCHOR_END: check-tail
108
109    // Cleanup
110    basin
111        .delete_stream(s2_sdk::types::DeleteStreamInput::new(stream_name))
112        .await?;
113
114    println!("Streams examples completed");
115
116    // The following read session examples are for documentation snippets only.
117    // They are not executed because they would block waiting for new records.
118    if std::env::var("RUN_READ_SESSIONS").is_err() {
119        return Ok(());
120    }
121
122    // ANCHOR: read-session
123    let mut session = stream
124        .read_session(
125            ReadInput::new().with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0))),
126            ReadSessionConfig::default(),
127        )
128        .await?;
129
130    while let Some(batch) = session.next().await {
131        let batch = batch?;
132        for record in batch.records {
133            println!("[{}] {:?}", record.seq_num, record.body);
134        }
135    }
136    // ANCHOR_END: read-session
137
138    // ANCHOR: read-session-tail-offset
139    // Start reading from 10 records before the current tail
140    let mut session = stream
141        .read_session(
142            ReadInput::new().with_start(ReadStart::new().with_from(ReadFrom::TailOffset(10))),
143            ReadSessionConfig::default(),
144        )
145        .await?;
146
147    while let Some(batch) = session.next().await {
148        let batch = batch?;
149        for record in batch.records {
150            println!("[{}] {:?}", record.seq_num, record.body);
151        }
152    }
153    // ANCHOR_END: read-session-tail-offset
154
155    // ANCHOR: read-session-timestamp
156    // Start reading from a specific timestamp
157    let one_hour_ago = std::time::SystemTime::now()
158        .duration_since(std::time::UNIX_EPOCH)?
159        .as_millis() as u64
160        - 3600 * 1000;
161    let mut session = stream
162        .read_session(
163            ReadInput::new()
164                .with_start(ReadStart::new().with_from(ReadFrom::Timestamp(one_hour_ago))),
165            ReadSessionConfig::default(),
166        )
167        .await?;
168
169    while let Some(batch) = session.next().await {
170        let batch = batch?;
171        for record in batch.records {
172            println!("[{}] {:?}", record.seq_num, record.body);
173        }
174    }
175    // ANCHOR_END: read-session-timestamp
176
177    // ANCHOR: read-session-until
178    // Read records until a specific timestamp
179    let one_hour_ago = std::time::SystemTime::now()
180        .duration_since(std::time::UNIX_EPOCH)?
181        .as_millis() as u64
182        - 3600 * 1000;
183    let mut session = stream
184        .read_session(
185            ReadInput::new()
186                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
187                .with_stop(ReadStop::new().with_until(..one_hour_ago)),
188            ReadSessionConfig::default(),
189        )
190        .await?;
191
192    while let Some(batch) = session.next().await {
193        let batch = batch?;
194        for record in batch.records {
195            println!("[{}] {:?}", record.seq_num, record.body);
196        }
197    }
198    // ANCHOR_END: read-session-until
199
200    // ANCHOR: read-session-wait
201    // Read all available records, and once reaching the current tail, wait an additional 30 seconds
202    // for new ones
203    let mut session = stream
204        .read_session(
205            ReadInput::new()
206                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
207                .with_stop(ReadStop::new().with_wait(30)),
208            ReadSessionConfig::default(),
209        )
210        .await?;
211
212    while let Some(batch) = session.next().await {
213        let batch = batch?;
214        for record in batch.records {
215            println!("[{}] {:?}", record.seq_num, record.body);
216        }
217    }
218    // ANCHOR_END: read-session-wait
219
220    Ok(())
221}

Source

pub async fn append(&self, input: AppendInput) -> Result<AppendAck, AppendError>

Append records.

Examples found in repository ?

examples/explicit_trim.rs (line 30)

7async fn main() -> Result<(), Box<dyn std::error::Error>> {
8    let access_token =
9        std::env::var("S2_ACCESS_TOKEN").map_err(|_| "S2_ACCESS_TOKEN env var not set")?;
10    let basin_name: BasinName = std::env::var("S2_BASIN")
11        .map_err(|_| "S2_BASIN env var not set")?
12        .parse()?;
13    let stream_name: StreamName = std::env::var("S2_STREAM")
14        .map_err(|_| "S2_STREAM env var not set")?
15        .parse()?;
16
17    let s2 = S2::new(S2Config::new(access_token))?;
18    let stream = s2.basin(basin_name).stream(stream_name);
19
20    let tail = stream.check_tail().await?;
21    if tail.seq_num == 0 {
22        println!("Empty stream");
23        return Ok(());
24    }
25
26    let input = AppendInput::new(AppendRecordBatch::try_from_iter([CommandRecord::trim(
27        tail.seq_num - 1,
28    )
29    .into()])?);
30    stream.append(input).await?;
31    println!("Trim requested");
32
33    Ok(())
34}

More examples

Hide additional examples

examples/docs_encryption.rs (lines 55-57)

15async fn main() -> Result<(), Box<dyn std::error::Error>> {
16    let access_token = std::env::var("S2_ACCESS_TOKEN")?;
17    let basin_name: BasinName = std::env::var("S2_BASIN")?.parse()?;
18    let stream_name: StreamName = format!(
19        "docs-encryption-{}",
20        std::time::SystemTime::now()
21            .duration_since(std::time::UNIX_EPOCH)?
22            .as_millis()
23    )
24    .parse()?;
25
26    let client = S2::new(S2Config::new(access_token))?;
27
28    // ANCHOR: basin-cipher
29    client
30        .create_basin(
31            CreateBasinInput::new(basin_name.clone())
32                .with_config(BasinConfig::new().with_stream_cipher(EncryptionAlgorithm::Aegis256)),
33        )
34        .await?;
35
36    client
37        .reconfigure_basin(ReconfigureBasinInput::new(
38            basin_name.clone(),
39            BasinReconfiguration::new().with_stream_cipher(EncryptionAlgorithm::Aes256Gcm),
40        ))
41        .await?;
42    // ANCHOR_END: basin-cipher
43
44    let basin = client.basin(basin_name.clone());
45    basin
46        .create_stream(CreateStreamInput::new(stream_name.clone()))
47        .await?;
48
49    // ANCHOR: append-read
50    let stream = basin
51        .stream(stream_name.clone())
52        .with_encryption_key(std::env::var("S2_ENCRYPTION_KEY")?.parse()?);
53
54    stream
55        .append(AppendInput::new(AppendRecordBatch::try_from_iter([
56            AppendRecord::new("top secret")?,
57        ])?))
58        .await?;
59
60    let batch = stream
61        .read(
62            ReadInput::new()
63                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
64                .with_stop(ReadStop::new().with_limits(ReadLimits::new().with_count(10))),
65        )
66        .await?;
67    // ANCHOR_END: append-read
68
69    println!("Read {} encrypted record(s)", batch.records.len());
70
71    basin
72        .delete_stream(DeleteStreamInput::new(stream_name))
73        .await?;
74
75    Ok(())
76}

examples/caught_up.rs (lines 45-48)

21async fn main() -> Result<(), Box<dyn std::error::Error>> {
22    let access_token =
23        std::env::var("S2_ACCESS_TOKEN").map_err(|_| "S2_ACCESS_TOKEN env var not set")?;
24    let mut config = S2Config::new(access_token);
25    if std::env::var_os("S2_ACCOUNT_ENDPOINT").is_some()
26        || std::env::var_os("S2_BASIN_ENDPOINT").is_some()
27    {
28        config = config.with_endpoints(S2Endpoints::from_env()?);
29    }
30
31    let suffix = &uuid::Uuid::new_v4().simple().to_string()[..8];
32    let basin_name: BasinName = format!("caught-up-{suffix}").parse()?;
33    let stream_name: StreamName = "example".parse()?;
34    let s2 = S2::new(config)?;
35    let basin = s2.basin(basin_name.clone());
36
37    s2.create_basin(CreateBasinInput::new(basin_name.clone()))
38        .await?;
39    basin
40        .create_stream(CreateStreamInput::new(stream_name.clone()))
41        .await?;
42    let stream = basin.stream(stream_name.clone());
43
44    stream
45        .append(AppendInput::new(AppendRecordBatch::try_from_iter([
46            AppendRecord::new("first")?,
47            AppendRecord::new("second")?,
48        ])?))
49        .await?;
50
51    let mut session = stream
52        .read_session(
53            ReadInput::new().with_start(ReadStart::new().with_from(ReadFrom::TailOffset(2))),
54            ReadSessionConfig::default(),
55        )
56        .await?;
57    let mut caught_up = session.caught_up();
58
59    loop {
60        tokio::select! {
61            tail = &mut caught_up => {
62                println!("Caught up through sequence number {}", tail?.seq_num);
63                break;
64            }
65            Some(batch) = session.next() => {
66                print_batch("Read before catching up", &batch?);
67            }
68        }
69    }
70
71    let ack = stream
72        .append(AppendInput::new(AppendRecordBatch::try_from_iter([
73            AppendRecord::new("third")?,
74        ])?))
75        .await?;
76    println!(
77        "Appended another record at sequence number {}",
78        ack.start.seq_num
79    );
80
81    while let Some(batch) = session.next().await {
82        let batch = batch?;
83        print_batch("Read after catching up", &batch);
84        if batch
85            .records
86            .iter()
87            .any(|record| record.seq_num == ack.start.seq_num)
88        {
89            break;
90        }
91    }
92    println!("Session is caught up again: {}", session.is_caught_up());
93
94    drop(session);
95    basin
96        .delete_stream(DeleteStreamInput::new(stream_name))
97        .await?;
98    s2.delete_basin(DeleteBasinInput::new(basin_name)).await?;
99
100    Ok(())
101}

examples/docs_streams.rs (lines 43-46)

20async fn main() -> Result<(), Box<dyn std::error::Error>> {
21    let access_token = std::env::var("S2_ACCESS_TOKEN")?;
22    let basin_name: BasinName = std::env::var("S2_BASIN")?.parse()?;
23
24    let client = S2::new(S2Config::new(access_token))?;
25    let basin = client.basin(basin_name);
26
27    // Create a temporary stream for examples
28    let stream_name: StreamName = format!(
29        "docs-streams-{}",
30        std::time::SystemTime::now()
31            .duration_since(std::time::UNIX_EPOCH)?
32            .as_millis()
33    )
34    .parse()?;
35    basin
36        .create_stream(s2_sdk::types::CreateStreamInput::new(stream_name.clone()))
37        .await?;
38
39    // ANCHOR: simple-append
40    let stream = basin.stream(stream_name.clone());
41
42    let ack = stream
43        .append(AppendInput::new(AppendRecordBatch::try_from_iter([
44            AppendRecord::new("first event")?,
45            AppendRecord::new("second event")?,
46        ])?))
47        .await?;
48
49    // ack tells us where the records landed
50    println!(
51        "Wrote records {} through {}",
52        ack.start.seq_num,
53        ack.end.seq_num - 1
54    );
55    // ANCHOR_END: simple-append
56
57    // ANCHOR: simple-read
58    let batch = stream
59        .read(
60            ReadInput::new()
61                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
62                .with_stop(ReadStop::new().with_limits(ReadLimits::new().with_count(100))),
63        )
64        .await?;
65
66    for record in batch.records {
67        println!("[{}] {:?}", record.seq_num, record.body);
68    }
69    // ANCHOR_END: simple-read
70
71    // ANCHOR: append-session
72    let session = stream.append_session(AppendSessionConfig::new());
73
74    // Submit a batch - this enqueues it and returns a ticket
75    let records = AppendRecordBatch::try_from_iter([
76        AppendRecord::new("event-1")?,
77        AppendRecord::new("event-2")?,
78    ])?;
79    let ticket = session.submit(AppendInput::new(records)).await?;
80
81    // Wait for durability
82    let ack = ticket.await?;
83    println!("Durable at seqNum {}", ack.start.seq_num);
84
85    session.close().await?;
86    // ANCHOR_END: append-session
87
88    // ANCHOR: producer
89    let producer = stream.producer(
90        ProducerConfig::new()
91            .with_batching(BatchingConfig::new().with_linger(Duration::from_millis(5))),
92    );
93
94    // Submit individual records
95    let ticket = producer.submit(AppendRecord::new("my event")?).await?;
96
97    // Get the exact sequence number
98    let ack = ticket.await?;
99    println!("Record durable at seqNum {}", ack.seq_num);
100
101    producer.close().await?;
102    // ANCHOR_END: producer
103
104    // ANCHOR: check-tail
105    let tail = stream.check_tail().await?;
106    println!("Stream has {} records", tail.seq_num);
107    // ANCHOR_END: check-tail
108
109    // Cleanup
110    basin
111        .delete_stream(s2_sdk::types::DeleteStreamInput::new(stream_name))
112        .await?;
113
114    println!("Streams examples completed");
115
116    // The following read session examples are for documentation snippets only.
117    // They are not executed because they would block waiting for new records.
118    if std::env::var("RUN_READ_SESSIONS").is_err() {
119        return Ok(());
120    }
121
122    // ANCHOR: read-session
123    let mut session = stream
124        .read_session(
125            ReadInput::new().with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0))),
126            ReadSessionConfig::default(),
127        )
128        .await?;
129
130    while let Some(batch) = session.next().await {
131        let batch = batch?;
132        for record in batch.records {
133            println!("[{}] {:?}", record.seq_num, record.body);
134        }
135    }
136    // ANCHOR_END: read-session
137
138    // ANCHOR: read-session-tail-offset
139    // Start reading from 10 records before the current tail
140    let mut session = stream
141        .read_session(
142            ReadInput::new().with_start(ReadStart::new().with_from(ReadFrom::TailOffset(10))),
143            ReadSessionConfig::default(),
144        )
145        .await?;
146
147    while let Some(batch) = session.next().await {
148        let batch = batch?;
149        for record in batch.records {
150            println!("[{}] {:?}", record.seq_num, record.body);
151        }
152    }
153    // ANCHOR_END: read-session-tail-offset
154
155    // ANCHOR: read-session-timestamp
156    // Start reading from a specific timestamp
157    let one_hour_ago = std::time::SystemTime::now()
158        .duration_since(std::time::UNIX_EPOCH)?
159        .as_millis() as u64
160        - 3600 * 1000;
161    let mut session = stream
162        .read_session(
163            ReadInput::new()
164                .with_start(ReadStart::new().with_from(ReadFrom::Timestamp(one_hour_ago))),
165            ReadSessionConfig::default(),
166        )
167        .await?;
168
169    while let Some(batch) = session.next().await {
170        let batch = batch?;
171        for record in batch.records {
172            println!("[{}] {:?}", record.seq_num, record.body);
173        }
174    }
175    // ANCHOR_END: read-session-timestamp
176
177    // ANCHOR: read-session-until
178    // Read records until a specific timestamp
179    let one_hour_ago = std::time::SystemTime::now()
180        .duration_since(std::time::UNIX_EPOCH)?
181        .as_millis() as u64
182        - 3600 * 1000;
183    let mut session = stream
184        .read_session(
185            ReadInput::new()
186                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
187                .with_stop(ReadStop::new().with_until(..one_hour_ago)),
188            ReadSessionConfig::default(),
189        )
190        .await?;
191
192    while let Some(batch) = session.next().await {
193        let batch = batch?;
194        for record in batch.records {
195            println!("[{}] {:?}", record.seq_num, record.body);
196        }
197    }
198    // ANCHOR_END: read-session-until
199
200    // ANCHOR: read-session-wait
201    // Read all available records, and once reaching the current tail, wait an additional 30 seconds
202    // for new ones
203    let mut session = stream
204        .read_session(
205            ReadInput::new()
206                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
207                .with_stop(ReadStop::new().with_wait(30)),
208            ReadSessionConfig::default(),
209        )
210        .await?;
211
212    while let Some(batch) = session.next().await {
213        let batch = batch?;
214        for record in batch.records {
215            println!("[{}] {:?}", record.seq_num, record.body);
216        }
217    }
218    // ANCHOR_END: read-session-wait
219
220    Ok(())
221}

Source

pub async fn read(&self, input: ReadInput) -> Result<ReadBatch, ReadError>

Read records.

Examples found in repository ?

examples/get_latest_record.rs (line 25)

9async fn main() -> Result<(), Box<dyn std::error::Error>> {
10    let access_token =
11        std::env::var("S2_ACCESS_TOKEN").map_err(|_| "S2_ACCESS_TOKEN env var not set")?;
12    let basin_name: BasinName = std::env::var("S2_BASIN")
13        .map_err(|_| "S2_BASIN env var not set")?
14        .parse()?;
15    let stream_name: StreamName = std::env::var("S2_STREAM")
16        .map_err(|_| "S2_STREAM env var not set")?
17        .parse()?;
18
19    let s2 = S2::new(S2Config::new(access_token))?;
20    let stream = s2.basin(basin_name).stream(stream_name);
21
22    let input = ReadInput::new()
23        .with_start(ReadStart::new().with_from(ReadFrom::TailOffset(1)))
24        .with_stop(ReadStop::new().with_limits(ReadLimits::new().with_count(1)));
25    let batch = stream.read(input).await?;
26    println!("{batch:#?}");
27
28    Ok(())
29}

More examples

Hide additional examples

examples/docs_encryption.rs (lines 61-65)

15async fn main() -> Result<(), Box<dyn std::error::Error>> {
16    let access_token = std::env::var("S2_ACCESS_TOKEN")?;
17    let basin_name: BasinName = std::env::var("S2_BASIN")?.parse()?;
18    let stream_name: StreamName = format!(
19        "docs-encryption-{}",
20        std::time::SystemTime::now()
21            .duration_since(std::time::UNIX_EPOCH)?
22            .as_millis()
23    )
24    .parse()?;
25
26    let client = S2::new(S2Config::new(access_token))?;
27
28    // ANCHOR: basin-cipher
29    client
30        .create_basin(
31            CreateBasinInput::new(basin_name.clone())
32                .with_config(BasinConfig::new().with_stream_cipher(EncryptionAlgorithm::Aegis256)),
33        )
34        .await?;
35
36    client
37        .reconfigure_basin(ReconfigureBasinInput::new(
38            basin_name.clone(),
39            BasinReconfiguration::new().with_stream_cipher(EncryptionAlgorithm::Aes256Gcm),
40        ))
41        .await?;
42    // ANCHOR_END: basin-cipher
43
44    let basin = client.basin(basin_name.clone());
45    basin
46        .create_stream(CreateStreamInput::new(stream_name.clone()))
47        .await?;
48
49    // ANCHOR: append-read
50    let stream = basin
51        .stream(stream_name.clone())
52        .with_encryption_key(std::env::var("S2_ENCRYPTION_KEY")?.parse()?);
53
54    stream
55        .append(AppendInput::new(AppendRecordBatch::try_from_iter([
56            AppendRecord::new("top secret")?,
57        ])?))
58        .await?;
59
60    let batch = stream
61        .read(
62            ReadInput::new()
63                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
64                .with_stop(ReadStop::new().with_limits(ReadLimits::new().with_count(10))),
65        )
66        .await?;
67    // ANCHOR_END: append-read
68
69    println!("Read {} encrypted record(s)", batch.records.len());
70
71    basin
72        .delete_stream(DeleteStreamInput::new(stream_name))
73        .await?;
74
75    Ok(())
76}

examples/docs_streams.rs (lines 59-63)

20async fn main() -> Result<(), Box<dyn std::error::Error>> {
21    let access_token = std::env::var("S2_ACCESS_TOKEN")?;
22    let basin_name: BasinName = std::env::var("S2_BASIN")?.parse()?;
23
24    let client = S2::new(S2Config::new(access_token))?;
25    let basin = client.basin(basin_name);
26
27    // Create a temporary stream for examples
28    let stream_name: StreamName = format!(
29        "docs-streams-{}",
30        std::time::SystemTime::now()
31            .duration_since(std::time::UNIX_EPOCH)?
32            .as_millis()
33    )
34    .parse()?;
35    basin
36        .create_stream(s2_sdk::types::CreateStreamInput::new(stream_name.clone()))
37        .await?;
38
39    // ANCHOR: simple-append
40    let stream = basin.stream(stream_name.clone());
41
42    let ack = stream
43        .append(AppendInput::new(AppendRecordBatch::try_from_iter([
44            AppendRecord::new("first event")?,
45            AppendRecord::new("second event")?,
46        ])?))
47        .await?;
48
49    // ack tells us where the records landed
50    println!(
51        "Wrote records {} through {}",
52        ack.start.seq_num,
53        ack.end.seq_num - 1
54    );
55    // ANCHOR_END: simple-append
56
57    // ANCHOR: simple-read
58    let batch = stream
59        .read(
60            ReadInput::new()
61                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
62                .with_stop(ReadStop::new().with_limits(ReadLimits::new().with_count(100))),
63        )
64        .await?;
65
66    for record in batch.records {
67        println!("[{}] {:?}", record.seq_num, record.body);
68    }
69    // ANCHOR_END: simple-read
70
71    // ANCHOR: append-session
72    let session = stream.append_session(AppendSessionConfig::new());
73
74    // Submit a batch - this enqueues it and returns a ticket
75    let records = AppendRecordBatch::try_from_iter([
76        AppendRecord::new("event-1")?,
77        AppendRecord::new("event-2")?,
78    ])?;
79    let ticket = session.submit(AppendInput::new(records)).await?;
80
81    // Wait for durability
82    let ack = ticket.await?;
83    println!("Durable at seqNum {}", ack.start.seq_num);
84
85    session.close().await?;
86    // ANCHOR_END: append-session
87
88    // ANCHOR: producer
89    let producer = stream.producer(
90        ProducerConfig::new()
91            .with_batching(BatchingConfig::new().with_linger(Duration::from_millis(5))),
92    );
93
94    // Submit individual records
95    let ticket = producer.submit(AppendRecord::new("my event")?).await?;
96
97    // Get the exact sequence number
98    let ack = ticket.await?;
99    println!("Record durable at seqNum {}", ack.seq_num);
100
101    producer.close().await?;
102    // ANCHOR_END: producer
103
104    // ANCHOR: check-tail
105    let tail = stream.check_tail().await?;
106    println!("Stream has {} records", tail.seq_num);
107    // ANCHOR_END: check-tail
108
109    // Cleanup
110    basin
111        .delete_stream(s2_sdk::types::DeleteStreamInput::new(stream_name))
112        .await?;
113
114    println!("Streams examples completed");
115
116    // The following read session examples are for documentation snippets only.
117    // They are not executed because they would block waiting for new records.
118    if std::env::var("RUN_READ_SESSIONS").is_err() {
119        return Ok(());
120    }
121
122    // ANCHOR: read-session
123    let mut session = stream
124        .read_session(
125            ReadInput::new().with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0))),
126            ReadSessionConfig::default(),
127        )
128        .await?;
129
130    while let Some(batch) = session.next().await {
131        let batch = batch?;
132        for record in batch.records {
133            println!("[{}] {:?}", record.seq_num, record.body);
134        }
135    }
136    // ANCHOR_END: read-session
137
138    // ANCHOR: read-session-tail-offset
139    // Start reading from 10 records before the current tail
140    let mut session = stream
141        .read_session(
142            ReadInput::new().with_start(ReadStart::new().with_from(ReadFrom::TailOffset(10))),
143            ReadSessionConfig::default(),
144        )
145        .await?;
146
147    while let Some(batch) = session.next().await {
148        let batch = batch?;
149        for record in batch.records {
150            println!("[{}] {:?}", record.seq_num, record.body);
151        }
152    }
153    // ANCHOR_END: read-session-tail-offset
154
155    // ANCHOR: read-session-timestamp
156    // Start reading from a specific timestamp
157    let one_hour_ago = std::time::SystemTime::now()
158        .duration_since(std::time::UNIX_EPOCH)?
159        .as_millis() as u64
160        - 3600 * 1000;
161    let mut session = stream
162        .read_session(
163            ReadInput::new()
164                .with_start(ReadStart::new().with_from(ReadFrom::Timestamp(one_hour_ago))),
165            ReadSessionConfig::default(),
166        )
167        .await?;
168
169    while let Some(batch) = session.next().await {
170        let batch = batch?;
171        for record in batch.records {
172            println!("[{}] {:?}", record.seq_num, record.body);
173        }
174    }
175    // ANCHOR_END: read-session-timestamp
176
177    // ANCHOR: read-session-until
178    // Read records until a specific timestamp
179    let one_hour_ago = std::time::SystemTime::now()
180        .duration_since(std::time::UNIX_EPOCH)?
181        .as_millis() as u64
182        - 3600 * 1000;
183    let mut session = stream
184        .read_session(
185            ReadInput::new()
186                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
187                .with_stop(ReadStop::new().with_until(..one_hour_ago)),
188            ReadSessionConfig::default(),
189        )
190        .await?;
191
192    while let Some(batch) = session.next().await {
193        let batch = batch?;
194        for record in batch.records {
195            println!("[{}] {:?}", record.seq_num, record.body);
196        }
197    }
198    // ANCHOR_END: read-session-until
199
200    // ANCHOR: read-session-wait
201    // Read all available records, and once reaching the current tail, wait an additional 30 seconds
202    // for new ones
203    let mut session = stream
204        .read_session(
205            ReadInput::new()
206                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
207                .with_stop(ReadStop::new().with_wait(30)),
208            ReadSessionConfig::default(),
209        )
210        .await?;
211
212    while let Some(batch) = session.next().await {
213        let batch = batch?;
214        for record in batch.records {
215            println!("[{}] {:?}", record.seq_num, record.body);
216        }
217    }
218    // ANCHOR_END: read-session-wait
219
220    Ok(())
221}

Source

pub fn append_session(&self, config: AppendSessionConfig) -> AppendSession

Create an append session for submitting AppendInputs.

Examples found in repository ?

examples/docs_streams.rs (line 72)

20async fn main() -> Result<(), Box<dyn std::error::Error>> {
21    let access_token = std::env::var("S2_ACCESS_TOKEN")?;
22    let basin_name: BasinName = std::env::var("S2_BASIN")?.parse()?;
23
24    let client = S2::new(S2Config::new(access_token))?;
25    let basin = client.basin(basin_name);
26
27    // Create a temporary stream for examples
28    let stream_name: StreamName = format!(
29        "docs-streams-{}",
30        std::time::SystemTime::now()
31            .duration_since(std::time::UNIX_EPOCH)?
32            .as_millis()
33    )
34    .parse()?;
35    basin
36        .create_stream(s2_sdk::types::CreateStreamInput::new(stream_name.clone()))
37        .await?;
38
39    // ANCHOR: simple-append
40    let stream = basin.stream(stream_name.clone());
41
42    let ack = stream
43        .append(AppendInput::new(AppendRecordBatch::try_from_iter([
44            AppendRecord::new("first event")?,
45            AppendRecord::new("second event")?,
46        ])?))
47        .await?;
48
49    // ack tells us where the records landed
50    println!(
51        "Wrote records {} through {}",
52        ack.start.seq_num,
53        ack.end.seq_num - 1
54    );
55    // ANCHOR_END: simple-append
56
57    // ANCHOR: simple-read
58    let batch = stream
59        .read(
60            ReadInput::new()
61                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
62                .with_stop(ReadStop::new().with_limits(ReadLimits::new().with_count(100))),
63        )
64        .await?;
65
66    for record in batch.records {
67        println!("[{}] {:?}", record.seq_num, record.body);
68    }
69    // ANCHOR_END: simple-read
70
71    // ANCHOR: append-session
72    let session = stream.append_session(AppendSessionConfig::new());
73
74    // Submit a batch - this enqueues it and returns a ticket
75    let records = AppendRecordBatch::try_from_iter([
76        AppendRecord::new("event-1")?,
77        AppendRecord::new("event-2")?,
78    ])?;
79    let ticket = session.submit(AppendInput::new(records)).await?;
80
81    // Wait for durability
82    let ack = ticket.await?;
83    println!("Durable at seqNum {}", ack.start.seq_num);
84
85    session.close().await?;
86    // ANCHOR_END: append-session
87
88    // ANCHOR: producer
89    let producer = stream.producer(
90        ProducerConfig::new()
91            .with_batching(BatchingConfig::new().with_linger(Duration::from_millis(5))),
92    );
93
94    // Submit individual records
95    let ticket = producer.submit(AppendRecord::new("my event")?).await?;
96
97    // Get the exact sequence number
98    let ack = ticket.await?;
99    println!("Record durable at seqNum {}", ack.seq_num);
100
101    producer.close().await?;
102    // ANCHOR_END: producer
103
104    // ANCHOR: check-tail
105    let tail = stream.check_tail().await?;
106    println!("Stream has {} records", tail.seq_num);
107    // ANCHOR_END: check-tail
108
109    // Cleanup
110    basin
111        .delete_stream(s2_sdk::types::DeleteStreamInput::new(stream_name))
112        .await?;
113
114    println!("Streams examples completed");
115
116    // The following read session examples are for documentation snippets only.
117    // They are not executed because they would block waiting for new records.
118    if std::env::var("RUN_READ_SESSIONS").is_err() {
119        return Ok(());
120    }
121
122    // ANCHOR: read-session
123    let mut session = stream
124        .read_session(
125            ReadInput::new().with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0))),
126            ReadSessionConfig::default(),
127        )
128        .await?;
129
130    while let Some(batch) = session.next().await {
131        let batch = batch?;
132        for record in batch.records {
133            println!("[{}] {:?}", record.seq_num, record.body);
134        }
135    }
136    // ANCHOR_END: read-session
137
138    // ANCHOR: read-session-tail-offset
139    // Start reading from 10 records before the current tail
140    let mut session = stream
141        .read_session(
142            ReadInput::new().with_start(ReadStart::new().with_from(ReadFrom::TailOffset(10))),
143            ReadSessionConfig::default(),
144        )
145        .await?;
146
147    while let Some(batch) = session.next().await {
148        let batch = batch?;
149        for record in batch.records {
150            println!("[{}] {:?}", record.seq_num, record.body);
151        }
152    }
153    // ANCHOR_END: read-session-tail-offset
154
155    // ANCHOR: read-session-timestamp
156    // Start reading from a specific timestamp
157    let one_hour_ago = std::time::SystemTime::now()
158        .duration_since(std::time::UNIX_EPOCH)?
159        .as_millis() as u64
160        - 3600 * 1000;
161    let mut session = stream
162        .read_session(
163            ReadInput::new()
164                .with_start(ReadStart::new().with_from(ReadFrom::Timestamp(one_hour_ago))),
165            ReadSessionConfig::default(),
166        )
167        .await?;
168
169    while let Some(batch) = session.next().await {
170        let batch = batch?;
171        for record in batch.records {
172            println!("[{}] {:?}", record.seq_num, record.body);
173        }
174    }
175    // ANCHOR_END: read-session-timestamp
176
177    // ANCHOR: read-session-until
178    // Read records until a specific timestamp
179    let one_hour_ago = std::time::SystemTime::now()
180        .duration_since(std::time::UNIX_EPOCH)?
181        .as_millis() as u64
182        - 3600 * 1000;
183    let mut session = stream
184        .read_session(
185            ReadInput::new()
186                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
187                .with_stop(ReadStop::new().with_until(..one_hour_ago)),
188            ReadSessionConfig::default(),
189        )
190        .await?;
191
192    while let Some(batch) = session.next().await {
193        let batch = batch?;
194        for record in batch.records {
195            println!("[{}] {:?}", record.seq_num, record.body);
196        }
197    }
198    // ANCHOR_END: read-session-until
199
200    // ANCHOR: read-session-wait
201    // Read all available records, and once reaching the current tail, wait an additional 30 seconds
202    // for new ones
203    let mut session = stream
204        .read_session(
205            ReadInput::new()
206                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
207                .with_stop(ReadStop::new().with_wait(30)),
208            ReadSessionConfig::default(),
209        )
210        .await?;
211
212    while let Some(batch) = session.next().await {
213        let batch = batch?;
214        for record in batch.records {
215            println!("[{}] {:?}", record.seq_num, record.body);
216        }
217    }
218    // ANCHOR_END: read-session-wait
219
220    Ok(())
221}

Source

pub fn producer(&self, config: ProducerConfig) -> Producer

Create a producer for submitting individual AppendRecords.

Examples found in repository ?

examples/producer.rs (line 21)

8async fn main() -> Result<(), Box<dyn std::error::Error>> {
9    let access_token =
10        std::env::var("S2_ACCESS_TOKEN").map_err(|_| "S2_ACCESS_TOKEN env var not set")?;
11    let basin_name: BasinName = std::env::var("S2_BASIN")
12        .map_err(|_| "S2_BASIN env var not set")?
13        .parse()?;
14    let stream_name: StreamName = std::env::var("S2_STREAM")
15        .map_err(|_| "S2_STREAM env var not set")?
16        .parse()?;
17
18    let s2 = S2::new(S2Config::new(access_token))?;
19    let stream = s2.basin(basin_name).stream(stream_name);
20
21    let producer = stream.producer(ProducerConfig::new());
22
23    let ticket1 = producer.submit(AppendRecord::new("lorem")?).await?;
24    let ticket2 = producer.submit(AppendRecord::new("ipsum")?).await?;
25
26    let ack1 = ticket1.await?;
27    let ack2 = ticket2.await?;
28    println!("Record 1 seq_num: {}", ack1.seq_num);
29    println!("Record 2 seq_num: {}", ack2.seq_num);
30
31    producer.close().await?;
32
33    Ok(())
34}

More examples

Hide additional examples

examples/docs_streams.rs (lines 89-92)

20async fn main() -> Result<(), Box<dyn std::error::Error>> {
21    let access_token = std::env::var("S2_ACCESS_TOKEN")?;
22    let basin_name: BasinName = std::env::var("S2_BASIN")?.parse()?;
23
24    let client = S2::new(S2Config::new(access_token))?;
25    let basin = client.basin(basin_name);
26
27    // Create a temporary stream for examples
28    let stream_name: StreamName = format!(
29        "docs-streams-{}",
30        std::time::SystemTime::now()
31            .duration_since(std::time::UNIX_EPOCH)?
32            .as_millis()
33    )
34    .parse()?;
35    basin
36        .create_stream(s2_sdk::types::CreateStreamInput::new(stream_name.clone()))
37        .await?;
38
39    // ANCHOR: simple-append
40    let stream = basin.stream(stream_name.clone());
41
42    let ack = stream
43        .append(AppendInput::new(AppendRecordBatch::try_from_iter([
44            AppendRecord::new("first event")?,
45            AppendRecord::new("second event")?,
46        ])?))
47        .await?;
48
49    // ack tells us where the records landed
50    println!(
51        "Wrote records {} through {}",
52        ack.start.seq_num,
53        ack.end.seq_num - 1
54    );
55    // ANCHOR_END: simple-append
56
57    // ANCHOR: simple-read
58    let batch = stream
59        .read(
60            ReadInput::new()
61                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
62                .with_stop(ReadStop::new().with_limits(ReadLimits::new().with_count(100))),
63        )
64        .await?;
65
66    for record in batch.records {
67        println!("[{}] {:?}", record.seq_num, record.body);
68    }
69    // ANCHOR_END: simple-read
70
71    // ANCHOR: append-session
72    let session = stream.append_session(AppendSessionConfig::new());
73
74    // Submit a batch - this enqueues it and returns a ticket
75    let records = AppendRecordBatch::try_from_iter([
76        AppendRecord::new("event-1")?,
77        AppendRecord::new("event-2")?,
78    ])?;
79    let ticket = session.submit(AppendInput::new(records)).await?;
80
81    // Wait for durability
82    let ack = ticket.await?;
83    println!("Durable at seqNum {}", ack.start.seq_num);
84
85    session.close().await?;
86    // ANCHOR_END: append-session
87
88    // ANCHOR: producer
89    let producer = stream.producer(
90        ProducerConfig::new()
91            .with_batching(BatchingConfig::new().with_linger(Duration::from_millis(5))),
92    );
93
94    // Submit individual records
95    let ticket = producer.submit(AppendRecord::new("my event")?).await?;
96
97    // Get the exact sequence number
98    let ack = ticket.await?;
99    println!("Record durable at seqNum {}", ack.seq_num);
100
101    producer.close().await?;
102    // ANCHOR_END: producer
103
104    // ANCHOR: check-tail
105    let tail = stream.check_tail().await?;
106    println!("Stream has {} records", tail.seq_num);
107    // ANCHOR_END: check-tail
108
109    // Cleanup
110    basin
111        .delete_stream(s2_sdk::types::DeleteStreamInput::new(stream_name))
112        .await?;
113
114    println!("Streams examples completed");
115
116    // The following read session examples are for documentation snippets only.
117    // They are not executed because they would block waiting for new records.
118    if std::env::var("RUN_READ_SESSIONS").is_err() {
119        return Ok(());
120    }
121
122    // ANCHOR: read-session
123    let mut session = stream
124        .read_session(
125            ReadInput::new().with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0))),
126            ReadSessionConfig::default(),
127        )
128        .await?;
129
130    while let Some(batch) = session.next().await {
131        let batch = batch?;
132        for record in batch.records {
133            println!("[{}] {:?}", record.seq_num, record.body);
134        }
135    }
136    // ANCHOR_END: read-session
137
138    // ANCHOR: read-session-tail-offset
139    // Start reading from 10 records before the current tail
140    let mut session = stream
141        .read_session(
142            ReadInput::new().with_start(ReadStart::new().with_from(ReadFrom::TailOffset(10))),
143            ReadSessionConfig::default(),
144        )
145        .await?;
146
147    while let Some(batch) = session.next().await {
148        let batch = batch?;
149        for record in batch.records {
150            println!("[{}] {:?}", record.seq_num, record.body);
151        }
152    }
153    // ANCHOR_END: read-session-tail-offset
154
155    // ANCHOR: read-session-timestamp
156    // Start reading from a specific timestamp
157    let one_hour_ago = std::time::SystemTime::now()
158        .duration_since(std::time::UNIX_EPOCH)?
159        .as_millis() as u64
160        - 3600 * 1000;
161    let mut session = stream
162        .read_session(
163            ReadInput::new()
164                .with_start(ReadStart::new().with_from(ReadFrom::Timestamp(one_hour_ago))),
165            ReadSessionConfig::default(),
166        )
167        .await?;
168
169    while let Some(batch) = session.next().await {
170        let batch = batch?;
171        for record in batch.records {
172            println!("[{}] {:?}", record.seq_num, record.body);
173        }
174    }
175    // ANCHOR_END: read-session-timestamp
176
177    // ANCHOR: read-session-until
178    // Read records until a specific timestamp
179    let one_hour_ago = std::time::SystemTime::now()
180        .duration_since(std::time::UNIX_EPOCH)?
181        .as_millis() as u64
182        - 3600 * 1000;
183    let mut session = stream
184        .read_session(
185            ReadInput::new()
186                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
187                .with_stop(ReadStop::new().with_until(..one_hour_ago)),
188            ReadSessionConfig::default(),
189        )
190        .await?;
191
192    while let Some(batch) = session.next().await {
193        let batch = batch?;
194        for record in batch.records {
195            println!("[{}] {:?}", record.seq_num, record.body);
196        }
197    }
198    // ANCHOR_END: read-session-until
199
200    // ANCHOR: read-session-wait
201    // Read all available records, and once reaching the current tail, wait an additional 30 seconds
202    // for new ones
203    let mut session = stream
204        .read_session(
205            ReadInput::new()
206                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
207                .with_stop(ReadStop::new().with_wait(30)),
208            ReadSessionConfig::default(),
209        )
210        .await?;
211
212    while let Some(batch) = session.next().await {
213        let batch = batch?;
214        for record in batch.records {
215            println!("[{}] {:?}", record.seq_num, record.body);
216        }
217    }
218    // ANCHOR_END: read-session-wait
219
220    Ok(())
221}

Source

pub async fn read_session( &self, input: ReadInput, config: ReadSessionConfig, ) -> Result<ReadSession, ReadSessionError>

Create a read session.

Examples found in repository ?

examples/consumer.rs (line 19)

9async fn main() -> Result<(), Box<dyn std::error::Error>> {
10    let access_token = std::env::var("S2_ACCESS_TOKEN")?;
11    let basin_name: BasinName = std::env::var("S2_BASIN")?.parse()?;
12    let stream_name: StreamName = std::env::var("S2_STREAM")?.parse()?;
13
14    let s2 = S2::new(S2Config::new(access_token))?;
15    let stream = s2.basin(basin_name).stream(stream_name);
16
17    let input = ReadInput::new();
18    let mut batches = stream
19        .read_session(input, ReadSessionConfig::default())
20        .await?;
21    loop {
22        select! {
23            batch = batches.next() => {
24                let Some(batch) = batch else { break };
25                let batch = batch?;
26                println!("{batch:?}");
27            }
28            _ = tokio::signal::ctrl_c() => break,
29        }
30    }
31
32    Ok(())
33}

More examples

Hide additional examples

examples/caught_up.rs (lines 52-55)

21async fn main() -> Result<(), Box<dyn std::error::Error>> {
22    let access_token =
23        std::env::var("S2_ACCESS_TOKEN").map_err(|_| "S2_ACCESS_TOKEN env var not set")?;
24    let mut config = S2Config::new(access_token);
25    if std::env::var_os("S2_ACCOUNT_ENDPOINT").is_some()
26        || std::env::var_os("S2_BASIN_ENDPOINT").is_some()
27    {
28        config = config.with_endpoints(S2Endpoints::from_env()?);
29    }
30
31    let suffix = &uuid::Uuid::new_v4().simple().to_string()[..8];
32    let basin_name: BasinName = format!("caught-up-{suffix}").parse()?;
33    let stream_name: StreamName = "example".parse()?;
34    let s2 = S2::new(config)?;
35    let basin = s2.basin(basin_name.clone());
36
37    s2.create_basin(CreateBasinInput::new(basin_name.clone()))
38        .await?;
39    basin
40        .create_stream(CreateStreamInput::new(stream_name.clone()))
41        .await?;
42    let stream = basin.stream(stream_name.clone());
43
44    stream
45        .append(AppendInput::new(AppendRecordBatch::try_from_iter([
46            AppendRecord::new("first")?,
47            AppendRecord::new("second")?,
48        ])?))
49        .await?;
50
51    let mut session = stream
52        .read_session(
53            ReadInput::new().with_start(ReadStart::new().with_from(ReadFrom::TailOffset(2))),
54            ReadSessionConfig::default(),
55        )
56        .await?;
57    let mut caught_up = session.caught_up();
58
59    loop {
60        tokio::select! {
61            tail = &mut caught_up => {
62                println!("Caught up through sequence number {}", tail?.seq_num);
63                break;
64            }
65            Some(batch) = session.next() => {
66                print_batch("Read before catching up", &batch?);
67            }
68        }
69    }
70
71    let ack = stream
72        .append(AppendInput::new(AppendRecordBatch::try_from_iter([
73            AppendRecord::new("third")?,
74        ])?))
75        .await?;
76    println!(
77        "Appended another record at sequence number {}",
78        ack.start.seq_num
79    );
80
81    while let Some(batch) = session.next().await {
82        let batch = batch?;
83        print_batch("Read after catching up", &batch);
84        if batch
85            .records
86            .iter()
87            .any(|record| record.seq_num == ack.start.seq_num)
88        {
89            break;
90        }
91    }
92    println!("Session is caught up again: {}", session.is_caught_up());
93
94    drop(session);
95    basin
96        .delete_stream(DeleteStreamInput::new(stream_name))
97        .await?;
98    s2.delete_basin(DeleteBasinInput::new(basin_name)).await?;
99
100    Ok(())
101}

examples/docs_streams.rs (lines 124-127)

20async fn main() -> Result<(), Box<dyn std::error::Error>> {
21    let access_token = std::env::var("S2_ACCESS_TOKEN")?;
22    let basin_name: BasinName = std::env::var("S2_BASIN")?.parse()?;
23
24    let client = S2::new(S2Config::new(access_token))?;
25    let basin = client.basin(basin_name);
26
27    // Create a temporary stream for examples
28    let stream_name: StreamName = format!(
29        "docs-streams-{}",
30        std::time::SystemTime::now()
31            .duration_since(std::time::UNIX_EPOCH)?
32            .as_millis()
33    )
34    .parse()?;
35    basin
36        .create_stream(s2_sdk::types::CreateStreamInput::new(stream_name.clone()))
37        .await?;
38
39    // ANCHOR: simple-append
40    let stream = basin.stream(stream_name.clone());
41
42    let ack = stream
43        .append(AppendInput::new(AppendRecordBatch::try_from_iter([
44            AppendRecord::new("first event")?,
45            AppendRecord::new("second event")?,
46        ])?))
47        .await?;
48
49    // ack tells us where the records landed
50    println!(
51        "Wrote records {} through {}",
52        ack.start.seq_num,
53        ack.end.seq_num - 1
54    );
55    // ANCHOR_END: simple-append
56
57    // ANCHOR: simple-read
58    let batch = stream
59        .read(
60            ReadInput::new()
61                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
62                .with_stop(ReadStop::new().with_limits(ReadLimits::new().with_count(100))),
63        )
64        .await?;
65
66    for record in batch.records {
67        println!("[{}] {:?}", record.seq_num, record.body);
68    }
69    // ANCHOR_END: simple-read
70
71    // ANCHOR: append-session
72    let session = stream.append_session(AppendSessionConfig::new());
73
74    // Submit a batch - this enqueues it and returns a ticket
75    let records = AppendRecordBatch::try_from_iter([
76        AppendRecord::new("event-1")?,
77        AppendRecord::new("event-2")?,
78    ])?;
79    let ticket = session.submit(AppendInput::new(records)).await?;
80
81    // Wait for durability
82    let ack = ticket.await?;
83    println!("Durable at seqNum {}", ack.start.seq_num);
84
85    session.close().await?;
86    // ANCHOR_END: append-session
87
88    // ANCHOR: producer
89    let producer = stream.producer(
90        ProducerConfig::new()
91            .with_batching(BatchingConfig::new().with_linger(Duration::from_millis(5))),
92    );
93
94    // Submit individual records
95    let ticket = producer.submit(AppendRecord::new("my event")?).await?;
96
97    // Get the exact sequence number
98    let ack = ticket.await?;
99    println!("Record durable at seqNum {}", ack.seq_num);
100
101    producer.close().await?;
102    // ANCHOR_END: producer
103
104    // ANCHOR: check-tail
105    let tail = stream.check_tail().await?;
106    println!("Stream has {} records", tail.seq_num);
107    // ANCHOR_END: check-tail
108
109    // Cleanup
110    basin
111        .delete_stream(s2_sdk::types::DeleteStreamInput::new(stream_name))
112        .await?;
113
114    println!("Streams examples completed");
115
116    // The following read session examples are for documentation snippets only.
117    // They are not executed because they would block waiting for new records.
118    if std::env::var("RUN_READ_SESSIONS").is_err() {
119        return Ok(());
120    }
121
122    // ANCHOR: read-session
123    let mut session = stream
124        .read_session(
125            ReadInput::new().with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0))),
126            ReadSessionConfig::default(),
127        )
128        .await?;
129
130    while let Some(batch) = session.next().await {
131        let batch = batch?;
132        for record in batch.records {
133            println!("[{}] {:?}", record.seq_num, record.body);
134        }
135    }
136    // ANCHOR_END: read-session
137
138    // ANCHOR: read-session-tail-offset
139    // Start reading from 10 records before the current tail
140    let mut session = stream
141        .read_session(
142            ReadInput::new().with_start(ReadStart::new().with_from(ReadFrom::TailOffset(10))),
143            ReadSessionConfig::default(),
144        )
145        .await?;
146
147    while let Some(batch) = session.next().await {
148        let batch = batch?;
149        for record in batch.records {
150            println!("[{}] {:?}", record.seq_num, record.body);
151        }
152    }
153    // ANCHOR_END: read-session-tail-offset
154
155    // ANCHOR: read-session-timestamp
156    // Start reading from a specific timestamp
157    let one_hour_ago = std::time::SystemTime::now()
158        .duration_since(std::time::UNIX_EPOCH)?
159        .as_millis() as u64
160        - 3600 * 1000;
161    let mut session = stream
162        .read_session(
163            ReadInput::new()
164                .with_start(ReadStart::new().with_from(ReadFrom::Timestamp(one_hour_ago))),
165            ReadSessionConfig::default(),
166        )
167        .await?;
168
169    while let Some(batch) = session.next().await {
170        let batch = batch?;
171        for record in batch.records {
172            println!("[{}] {:?}", record.seq_num, record.body);
173        }
174    }
175    // ANCHOR_END: read-session-timestamp
176
177    // ANCHOR: read-session-until
178    // Read records until a specific timestamp
179    let one_hour_ago = std::time::SystemTime::now()
180        .duration_since(std::time::UNIX_EPOCH)?
181        .as_millis() as u64
182        - 3600 * 1000;
183    let mut session = stream
184        .read_session(
185            ReadInput::new()
186                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
187                .with_stop(ReadStop::new().with_until(..one_hour_ago)),
188            ReadSessionConfig::default(),
189        )
190        .await?;
191
192    while let Some(batch) = session.next().await {
193        let batch = batch?;
194        for record in batch.records {
195            println!("[{}] {:?}", record.seq_num, record.body);
196        }
197    }
198    // ANCHOR_END: read-session-until
199
200    // ANCHOR: read-session-wait
201    // Read all available records, and once reaching the current tail, wait an additional 30 seconds
202    // for new ones
203    let mut session = stream
204        .read_session(
205            ReadInput::new()
206                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
207                .with_stop(ReadStop::new().with_wait(30)),
208            ReadSessionConfig::default(),
209        )
210        .await?;
211
212    while let Some(batch) = session.next().await {
213        let batch = batch?;
214        for record in batch.records {
215            println!("[{}] {:?}", record.seq_num, record.body);
216        }
217    }
218    // ANCHOR_END: read-session-wait
219
220    Ok(())
221}

Trait Implementations§

Source §

impl Clone for S2Stream

Source §

fn clone(&self) -> S2Stream

Returns a duplicate of the value. Read more

1.0.0 (const: unstable) · Source§

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

Performs copy-assignment from source. Read more

Source §

impl Debug for S2Stream

Source §

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

Formats the value using the given formatter. Read more

Auto Trait Implementations§

§

impl UnsafeUnpin for S2Stream

Blanket Implementations§

Source §

impl<T> Any for T
where T: 'static + ?Sized,

Source §

fn type_id(&self) -> TypeId

Gets the TypeId of self. Read more

Source §

impl<T> Borrow<T> for T
where T: ?Sized,

Source §

fn borrow(&self) -> &T

Immutably borrows from an owned value. Read more

Source §

impl<T> BorrowMut<T> for T
where T: ?Sized,

Source §

fn borrow_mut(&mut self) -> &mut T

Mutably borrows from an owned value. Read more

Source §

impl<T> CloneToUninit for T
where T: Clone,

Source §

unsafe fn clone_to_uninit(&self, dest: *mut u8)

🔬This is a nightly-only experimental API. (clone_to_uninit)

Performs copy-assignment from self to dest. Read more

Source §

impl<T> From<T> for T

Source §

fn from(t: T) -> T

Returns the argument unchanged.

Source §

impl<T> Instrument for T

Source §

fn instrument(self, span: Span) -> Instrumented<Self>

Instruments this type with the provided Span, returning an Instrumented wrapper. Read more

Source §

fn in_current_span(self) -> Instrumented<Self>

Instruments this type with the current Span, returning an Instrumented wrapper. Read more

Source §

impl<T, U> Into for T
where U: From<T>,

Source §

fn into(self) -> U

Calls U::from(self).

That is, this conversion is whatever the implementation of From<T> for U chooses to do.

Source §

impl<T> IntoEither for T

Source §

fn into_either(self, into_left: bool) -> Either<Self, Self>

Converts self into a Left variant of Either<Self, Self> if into_left is true. Converts self into a Right variant of Either<Self, Self> otherwise. Read more

Source §

fn into_either_with<F>(self, into_left: F) -> Either<Self, Self>
where F: FnOnce(&Self) -> bool,

Converts self into a Left variant of Either<Self, Self> if into_left(&self) returns true. Converts self into a Right variant of Either<Self, Self> otherwise. Read more

Source §