Struct S2Basin

Source

pub struct S2Basin { /* private fields */ }

Expand description

A basin in an S2 account.

See S2::basin.

Implementations§

Source §

impl S2Basin

Source

pub fn stream(&self, name: StreamName) -> S2Stream

Get an S2Stream.

Examples found in repository ?

examples/docs_overview.rs (line 12)

5fn main() -> Result<(), Box<dyn std::error::Error>> {
6    // ANCHOR: create-client
7    use s2_sdk::{S2, types::S2Config};
8
9    let client = S2::new(S2Config::new(std::env::var("S2_ACCESS_TOKEN")?))?;
10
11    let basin = client.basin("my-basin".parse()?);
12    let stream = basin.stream("my-stream".parse()?);
13    // ANCHOR_END: create-client
14
15    println!("Created client for stream: {:?}", stream);
16    Ok(())
17}

More examples

Hide additional examples

examples/consumer.rs (line 15)

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.read_session(input).await?;
19    loop {
20        select! {
21            batch = batches.next() => {
22                let Some(batch) = batch else { break };
23                let batch = batch?;
24                println!("{batch:?}");
25            }
26            _ = tokio::signal::ctrl_c() => break,
27        }
28    }
29
30    Ok(())
31}

examples/get_latest_record.rs (line 20)

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}

examples/explicit_trim.rs (line 18)

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}

examples/producer.rs (line 19)

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}

examples/docs_streams.rs (line 40)

20async fn main() -> Result<(), Box<dyn std::error::Error>> {
21    let 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(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 records = AppendRecordBatch::try_from_iter([
43        AppendRecord::new("first event")?,
44        AppendRecord::new("second event")?,
45    ])?;
46
47    let ack = stream.append(AppendInput::new(records)).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(ReadInput::new().with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0))))
125        .await?;
126
127    while let Some(batch) = session.next().await {
128        let batch = batch?;
129        for record in batch.records {
130            println!("[{}] {:?}", record.seq_num, record.body);
131        }
132    }
133    // ANCHOR_END: read-session
134
135    // ANCHOR: read-session-tail-offset
136    // Start reading from 10 records before the current tail
137    let mut session = stream
138        .read_session(
139            ReadInput::new().with_start(ReadStart::new().with_from(ReadFrom::TailOffset(10))),
140        )
141        .await?;
142
143    while let Some(batch) = session.next().await {
144        let batch = batch?;
145        for record in batch.records {
146            println!("[{}] {:?}", record.seq_num, record.body);
147        }
148    }
149    // ANCHOR_END: read-session-tail-offset
150
151    // ANCHOR: read-session-timestamp
152    // Start reading from a specific timestamp
153    let one_hour_ago = std::time::SystemTime::now()
154        .duration_since(std::time::UNIX_EPOCH)?
155        .as_millis() as u64
156        - 3600 * 1000;
157    let mut session = stream
158        .read_session(
159            ReadInput::new()
160                .with_start(ReadStart::new().with_from(ReadFrom::Timestamp(one_hour_ago))),
161        )
162        .await?;
163
164    while let Some(batch) = session.next().await {
165        let batch = batch?;
166        for record in batch.records {
167            println!("[{}] {:?}", record.seq_num, record.body);
168        }
169    }
170    // ANCHOR_END: read-session-timestamp
171
172    // ANCHOR: read-session-until
173    // Read records until a specific timestamp
174    let one_hour_ago = std::time::SystemTime::now()
175        .duration_since(std::time::UNIX_EPOCH)?
176        .as_millis() as u64
177        - 3600 * 1000;
178    let mut session = stream
179        .read_session(
180            ReadInput::new()
181                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
182                .with_stop(ReadStop::new().with_until(..one_hour_ago)),
183        )
184        .await?;
185
186    while let Some(batch) = session.next().await {
187        let batch = batch?;
188        for record in batch.records {
189            println!("[{}] {:?}", record.seq_num, record.body);
190        }
191    }
192    // ANCHOR_END: read-session-until
193
194    // ANCHOR: read-session-wait
195    // Read all available records, and once reaching the current tail, wait an additional 30 seconds
196    // for new ones
197    let mut session = stream
198        .read_session(
199            ReadInput::new()
200                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
201                .with_stop(ReadStop::new().with_wait(30)),
202        )
203        .await?;
204
205    while let Some(batch) = session.next().await {
206        let batch = batch?;
207        for record in batch.records {
208            println!("[{}] {:?}", record.seq_num, record.body);
209        }
210    }
211    // ANCHOR_END: read-session-wait
212
213    Ok(())
214}

Source

pub async fn list_streams( &self, input: ListStreamsInput, ) -> Result<Page<StreamInfo>, S2Error>

List a page of streams.

See list_all_streams for automatic pagination.

Examples found in repository ?

examples/list_streams.rs (line 18)

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
14    let s2 = S2::new(S2Config::new(access_token))?;
15    let basin = s2.basin(basin_name);
16
17    let input = ListStreamsInput::new().with_prefix("my-".parse()?);
18    let page = basin.list_streams(input).await?;
19    println!("{page:#?}");
20
21    Ok(())
22}

More examples

Hide additional examples

examples/docs_account_and_basins.rs (line 47)

17async fn main() -> Result<(), Box<dyn std::error::Error>> {
18    let token = std::env::var("S2_ACCESS_TOKEN")?;
19    let basin_name: BasinName = std::env::var("S2_BASIN")?.parse()?;
20
21    let client = S2::new(S2Config::new(token))?;
22
23    // ANCHOR: basin-operations
24    // List basins
25    let basins = client.list_basins(ListBasinsInput::new()).await?;
26
27    // Create a basin
28    client
29        .create_basin(CreateBasinInput::new("my-events".parse()?))
30        .await?;
31
32    // Get configuration
33    let config = client.get_basin_config("my-events".parse()?).await?;
34
35    // Delete
36    client
37        .delete_basin(DeleteBasinInput::new("my-events".parse()?))
38        .await?;
39    // ANCHOR_END: basin-operations
40    println!("Basins: {:?}, config: {:?}", basins, config);
41
42    let basin = client.basin(basin_name);
43
44    // ANCHOR: stream-operations
45    // List streams
46    let streams = basin
47        .list_streams(ListStreamsInput::new().with_prefix("user-".parse()?))
48        .await?;
49
50    // Create a stream
51    basin
52        .create_stream(CreateStreamInput::new("user-actions".parse()?))
53        .await?;
54
55    // Get configuration
56    let config = basin.get_stream_config("user-actions".parse()?).await?;
57
58    // Delete
59    basin
60        .delete_stream(DeleteStreamInput::new("user-actions".parse()?))
61        .await?;
62    // ANCHOR_END: stream-operations
63    println!("Streams: {:?}, config: {:?}", streams, config);
64
65    // ANCHOR: access-token-basic
66    // List tokens (returns metadata, not the secret)
67    let tokens = client.list_access_tokens(Default::default()).await?;
68
69    // Issue a token scoped to streams under "users/1234/"
70    let result = client
71        .issue_access_token(
72            IssueAccessTokenInput::new(
73                "user-1234-rw-token".parse()?,
74                AccessTokenScopeInput::from_op_group_perms(
75                    OperationGroupPermissions::new()
76                        .with_stream(ReadWritePermissions::read_write()),
77                )
78                .with_basins(BasinMatcher::Prefix("".parse()?)) // all basins
79                .with_streams(StreamMatcher::Prefix("users/1234/".parse()?)),
80            )
81            .with_expires_at("2027-01-01T00:00:00Z".parse()?),
82        )
83        .await?;
84
85    // Revoke a token
86    client
87        .revoke_access_token("user-1234-rw-token".parse()?)
88        .await?;
89    // ANCHOR_END: access-token-basic
90    println!("Tokens: {:?}, issued: {:?}", tokens, result);
91
92    // ANCHOR: access-token-restricted
93    client
94        .issue_access_token(IssueAccessTokenInput::new(
95            "restricted-token".parse()?,
96            AccessTokenScopeInput::from_op_group_perms(
97                OperationGroupPermissions::new().with_stream(ReadWritePermissions::read_only()),
98            )
99            .with_basins(BasinMatcher::Exact("production".parse()?))
100            .with_streams(StreamMatcher::Prefix("logs/".parse()?)),
101        ))
102        .await?;
103    // ANCHOR_END: access-token-restricted
104
105    // Pagination example - not executed by default
106    if false {
107        // ANCHOR: pagination
108        // Iterate through all streams with automatic pagination
109        let mut stream = basin.list_all_streams(ListAllStreamsInput::new());
110        while let Some(info) = stream.next().await {
111            let info = info?;
112            println!("{}", info.name);
113        }
114        // ANCHOR_END: pagination
115    }
116
117    Ok(())
118}

Source

pub fn list_all_streams( &self, input: ListAllStreamsInput, ) -> Streaming<StreamInfo>

List all streams, paginating automatically.

Examples found in repository ?

examples/docs_account_and_basins.rs (line 109)

17async fn main() -> Result<(), Box<dyn std::error::Error>> {
18    let token = std::env::var("S2_ACCESS_TOKEN")?;
19    let basin_name: BasinName = std::env::var("S2_BASIN")?.parse()?;
20
21    let client = S2::new(S2Config::new(token))?;
22
23    // ANCHOR: basin-operations
24    // List basins
25    let basins = client.list_basins(ListBasinsInput::new()).await?;
26
27    // Create a basin
28    client
29        .create_basin(CreateBasinInput::new("my-events".parse()?))
30        .await?;
31
32    // Get configuration
33    let config = client.get_basin_config("my-events".parse()?).await?;
34
35    // Delete
36    client
37        .delete_basin(DeleteBasinInput::new("my-events".parse()?))
38        .await?;
39    // ANCHOR_END: basin-operations
40    println!("Basins: {:?}, config: {:?}", basins, config);
41
42    let basin = client.basin(basin_name);
43
44    // ANCHOR: stream-operations
45    // List streams
46    let streams = basin
47        .list_streams(ListStreamsInput::new().with_prefix("user-".parse()?))
48        .await?;
49
50    // Create a stream
51    basin
52        .create_stream(CreateStreamInput::new("user-actions".parse()?))
53        .await?;
54
55    // Get configuration
56    let config = basin.get_stream_config("user-actions".parse()?).await?;
57
58    // Delete
59    basin
60        .delete_stream(DeleteStreamInput::new("user-actions".parse()?))
61        .await?;
62    // ANCHOR_END: stream-operations
63    println!("Streams: {:?}, config: {:?}", streams, config);
64
65    // ANCHOR: access-token-basic
66    // List tokens (returns metadata, not the secret)
67    let tokens = client.list_access_tokens(Default::default()).await?;
68
69    // Issue a token scoped to streams under "users/1234/"
70    let result = client
71        .issue_access_token(
72            IssueAccessTokenInput::new(
73                "user-1234-rw-token".parse()?,
74                AccessTokenScopeInput::from_op_group_perms(
75                    OperationGroupPermissions::new()
76                        .with_stream(ReadWritePermissions::read_write()),
77                )
78                .with_basins(BasinMatcher::Prefix("".parse()?)) // all basins
79                .with_streams(StreamMatcher::Prefix("users/1234/".parse()?)),
80            )
81            .with_expires_at("2027-01-01T00:00:00Z".parse()?),
82        )
83        .await?;
84
85    // Revoke a token
86    client
87        .revoke_access_token("user-1234-rw-token".parse()?)
88        .await?;
89    // ANCHOR_END: access-token-basic
90    println!("Tokens: {:?}, issued: {:?}", tokens, result);
91
92    // ANCHOR: access-token-restricted
93    client
94        .issue_access_token(IssueAccessTokenInput::new(
95            "restricted-token".parse()?,
96            AccessTokenScopeInput::from_op_group_perms(
97                OperationGroupPermissions::new().with_stream(ReadWritePermissions::read_only()),
98            )
99            .with_basins(BasinMatcher::Exact("production".parse()?))
100            .with_streams(StreamMatcher::Prefix("logs/".parse()?)),
101        ))
102        .await?;
103    // ANCHOR_END: access-token-restricted
104
105    // Pagination example - not executed by default
106    if false {
107        // ANCHOR: pagination
108        // Iterate through all streams with automatic pagination
109        let mut stream = basin.list_all_streams(ListAllStreamsInput::new());
110        while let Some(info) = stream.next().await {
111            let info = info?;
112            println!("{}", info.name);
113        }
114        // ANCHOR_END: pagination
115    }
116
117    Ok(())
118}

Source

pub async fn create_stream( &self, input: CreateStreamInput, ) -> Result<StreamInfo, S2Error>

Create a stream.

Examples found in repository ?

examples/create_stream.rs (line 28)

10async fn main() -> Result<(), Box<dyn std::error::Error>> {
11    let access_token =
12        std::env::var("S2_ACCESS_TOKEN").map_err(|_| "S2_ACCESS_TOKEN env var not set")?;
13    let basin_name: BasinName = std::env::var("S2_BASIN")
14        .map_err(|_| "S2_BASIN env var not set")?
15        .parse()?;
16    let stream_name: StreamName = std::env::var("S2_STREAM")
17        .map_err(|_| "S2_STREAM env var not set")?
18        .parse()?;
19
20    let s2 = S2::new(S2Config::new(access_token))?;
21    let basin = s2.basin(basin_name);
22
23    let input = CreateStreamInput::new(stream_name.clone()).with_config(
24        StreamConfig::new().with_timestamping(
25            TimestampingConfig::new().with_mode(TimestampingMode::ClientRequire),
26        ),
27    );
28    let stream_info = basin.create_stream(input).await?;
29    println!("{stream_info:#?}");
30
31    let stream_config = basin.get_stream_config(stream_name).await?;
32    println!("{stream_config:#?}");
33
34    Ok(())
35}

More examples

Hide additional examples

examples/docs_account_and_basins.rs (line 52)

17async fn main() -> Result<(), Box<dyn std::error::Error>> {
18    let token = std::env::var("S2_ACCESS_TOKEN")?;
19    let basin_name: BasinName = std::env::var("S2_BASIN")?.parse()?;
20
21    let client = S2::new(S2Config::new(token))?;
22
23    // ANCHOR: basin-operations
24    // List basins
25    let basins = client.list_basins(ListBasinsInput::new()).await?;
26
27    // Create a basin
28    client
29        .create_basin(CreateBasinInput::new("my-events".parse()?))
30        .await?;
31
32    // Get configuration
33    let config = client.get_basin_config("my-events".parse()?).await?;
34
35    // Delete
36    client
37        .delete_basin(DeleteBasinInput::new("my-events".parse()?))
38        .await?;
39    // ANCHOR_END: basin-operations
40    println!("Basins: {:?}, config: {:?}", basins, config);
41
42    let basin = client.basin(basin_name);
43
44    // ANCHOR: stream-operations
45    // List streams
46    let streams = basin
47        .list_streams(ListStreamsInput::new().with_prefix("user-".parse()?))
48        .await?;
49
50    // Create a stream
51    basin
52        .create_stream(CreateStreamInput::new("user-actions".parse()?))
53        .await?;
54
55    // Get configuration
56    let config = basin.get_stream_config("user-actions".parse()?).await?;
57
58    // Delete
59    basin
60        .delete_stream(DeleteStreamInput::new("user-actions".parse()?))
61        .await?;
62    // ANCHOR_END: stream-operations
63    println!("Streams: {:?}, config: {:?}", streams, config);
64
65    // ANCHOR: access-token-basic
66    // List tokens (returns metadata, not the secret)
67    let tokens = client.list_access_tokens(Default::default()).await?;
68
69    // Issue a token scoped to streams under "users/1234/"
70    let result = client
71        .issue_access_token(
72            IssueAccessTokenInput::new(
73                "user-1234-rw-token".parse()?,
74                AccessTokenScopeInput::from_op_group_perms(
75                    OperationGroupPermissions::new()
76                        .with_stream(ReadWritePermissions::read_write()),
77                )
78                .with_basins(BasinMatcher::Prefix("".parse()?)) // all basins
79                .with_streams(StreamMatcher::Prefix("users/1234/".parse()?)),
80            )
81            .with_expires_at("2027-01-01T00:00:00Z".parse()?),
82        )
83        .await?;
84
85    // Revoke a token
86    client
87        .revoke_access_token("user-1234-rw-token".parse()?)
88        .await?;
89    // ANCHOR_END: access-token-basic
90    println!("Tokens: {:?}, issued: {:?}", tokens, result);
91
92    // ANCHOR: access-token-restricted
93    client
94        .issue_access_token(IssueAccessTokenInput::new(
95            "restricted-token".parse()?,
96            AccessTokenScopeInput::from_op_group_perms(
97                OperationGroupPermissions::new().with_stream(ReadWritePermissions::read_only()),
98            )
99            .with_basins(BasinMatcher::Exact("production".parse()?))
100            .with_streams(StreamMatcher::Prefix("logs/".parse()?)),
101        ))
102        .await?;
103    // ANCHOR_END: access-token-restricted
104
105    // Pagination example - not executed by default
106    if false {
107        // ANCHOR: pagination
108        // Iterate through all streams with automatic pagination
109        let mut stream = basin.list_all_streams(ListAllStreamsInput::new());
110        while let Some(info) = stream.next().await {
111            let info = info?;
112            println!("{}", info.name);
113        }
114        // ANCHOR_END: pagination
115    }
116
117    Ok(())
118}

examples/docs_streams.rs (line 36)

20async fn main() -> Result<(), Box<dyn std::error::Error>> {
21    let 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(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 records = AppendRecordBatch::try_from_iter([
43        AppendRecord::new("first event")?,
44        AppendRecord::new("second event")?,
45    ])?;
46
47    let ack = stream.append(AppendInput::new(records)).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(ReadInput::new().with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0))))
125        .await?;
126
127    while let Some(batch) = session.next().await {
128        let batch = batch?;
129        for record in batch.records {
130            println!("[{}] {:?}", record.seq_num, record.body);
131        }
132    }
133    // ANCHOR_END: read-session
134
135    // ANCHOR: read-session-tail-offset
136    // Start reading from 10 records before the current tail
137    let mut session = stream
138        .read_session(
139            ReadInput::new().with_start(ReadStart::new().with_from(ReadFrom::TailOffset(10))),
140        )
141        .await?;
142
143    while let Some(batch) = session.next().await {
144        let batch = batch?;
145        for record in batch.records {
146            println!("[{}] {:?}", record.seq_num, record.body);
147        }
148    }
149    // ANCHOR_END: read-session-tail-offset
150
151    // ANCHOR: read-session-timestamp
152    // Start reading from a specific timestamp
153    let one_hour_ago = std::time::SystemTime::now()
154        .duration_since(std::time::UNIX_EPOCH)?
155        .as_millis() as u64
156        - 3600 * 1000;
157    let mut session = stream
158        .read_session(
159            ReadInput::new()
160                .with_start(ReadStart::new().with_from(ReadFrom::Timestamp(one_hour_ago))),
161        )
162        .await?;
163
164    while let Some(batch) = session.next().await {
165        let batch = batch?;
166        for record in batch.records {
167            println!("[{}] {:?}", record.seq_num, record.body);
168        }
169    }
170    // ANCHOR_END: read-session-timestamp
171
172    // ANCHOR: read-session-until
173    // Read records until a specific timestamp
174    let one_hour_ago = std::time::SystemTime::now()
175        .duration_since(std::time::UNIX_EPOCH)?
176        .as_millis() as u64
177        - 3600 * 1000;
178    let mut session = stream
179        .read_session(
180            ReadInput::new()
181                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
182                .with_stop(ReadStop::new().with_until(..one_hour_ago)),
183        )
184        .await?;
185
186    while let Some(batch) = session.next().await {
187        let batch = batch?;
188        for record in batch.records {
189            println!("[{}] {:?}", record.seq_num, record.body);
190        }
191    }
192    // ANCHOR_END: read-session-until
193
194    // ANCHOR: read-session-wait
195    // Read all available records, and once reaching the current tail, wait an additional 30 seconds
196    // for new ones
197    let mut session = stream
198        .read_session(
199            ReadInput::new()
200                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
201                .with_stop(ReadStop::new().with_wait(30)),
202        )
203        .await?;
204
205    while let Some(batch) = session.next().await {
206        let batch = batch?;
207        for record in batch.records {
208            println!("[{}] {:?}", record.seq_num, record.body);
209        }
210    }
211    // ANCHOR_END: read-session-wait
212
213    Ok(())
214}

Source

pub async fn get_stream_config( &self, name: StreamName, ) -> Result<StreamConfig, S2Error>

Get stream configuration.

Examples found in repository ?

examples/create_stream.rs (line 31)

10async fn main() -> Result<(), Box<dyn std::error::Error>> {
11    let access_token =
12        std::env::var("S2_ACCESS_TOKEN").map_err(|_| "S2_ACCESS_TOKEN env var not set")?;
13    let basin_name: BasinName = std::env::var("S2_BASIN")
14        .map_err(|_| "S2_BASIN env var not set")?
15        .parse()?;
16    let stream_name: StreamName = std::env::var("S2_STREAM")
17        .map_err(|_| "S2_STREAM env var not set")?
18        .parse()?;
19
20    let s2 = S2::new(S2Config::new(access_token))?;
21    let basin = s2.basin(basin_name);
22
23    let input = CreateStreamInput::new(stream_name.clone()).with_config(
24        StreamConfig::new().with_timestamping(
25            TimestampingConfig::new().with_mode(TimestampingMode::ClientRequire),
26        ),
27    );
28    let stream_info = basin.create_stream(input).await?;
29    println!("{stream_info:#?}");
30
31    let stream_config = basin.get_stream_config(stream_name).await?;
32    println!("{stream_config:#?}");
33
34    Ok(())
35}

More examples

Hide additional examples

examples/docs_account_and_basins.rs (line 56)

17async fn main() -> Result<(), Box<dyn std::error::Error>> {
18    let token = std::env::var("S2_ACCESS_TOKEN")?;
19    let basin_name: BasinName = std::env::var("S2_BASIN")?.parse()?;
20
21    let client = S2::new(S2Config::new(token))?;
22
23    // ANCHOR: basin-operations
24    // List basins
25    let basins = client.list_basins(ListBasinsInput::new()).await?;
26
27    // Create a basin
28    client
29        .create_basin(CreateBasinInput::new("my-events".parse()?))
30        .await?;
31
32    // Get configuration
33    let config = client.get_basin_config("my-events".parse()?).await?;
34
35    // Delete
36    client
37        .delete_basin(DeleteBasinInput::new("my-events".parse()?))
38        .await?;
39    // ANCHOR_END: basin-operations
40    println!("Basins: {:?}, config: {:?}", basins, config);
41
42    let basin = client.basin(basin_name);
43
44    // ANCHOR: stream-operations
45    // List streams
46    let streams = basin
47        .list_streams(ListStreamsInput::new().with_prefix("user-".parse()?))
48        .await?;
49
50    // Create a stream
51    basin
52        .create_stream(CreateStreamInput::new("user-actions".parse()?))
53        .await?;
54
55    // Get configuration
56    let config = basin.get_stream_config("user-actions".parse()?).await?;
57
58    // Delete
59    basin
60        .delete_stream(DeleteStreamInput::new("user-actions".parse()?))
61        .await?;
62    // ANCHOR_END: stream-operations
63    println!("Streams: {:?}, config: {:?}", streams, config);
64
65    // ANCHOR: access-token-basic
66    // List tokens (returns metadata, not the secret)
67    let tokens = client.list_access_tokens(Default::default()).await?;
68
69    // Issue a token scoped to streams under "users/1234/"
70    let result = client
71        .issue_access_token(
72            IssueAccessTokenInput::new(
73                "user-1234-rw-token".parse()?,
74                AccessTokenScopeInput::from_op_group_perms(
75                    OperationGroupPermissions::new()
76                        .with_stream(ReadWritePermissions::read_write()),
77                )
78                .with_basins(BasinMatcher::Prefix("".parse()?)) // all basins
79                .with_streams(StreamMatcher::Prefix("users/1234/".parse()?)),
80            )
81            .with_expires_at("2027-01-01T00:00:00Z".parse()?),
82        )
83        .await?;
84
85    // Revoke a token
86    client
87        .revoke_access_token("user-1234-rw-token".parse()?)
88        .await?;
89    // ANCHOR_END: access-token-basic
90    println!("Tokens: {:?}, issued: {:?}", tokens, result);
91
92    // ANCHOR: access-token-restricted
93    client
94        .issue_access_token(IssueAccessTokenInput::new(
95            "restricted-token".parse()?,
96            AccessTokenScopeInput::from_op_group_perms(
97                OperationGroupPermissions::new().with_stream(ReadWritePermissions::read_only()),
98            )
99            .with_basins(BasinMatcher::Exact("production".parse()?))
100            .with_streams(StreamMatcher::Prefix("logs/".parse()?)),
101        ))
102        .await?;
103    // ANCHOR_END: access-token-restricted
104
105    // Pagination example - not executed by default
106    if false {
107        // ANCHOR: pagination
108        // Iterate through all streams with automatic pagination
109        let mut stream = basin.list_all_streams(ListAllStreamsInput::new());
110        while let Some(info) = stream.next().await {
111            let info = info?;
112            println!("{}", info.name);
113        }
114        // ANCHOR_END: pagination
115    }
116
117    Ok(())
118}

Source

pub async fn delete_stream( &self, input: DeleteStreamInput, ) -> Result<(), S2Error>

Delete a stream.

Examples found in repository ?

examples/delete_stream.rs (line 21)

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 basin = s2.basin(basin_name);
19
20    let input = DeleteStreamInput::new(stream_name);
21    basin.delete_stream(input).await?;
22    println!("Deletion requested");
23
24    Ok(())
25}

More examples

Hide additional examples

examples/docs_account_and_basins.rs (line 60)

17async fn main() -> Result<(), Box<dyn std::error::Error>> {
18    let token = std::env::var("S2_ACCESS_TOKEN")?;
19    let basin_name: BasinName = std::env::var("S2_BASIN")?.parse()?;
20
21    let client = S2::new(S2Config::new(token))?;
22
23    // ANCHOR: basin-operations
24    // List basins
25    let basins = client.list_basins(ListBasinsInput::new()).await?;
26
27    // Create a basin
28    client
29        .create_basin(CreateBasinInput::new("my-events".parse()?))
30        .await?;
31
32    // Get configuration
33    let config = client.get_basin_config("my-events".parse()?).await?;
34
35    // Delete
36    client
37        .delete_basin(DeleteBasinInput::new("my-events".parse()?))
38        .await?;
39    // ANCHOR_END: basin-operations
40    println!("Basins: {:?}, config: {:?}", basins, config);
41
42    let basin = client.basin(basin_name);
43
44    // ANCHOR: stream-operations
45    // List streams
46    let streams = basin
47        .list_streams(ListStreamsInput::new().with_prefix("user-".parse()?))
48        .await?;
49
50    // Create a stream
51    basin
52        .create_stream(CreateStreamInput::new("user-actions".parse()?))
53        .await?;
54
55    // Get configuration
56    let config = basin.get_stream_config("user-actions".parse()?).await?;
57
58    // Delete
59    basin
60        .delete_stream(DeleteStreamInput::new("user-actions".parse()?))
61        .await?;
62    // ANCHOR_END: stream-operations
63    println!("Streams: {:?}, config: {:?}", streams, config);
64
65    // ANCHOR: access-token-basic
66    // List tokens (returns metadata, not the secret)
67    let tokens = client.list_access_tokens(Default::default()).await?;
68
69    // Issue a token scoped to streams under "users/1234/"
70    let result = client
71        .issue_access_token(
72            IssueAccessTokenInput::new(
73                "user-1234-rw-token".parse()?,
74                AccessTokenScopeInput::from_op_group_perms(
75                    OperationGroupPermissions::new()
76                        .with_stream(ReadWritePermissions::read_write()),
77                )
78                .with_basins(BasinMatcher::Prefix("".parse()?)) // all basins
79                .with_streams(StreamMatcher::Prefix("users/1234/".parse()?)),
80            )
81            .with_expires_at("2027-01-01T00:00:00Z".parse()?),
82        )
83        .await?;
84
85    // Revoke a token
86    client
87        .revoke_access_token("user-1234-rw-token".parse()?)
88        .await?;
89    // ANCHOR_END: access-token-basic
90    println!("Tokens: {:?}, issued: {:?}", tokens, result);
91
92    // ANCHOR: access-token-restricted
93    client
94        .issue_access_token(IssueAccessTokenInput::new(
95            "restricted-token".parse()?,
96            AccessTokenScopeInput::from_op_group_perms(
97                OperationGroupPermissions::new().with_stream(ReadWritePermissions::read_only()),
98            )
99            .with_basins(BasinMatcher::Exact("production".parse()?))
100            .with_streams(StreamMatcher::Prefix("logs/".parse()?)),
101        ))
102        .await?;
103    // ANCHOR_END: access-token-restricted
104
105    // Pagination example - not executed by default
106    if false {
107        // ANCHOR: pagination
108        // Iterate through all streams with automatic pagination
109        let mut stream = basin.list_all_streams(ListAllStreamsInput::new());
110        while let Some(info) = stream.next().await {
111            let info = info?;
112            println!("{}", info.name);
113        }
114        // ANCHOR_END: pagination
115    }
116
117    Ok(())
118}

examples/docs_streams.rs (line 111)

20async fn main() -> Result<(), Box<dyn std::error::Error>> {
21    let 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(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 records = AppendRecordBatch::try_from_iter([
43        AppendRecord::new("first event")?,
44        AppendRecord::new("second event")?,
45    ])?;
46
47    let ack = stream.append(AppendInput::new(records)).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(ReadInput::new().with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0))))
125        .await?;
126
127    while let Some(batch) = session.next().await {
128        let batch = batch?;
129        for record in batch.records {
130            println!("[{}] {:?}", record.seq_num, record.body);
131        }
132    }
133    // ANCHOR_END: read-session
134
135    // ANCHOR: read-session-tail-offset
136    // Start reading from 10 records before the current tail
137    let mut session = stream
138        .read_session(
139            ReadInput::new().with_start(ReadStart::new().with_from(ReadFrom::TailOffset(10))),
140        )
141        .await?;
142
143    while let Some(batch) = session.next().await {
144        let batch = batch?;
145        for record in batch.records {
146            println!("[{}] {:?}", record.seq_num, record.body);
147        }
148    }
149    // ANCHOR_END: read-session-tail-offset
150
151    // ANCHOR: read-session-timestamp
152    // Start reading from a specific timestamp
153    let one_hour_ago = std::time::SystemTime::now()
154        .duration_since(std::time::UNIX_EPOCH)?
155        .as_millis() as u64
156        - 3600 * 1000;
157    let mut session = stream
158        .read_session(
159            ReadInput::new()
160                .with_start(ReadStart::new().with_from(ReadFrom::Timestamp(one_hour_ago))),
161        )
162        .await?;
163
164    while let Some(batch) = session.next().await {
165        let batch = batch?;
166        for record in batch.records {
167            println!("[{}] {:?}", record.seq_num, record.body);
168        }
169    }
170    // ANCHOR_END: read-session-timestamp
171
172    // ANCHOR: read-session-until
173    // Read records until a specific timestamp
174    let one_hour_ago = std::time::SystemTime::now()
175        .duration_since(std::time::UNIX_EPOCH)?
176        .as_millis() as u64
177        - 3600 * 1000;
178    let mut session = stream
179        .read_session(
180            ReadInput::new()
181                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
182                .with_stop(ReadStop::new().with_until(..one_hour_ago)),
183        )
184        .await?;
185
186    while let Some(batch) = session.next().await {
187        let batch = batch?;
188        for record in batch.records {
189            println!("[{}] {:?}", record.seq_num, record.body);
190        }
191    }
192    // ANCHOR_END: read-session-until
193
194    // ANCHOR: read-session-wait
195    // Read all available records, and once reaching the current tail, wait an additional 30 seconds
196    // for new ones
197    let mut session = stream
198        .read_session(
199            ReadInput::new()
200                .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
201                .with_stop(ReadStop::new().with_wait(30)),
202        )
203        .await?;
204
205    while let Some(batch) = session.next().await {
206        let batch = batch?;
207        for record in batch.records {
208            println!("[{}] {:?}", record.seq_num, record.body);
209        }
210    }
211    // ANCHOR_END: read-session-wait
212
213    Ok(())
214}

Source

pub async fn reconfigure_stream( &self, input: ReconfigureStreamInput, ) -> Result<StreamConfig, S2Error>

Reconfigure a stream.

Examples found in repository ?

examples/reconfigure_stream.rs (line 27)

10async fn main() -> Result<(), Box<dyn std::error::Error>> {
11    let access_token =
12        std::env::var("S2_ACCESS_TOKEN").map_err(|_| "S2_ACCESS_TOKEN env var not set")?;
13    let basin_name: BasinName = std::env::var("S2_BASIN")
14        .map_err(|_| "S2_BASIN env var not set")?
15        .parse()?;
16    let stream_name: StreamName = std::env::var("S2_STREAM")
17        .map_err(|_| "S2_STREAM env var not set")?
18        .parse()?;
19
20    let s2 = S2::new(S2Config::new(access_token))?;
21    let basin = s2.basin(basin_name);
22
23    let input = ReconfigureStreamInput::new(
24        stream_name,
25        StreamReconfiguration::new().with_retention_policy(RetentionPolicy::Age(10 * 24 * 60 * 60)),
26    );
27    let config = basin.reconfigure_stream(input).await?;
28    println!("{config:#?}");
29
30    Ok(())
31}

Trait Implementations§

Source §

impl Clone for S2Basin

Source §

fn clone(&self) -> S2Basin

Returns a duplicate of the value. Read more

1.0.0 · Source§

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

Performs copy-assignment from source. Read more

Source §

impl Debug for S2Basin

Source §

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

Formats the value using the given formatter. Read more

Auto Trait Implementations§

§

impl !UnwindSafe for S2Basin

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 §

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

Source §

fn and<P, B, E>(self, other: P) -> And<T, P>
where T: Policy<B, E>, P: Policy<B, E>,

Create a new Policy that returns Action::Follow only if self and other return Action::Follow. Read more

Source §

fn or<P, B, E>(self, other: P) -> Or<T, P>
where T: Policy<B, E>, P: Policy<B, E>,

Create a new Policy that returns Action::Follow if either self or other returns Action::Follow. Read more

Source §

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

Source §

type Owned = T

The resulting type after obtaining ownership.

Source §

fn to_owned(&self) -> T

Creates owned data from borrowed data, usually by cloning. Read more

Source §

fn clone_into(&self, target: &mut T)

Uses borrowed data to replace owned data, usually by cloning. Read more

Source §

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

Source §

type Error = Infallible

The type returned in the event of a conversion error.

Source §

fn try_from(value: U) -> Result<T, <T as TryFrom>::Error>

Performs the conversion.

Source §

impl<T, U> TryInto for T
where U: TryFrom<T>,

Source §

type Error = >::Error

The type returned in the event of a conversion error.

Source §

fn try_into(self) -> Result<U, >::Error>

Performs the conversion.

Source §

impl<V, T> VZip<V> for T
where V: MultiLane<T>,

Source §

fn vzip(self) -> V

Source §

impl<T> WithSubscriber for T

Source §

fn with_subscriber<S>(self, subscriber: S) -> WithDispatch<Self>
where S: Into<Dispatch>,

Attaches the provided Subscriber to this type, returning a WithDispatch wrapper. Read more

Source §

fn with_current_subscriber(self) -> WithDispatch<Self>

Attaches the current default Subscriber to this type, returning a WithDispatch wrapper. Read more

Struct S2Basin Copy item path

Implementations§

impl S2Basin

pub fn stream(&self, name: StreamName) -> S2Stream

pub async fn list_streams( &self, input: ListStreamsInput, ) -> Result<Page<StreamInfo>, S2Error>

pub fn list_all_streams( &self, input: ListAllStreamsInput, ) -> Streaming<StreamInfo>

pub async fn create_stream( &self, input: CreateStreamInput, ) -> Result<StreamInfo, S2Error>

pub async fn get_stream_config( &self, name: StreamName, ) -> Result<StreamConfig, S2Error>

pub async fn delete_stream( &self, input: DeleteStreamInput, ) -> Result<(), S2Error>

pub async fn reconfigure_stream( &self, input: ReconfigureStreamInput, ) -> Result<StreamConfig, S2Error>

Trait Implementations§

impl Clone for S2Basin

fn clone(&self) -> S2Basin

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

impl Debug for S2Basin

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

Auto Trait Implementations§

impl Freeze for S2Basin

impl !RefUnwindSafe for S2Basin

impl Send for S2Basin

impl Sync for S2Basin

impl Unpin for S2Basin

impl !UnwindSafe for S2Basin

Blanket Implementations§

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

fn type_id(&self) -> TypeId

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

fn borrow(&self) -> &T

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

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

impl<T> CloneToUninit for Twhere T: Clone,

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

impl<T> From<T> for T

fn from(t: T) -> T

impl<T> Instrument for T

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

fn in_current_span(self) -> Instrumented<Self>

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

fn into(self) -> U

impl<T> IntoEither for T

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

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

impl<T> PolicyExt for Twhere T: ?Sized,

fn and<P, B, E>(self, other: P) -> And<T, P>where T: Policy<B, E>, P: Policy<B, E>,

fn or<P, B, E>(self, other: P) -> Or<T, P>where T: Policy<B, E>, P: Policy<B, E>,

impl<T> ToOwned for Twhere T: Clone,

type Owned = T

fn to_owned(&self) -> T

fn clone_into(&self, target: &mut T)

impl<T, U> TryFrom<U> for Twhere U: Into<T>,

type Error = Infallible

fn try_from(value: U) -> Result<T, <T as TryFrom<U>>::Error>

impl<T, U> TryInto<U> for Twhere U: TryFrom<T>,

type Error = <U as TryFrom<T>>::Error

fn try_into(self) -> Result<U, <U as TryFrom<T>>::Error>

impl<V, T> VZip<V> for Twhere V: MultiLane<T>,

fn vzip(self) -> V

impl<T> WithSubscriber for T

fn with_subscriber<S>(self, subscriber: S) -> WithDispatch<Self>where S: Into<Dispatch>,

fn with_current_subscriber(self) -> WithDispatch<Self>

Struct S2Basin

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

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

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

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

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

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

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

fn and<P, B, E>(self, other: P) -> And<T, P>
where T: Policy<B, E>, P: Policy<B, E>,

fn or<P, B, E>(self, other: P) -> Or<T, P>
where T: Policy<B, E>, P: Policy<B, E>,

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

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

impl<T, U> TryInto<U> for T
where U: TryFrom<T>,

impl<V, T> VZip<V> for T
where V: MultiLane<T>,

fn with_subscriber<S>(self, subscriber: S) -> WithDispatch<Self>
where S: Into<Dispatch>,