ruststream_kinesis/
stream.rs

1//! [`KinesisStream`]: the subscription descriptor.
2//!
3//! The consumer model decides both cost and latency, so it is explicit on the descriptor.
4//! Where the subscription starts is not part of it: that vocabulary is
5//! [`KinesisPosition`](crate::KinesisPosition), spoken through the framework's `start_at(..)`
6//! clause and the `Seekable` capability.
7
8use std::time::Duration;
9
10use ruststream::SubscriptionSource;
11
12use crate::broker::ConnectedKinesisBroker;
13use crate::error::KinesisError;
14use crate::subscriber::KinesisSubscriber;
15
16/// A subscription descriptor for one Kinesis stream.
17///
18/// Every shard resumes from its stored checkpoint, and a shard without one starts at the tip.
19/// To open somewhere else, wrap the descriptor in the framework's `start_at(..)` clause with a
20/// [`KinesisPosition`](crate::KinesisPosition).
21///
22/// Implements [`SubscriptionSource`], so it can sit inline in the `#[subscriber(..)]`
23/// decorator:
24///
25/// ```
26/// use ruststream_kinesis::KinesisStream;
27///
28/// let source = KinesisStream::new("orders").batch(500);
29/// # let _ = source;
30/// ```
31#[derive(Debug, Clone, PartialEq, Eq)]
32#[must_use]
33pub struct KinesisStream {
34    stream: String,
35    batch: i32,
36    poll_interval: Duration,
37    create_shards: Option<i32>,
38}
39
40impl KinesisStream {
41    /// Names the stream (name or ARN).
42    pub fn new(stream: impl Into<String>) -> Self {
43        Self {
44            stream: stream.into(),
45            batch: 1000,
46            // The service recommends waiting a second between reads to stay within the
47            // per-shard budget.
48            poll_interval: Duration::from_secs(1),
49            create_shards: None,
50        }
51    }
52
53    /// Records per read call (1..=10000). Defaults to 1000.
54    pub fn batch(mut self, batch: i32) -> Self {
55        self.batch = batch;
56        self
57    }
58
59    /// The pause between reads on an idle shard. Defaults to 1 second, the service's own
60    /// recommendation; lower values spend the 5-reads-per-second budget faster.
61    pub fn poll_interval(mut self, interval: Duration) -> Self {
62        self.poll_interval = interval;
63        self
64    }
65
66    /// Creates the stream with `shards` provisioned shards on subscribe when it does not
67    /// exist yet. Meant for local development and tests; production streams are usually
68    /// managed as infrastructure.
69    pub fn create_if_missing(mut self, shards: i32) -> Self {
70        self.create_shards = Some(shards);
71        self
72    }
73
74    /// The stream name this descriptor resolves.
75    #[must_use]
76    pub fn stream(&self) -> &str {
77        &self.stream
78    }
79
80    pub(crate) fn batch_value(&self) -> i32 {
81        self.batch
82    }
83
84    pub(crate) fn poll_value(&self) -> Duration {
85        self.poll_interval
86    }
87
88    pub(crate) fn create_value(&self) -> Option<i32> {
89        self.create_shards
90    }
91
92    /// Rejects descriptors that cannot form a subscription, before any I/O.
93    pub(crate) fn validate(&self) -> Result<(), KinesisError> {
94        if self.stream.is_empty() {
95            return Err(KinesisError::Invalid("stream must be non-empty".into()));
96        }
97        if !(1..=10_000).contains(&self.batch) {
98            return Err(KinesisError::Invalid(
99                "batch must be within 1..=10000 (the read cap)".into(),
100            ));
101        }
102        if let Some(shards) = self.create_shards
103            && shards < 1
104        {
105            return Err(KinesisError::Invalid(
106                "create_if_missing needs at least one shard".into(),
107            ));
108        }
109        Ok(())
110    }
111}
112
113impl SubscriptionSource<ConnectedKinesisBroker> for KinesisStream {
114    type Subscriber = KinesisSubscriber;
115
116    fn name(&self) -> &str {
117        self.stream()
118    }
119
120    async fn subscribe(
121        self,
122        connected: &ConnectedKinesisBroker,
123    ) -> Result<KinesisSubscriber, KinesisError> {
124        connected.subscribe_stream(self).await
125    }
126}
127
128#[cfg(test)]
129mod tests {
130    use super::*;
131
132    #[test]
133    fn invalid_descriptors_are_rejected_before_io() {
134        assert!(KinesisStream::new("").validate().is_err());
135        assert!(KinesisStream::new("s").batch(0).validate().is_err());
136        assert!(KinesisStream::new("s").batch(10_001).validate().is_err());
137        assert!(
138            KinesisStream::new("s")
139                .create_if_missing(0)
140                .validate()
141                .is_err()
142        );
143    }
144}
ruststream_kinesis/stream.rs

ruststream_kinesis/
stream.rs
