ruststream_fred/
stream.rs

1//! Builder describing one Redis Streams subscription.
2//!
3//! A subscription always reads through a consumer group. Two read modes are selected by
4//! constructor, never by a runtime flag, because they return disjoint message sets:
5//!
6//! * [`RedisStream::new`] reads fresh entries off the tail (`XREADGROUP > ...`).
7//! * [`RedisStream::reclaim`] reads stale pending entries another consumer never acked
8//!   (`XAUTOCLAIM`, idle at least `min_idle`) - the crash-recovery path.
9//!
10//! Inferring the mode from a numeric parameter would be a footgun (a stray idle timeout could
11//! silently stop fresh delivery), so the mode is part of the constructor name. Recovery is a
12//! separate `reclaim` subscriber on the same group: "two handlers per group".
13
14use std::sync::atomic::{AtomicU64, Ordering};
15use std::time::Duration;
16
17use ruststream::SubscriptionSource;
18
19use crate::broker::ConnectedRedisBroker;
20use crate::deadletter::PoisonPolicy;
21use crate::delay::{DelayConfig, DelayedRetry};
22use crate::{error::RedisError, subscriber::RedisSubscriber};
23
24const DEFAULT_COUNT: u64 = 64;
25const DEFAULT_BLOCK: Duration = Duration::from_secs(5);
26
27/// Generates an automatic consumer name when the caller does not set one. Distinct names keep
28/// each in-process subscriber's pending list separate within a shared group.
29fn auto_consumer() -> String {
30    static COUNTER: AtomicU64 = AtomicU64::new(0);
31    let n = COUNTER.fetch_add(1, Ordering::Relaxed);
32    format!("ruststream-{n}")
33}
34
35/// Where a freshly created consumer group starts reading from. Only consulted when the group does
36/// not yet exist; an existing group keeps its own cursor.
37#[derive(Debug, Clone, Default)]
38pub enum StreamStart {
39    /// Only entries added after the group is created (`$`). The default.
40    #[default]
41    New,
42    /// Every entry currently in the stream (`0`).
43    Beginning,
44    /// A specific entry ID, exclusive.
45    Id(String),
46}
47
48impl StreamStart {
49    pub(crate) fn as_id(&self) -> &str {
50        match self {
51            Self::New => "$",
52            Self::Beginning => "0",
53            Self::Id(id) => id,
54        }
55    }
56}
57
58#[derive(Debug, Clone)]
59pub(crate) enum ReadMode {
60    /// `XREADGROUP >` - fresh tail.
61    Fresh,
62    /// `XAUTOCLAIM` of entries idle at least this long.
63    Reclaim { min_idle: Duration },
64}
65
66/// Describes one Redis Streams subscription against a [`ConnectedRedisBroker`].
67///
68/// # Examples
69///
70/// ```
71/// use std::time::Duration;
72/// use ruststream_fred::RedisStream;
73///
74/// // Fresh tail: a normal worker reading new entries.
75/// let fresh = RedisStream::new("orders").group("workers").count(128);
76///
77/// // Recovery: reclaim entries a crashed worker left pending for over 30s.
78/// let recover = RedisStream::reclaim("orders", Duration::from_secs(30)).group("workers");
79/// # let _ = (fresh, recover);
80/// ```
81#[derive(Debug, Clone)]
82#[must_use]
83pub struct RedisStream {
84    key: String,
85    group: Option<String>,
86    consumer: Option<String>,
87    count: Option<u64>,
88    block: Option<Duration>,
89    start: StreamStart,
90    mode: ReadMode,
91    dead_letter: Option<String>,
92    max_deliveries: Option<u64>,
93    delayed_retry: Option<DelayedRetry>,
94}
95
96impl RedisStream {
97    /// A fresh-tail subscription on `key`: reads new entries via `XREADGROUP >`.
98    ///
99    /// A consumer group is required; set it with [`group`](Self::group).
100    pub fn new(key: impl Into<String>) -> Self {
101        Self {
102            key: key.into(),
103            group: None,
104            consumer: None,
105            count: None,
106            block: None,
107            start: StreamStart::New,
108            mode: ReadMode::Fresh,
109            dead_letter: None,
110            max_deliveries: None,
111            delayed_retry: None,
112        }
113    }
114
115    /// A recovery subscription on `key`: reclaims pending entries idle at least `min_idle` via
116    /// `XAUTOCLAIM`. Run it alongside a [`new`](Self::new) subscriber on the same group to pick up
117    /// messages a consumer fetched but died before acking.
118    ///
119    /// `min_idle` has no default and must exceed the longest legitimate handler runtime: set it too
120    /// low and a healthy consumer's in-flight message gets reclaimed and processed twice.
121    pub fn reclaim(key: impl Into<String>, min_idle: Duration) -> Self {
122        Self {
123            key: key.into(),
124            group: None,
125            consumer: None,
126            count: None,
127            block: None,
128            start: StreamStart::New,
129            mode: ReadMode::Reclaim { min_idle },
130            dead_letter: None,
131            max_deliveries: None,
132            delayed_retry: None,
133        }
134    }
135
136    /// Sets the consumer group. Required for every subscription.
137    pub fn group(mut self, group: impl Into<String>) -> Self {
138        self.group = Some(group.into());
139        self
140    }
141
142    /// Sets this consumer's name within the group. Defaults to an auto-generated unique name.
143    pub fn consumer(mut self, consumer: impl Into<String>) -> Self {
144        self.consumer = Some(consumer.into());
145        self
146    }
147
148    /// Upper bound on entries fetched per read. Defaults to 64.
149    pub const fn count(mut self, count: u64) -> Self {
150        self.count = Some(count);
151        self
152    }
153
154    /// How long one read blocks waiting for entries. Defaults to 5 seconds. In fresh-tail mode this
155    /// is the `XREADGROUP` server-side block; in reclaim mode `XAUTOCLAIM` does not block, so this is
156    /// the poll interval slept between scans that find nothing to reclaim.
157    pub const fn block(mut self, block: Duration) -> Self {
158        self.block = Some(block);
159        self
160    }
161
162    /// Where a newly created group starts reading. Ignored if the group already exists. Only
163    /// meaningful for the fresh-tail [`new`](Self::new) mode.
164    pub fn start_id(mut self, start: StreamStart) -> Self {
165        self.start = start;
166        self
167    }
168
169    /// Routes dropped and poison messages to the named dead-letter stream instead of discarding
170    /// them. Off by default. The copy is tagged with
171    /// [`DEAD_LETTER_REASON_HEADER`](crate::DEAD_LETTER_REASON_HEADER). See [`crate::deadletter`].
172    pub fn dead_letter(mut self, key: impl Into<String>) -> Self {
173        self.dead_letter = Some(key.into());
174        self
175    }
176
177    /// Caps how many times a message may be delivered before it is treated as poison (dead-lettered
178    /// or, with no dead-letter stream, discarded). Off by default.
179    ///
180    /// The cap is checked against both the framework retry-count header (the `nack`/republish loop)
181    /// and the native stream delivery count (the reclaim loop), so a message poisoning either way is
182    /// caught.
183    pub const fn max_deliveries(mut self, max: u64) -> Self {
184        self.max_deliveries = Some(max);
185        self
186    }
187
188    /// Opts this subscription into durable, crash-safe delayed retry backed by a ZSET delay queue.
189    ///
190    /// Off by default: without it, `retry_after(delay)` / `nack_after(delay)` degrade to the
191    /// runtime's broker-agnostic deferred re-publish (at-most-once over the delay window). With it,
192    /// a delayed delivery is `ZADD`ed to the named ZSET and replayed from there once due, so the
193    /// retry survives a process crash. See [`DelayedRetry`] for the key and TTL requirements.
194    ///
195    /// The sweeper that replays due entries runs inside this subscription's read loop, so its
196    /// granularity is the read [`block`](Self::block) interval.
197    pub fn delayed_retry(mut self, retry: DelayedRetry) -> Self {
198        self.delayed_retry = Some(retry);
199        self
200    }
201
202    /// The stream key this subscription reads.
203    #[must_use]
204    pub fn key(&self) -> &str {
205        &self.key
206    }
207
208    pub(crate) fn group_or_err(&self) -> Result<&str, RedisError> {
209        self.group.as_deref().ok_or_else(|| {
210            RedisError::InvalidOptions(format!(
211                "stream subscription on `{}` requires a consumer group: call .group(name)",
212                self.key
213            ))
214        })
215    }
216
217    pub(crate) fn consumer_or_auto(&self) -> String {
218        self.consumer.clone().unwrap_or_else(auto_consumer)
219    }
220
221    pub(crate) fn count_or_default(&self) -> u64 {
222        self.count.unwrap_or(DEFAULT_COUNT)
223    }
224
225    pub(crate) fn block_or_default(&self) -> Duration {
226        self.block.unwrap_or(DEFAULT_BLOCK)
227    }
228
229    pub(crate) const fn start(&self) -> &StreamStart {
230        &self.start
231    }
232
233    pub(crate) fn mode(&self) -> ReadMode {
234        self.mode.clone()
235    }
236
237    pub(crate) fn poison_policy(&self) -> PoisonPolicy {
238        PoisonPolicy {
239            dead_letter: self.dead_letter.clone(),
240            max_deliveries: self.max_deliveries,
241        }
242    }
243
244    pub(crate) fn delay_config(&self) -> Option<DelayConfig> {
245        self.delayed_retry.as_ref().map(DelayConfig::from_retry)
246    }
247}
248
249impl SubscriptionSource<ConnectedRedisBroker> for RedisStream {
250    type Subscriber = RedisSubscriber;
251
252    fn name(&self) -> &str {
253        self.key()
254    }
255
256    async fn subscribe(
257        self,
258        connected: &ConnectedRedisBroker,
259    ) -> Result<Self::Subscriber, RedisError> {
260        connected.subscribe(self).await
261    }
262}
263
264#[cfg(feature = "testing")]
265impl SubscriptionSource<crate::testing::ConnectedRedisTestBroker> for RedisStream {
266    type Subscriber = crate::testing::RedisTestSubscriber;
267
268    fn name(&self) -> &str {
269        self.key()
270    }
271
272    async fn subscribe(
273        self,
274        connected: &crate::testing::ConnectedRedisTestBroker,
275    ) -> Result<Self::Subscriber, RedisError> {
276        connected.subscribe(self.key()).await
277    }
278}
279
280#[cfg(test)]
281mod tests {
282    use super::*;
283
284    #[test]
285    fn group_is_required() {
286        let err = RedisStream::new("orders").group_or_err().unwrap_err();
287        assert!(matches!(err, RedisError::InvalidOptions(msg) if msg.contains("consumer group")));
288    }
289
290    #[test]
291    fn group_set_resolves() {
292        let s = RedisStream::new("orders").group("workers");
293        assert_eq!(s.group_or_err().expect("group set"), "workers");
294    }
295
296    #[test]
297    fn start_maps_to_redis_ids() {
298        assert_eq!(StreamStart::New.as_id(), "$");
299        assert_eq!(StreamStart::Beginning.as_id(), "0");
300        assert_eq!(StreamStart::Id("5-0".into()).as_id(), "5-0");
301    }
302
303    #[test]
304    fn reclaim_carries_min_idle() {
305        let s = RedisStream::reclaim("orders", Duration::from_secs(30)).group("g");
306        assert!(matches!(s.mode(), ReadMode::Reclaim { min_idle } if min_idle.as_secs() == 30));
307    }
308}
ruststream_fred/stream.rs

ruststream_fred/
stream.rs
