eventuary_core/io/
start_from.rs

1use chrono::{DateTime, Utc};
2
3use crate::io::NoCursor;
4
5#[derive(Debug, Clone, Default, PartialEq, Eq)]
6pub enum StartFrom<C = NoCursor> {
7    Earliest,
8    #[default]
9    Latest,
10    Timestamp(DateTime<Utc>),
11    After(C),
12}
13
14/// Marker for subscriptions that can be told to resume from a cursor.
15/// `CheckpointReader` calls `with_start(StartFrom::After(cursor))` on the
16/// inner subscription when it has a stored checkpoint.
17pub trait StartableSubscription<C>: Clone + Send + 'static {
18    fn with_start(self, start: StartFrom<C>) -> Self;
19
20    /// Seed this subscription with a collection of candidate start
21    /// positions. Default behavior: pick the smallest `StartFrom::After(c)`
22    /// from the vec and delegate to `with_start`. Other variants
23    /// (Earliest, Latest, Timestamp) are ignored by the default impl —
24    /// readers that support fan-in, dual historic+live consumption, or
25    /// topology-aware resume must override.
26    fn with_starts(self, starts: Vec<StartFrom<C>>) -> Self
27    where
28        C: Ord,
29    {
30        let min = starts
31            .into_iter()
32            .filter_map(|s| match s {
33                StartFrom::After(c) => Some(c),
34                _ => None,
35            })
36            .min();
37        match min {
38            Some(c) => self.with_start(StartFrom::After(c)),
39            None => self,
40        }
41    }
42}
43
44#[cfg(test)]
45mod tests {
46    use super::*;
47
48    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
49    struct TestCursor(i64);
50
51    #[test]
52    fn default_is_latest() {
53        let s: StartFrom = StartFrom::default();
54        assert_eq!(s, StartFrom::Latest);
55    }
56
57    #[test]
58    fn timestamp_variant() {
59        let t = Utc::now();
60        let s: StartFrom = StartFrom::Timestamp(t);
61        if let StartFrom::Timestamp(t2) = s {
62            assert_eq!(t, t2);
63        } else {
64            panic!("expected timestamp variant");
65        }
66    }
67
68    #[test]
69    fn after_variant_carries_cursor() {
70        let start = StartFrom::After(TestCursor(9));
71        assert_eq!(start, StartFrom::After(TestCursor(9)));
72    }
73
74    #[derive(Debug, Clone, Default)]
75    struct StartableSub {
76        start: StartFrom<i64>,
77    }
78
79    impl StartableSubscription<i64> for StartableSub {
80        fn with_start(mut self, start: StartFrom<i64>) -> Self {
81            self.start = start;
82            self
83        }
84    }
85
86    #[test]
87    fn with_starts_picks_min_after_cursor() {
88        let sub = StartableSub::default();
89        let starts = vec![
90            StartFrom::After(100_i64),
91            StartFrom::After(50_i64),
92            StartFrom::After(200_i64),
93        ];
94
95        let resumed = sub.with_starts(starts);
96
97        assert_eq!(resumed.start, StartFrom::After(50_i64));
98    }
99
100    #[test]
101    fn with_starts_empty_returns_unchanged() {
102        let sub = StartableSub::default();
103
104        let resumed = sub.with_starts(vec![]);
105
106        assert_eq!(resumed.start, StartFrom::Latest);
107    }
108
109    #[test]
110    fn with_starts_ignores_non_after_variants_in_default_impl() {
111        let sub = StartableSub::default();
112
113        let resumed = sub.with_starts(vec![StartFrom::Earliest, StartFrom::Latest]);
114
115        assert_eq!(resumed.start, StartFrom::Latest);
116    }
117}
eventuary_core/io/start_from.rs

eventuary_core/io/
start_from.rs
