eventuary_core/io/
position.rs

1use chrono::{DateTime, Utc};
2
3use crate::io::NoCursor;
4use crate::partition::Partition;
5
6#[derive(Debug, Clone, Default, PartialEq, Eq)]
7pub enum StartFrom<C = NoCursor> {
8    Earliest,
9    #[default]
10    Latest,
11    Timestamp(DateTime<Utc>),
12    After(C),
13}
14
15#[derive(Debug, Clone, Default, PartialEq, Eq)]
16pub enum StopAt<C = NoCursor> {
17    #[default]
18    Never,
19    CurrentEnd,
20    Cursor(C),
21}
22
23/// Capability trait for subscriptions that can be told to resume from a
24/// cursor. `CheckpointReader` calls `with_start(StartFrom::After(cursor))`
25/// on the inner subscription when it has a stored checkpoint.
26pub trait StartableSubscription<C>: Clone + Send + 'static {
27    fn with_start(self, start: StartFrom<C>) -> Self;
28
29    /// Seed this subscription with a collection of candidate start
30    /// positions. Default behavior: pick the smallest `StartFrom::After(c)`
31    /// from the vec and delegate to `with_start`. Other variants
32    /// (Earliest, Latest, Timestamp) are ignored by the default impl —
33    /// readers that support fan-in, dual historic+live consumption, or
34    /// topology-aware resume must override.
35    fn with_starts(self, starts: Vec<StartFrom<C>>) -> Self
36    where
37        C: Ord,
38    {
39        let min = starts
40            .into_iter()
41            .filter_map(|s| match s {
42                StartFrom::After(c) => Some(c),
43                _ => None,
44            })
45            .min();
46        match min {
47            Some(c) => self.with_start(StartFrom::After(c)),
48            None => self,
49        }
50    }
51}
52
53/// Capability trait for subscriptions that can be restricted to a single
54/// partition. `CoordinatedReader` calls `with_partition(lease.partition)` on
55/// the inner subscription before spawning each per-lease worker.
56///
57/// Sibling of [`StartableSubscription`]: same `Self -> Self` builder shape,
58/// same role (capability marker the reader composes against).
59pub trait PartitionableSubscription<C>: StartableSubscription<C> + Clone + Send + 'static {
60    fn with_partition(self, partition: Partition) -> Self;
61}
62
63#[cfg(test)]
64mod tests {
65    use super::*;
66
67    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
68    struct TestCursor(i64);
69
70    #[test]
71    fn start_from_default_is_latest() {
72        let start: StartFrom = StartFrom::default();
73        assert_eq!(start, StartFrom::Latest);
74    }
75
76    #[test]
77    fn start_from_timestamp_variant() {
78        let timestamp = Utc::now();
79        let start: StartFrom = StartFrom::Timestamp(timestamp);
80
81        assert_eq!(start, StartFrom::Timestamp(timestamp));
82    }
83
84    #[test]
85    fn start_from_after_variant_carries_cursor() {
86        let start = StartFrom::After(TestCursor(9));
87
88        assert_eq!(start, StartFrom::After(TestCursor(9)));
89    }
90
91    #[test]
92    fn stop_at_default_is_never() {
93        let stop: StopAt = StopAt::default();
94
95        assert_eq!(stop, StopAt::Never);
96    }
97
98    #[test]
99    fn stop_at_current_end_variant() {
100        let stop: StopAt<TestCursor> = StopAt::CurrentEnd;
101
102        assert_eq!(stop, StopAt::CurrentEnd);
103    }
104
105    #[test]
106    fn stop_at_cursor_variant_carries_cursor() {
107        let stop = StopAt::Cursor(TestCursor(10));
108
109        assert_eq!(stop, StopAt::Cursor(TestCursor(10)));
110    }
111
112    #[derive(Debug, Clone, Default)]
113    struct StartableSub {
114        start: StartFrom<i64>,
115    }
116
117    impl StartableSubscription<i64> for StartableSub {
118        fn with_start(mut self, start: StartFrom<i64>) -> Self {
119            self.start = start;
120            self
121        }
122    }
123
124    #[test]
125    fn with_starts_picks_min_after_cursor() {
126        let sub = StartableSub::default();
127        let starts = vec![
128            StartFrom::After(100_i64),
129            StartFrom::After(50_i64),
130            StartFrom::After(200_i64),
131        ];
132
133        let resumed = sub.with_starts(starts);
134
135        assert_eq!(resumed.start, StartFrom::After(50_i64));
136    }
137
138    #[test]
139    fn with_starts_empty_returns_unchanged() {
140        let sub = StartableSub::default();
141
142        let resumed = sub.with_starts(vec![]);
143
144        assert_eq!(resumed.start, StartFrom::Latest);
145    }
146
147    #[test]
148    fn with_starts_ignores_non_after_variants_in_default_impl() {
149        let sub = StartableSub::default();
150
151        let resumed = sub.with_starts(vec![StartFrom::Earliest, StartFrom::Latest]);
152
153        assert_eq!(resumed.start, StartFrom::Latest);
154    }
155
156    use crate::io::NoCursor;
157
158    #[derive(Clone)]
159    struct PartitionableStub;
160
161    impl StartableSubscription<NoCursor> for PartitionableStub {
162        fn with_start(self, _start: StartFrom<NoCursor>) -> Self {
163            self
164        }
165    }
166
167    impl PartitionableSubscription<NoCursor> for PartitionableStub {
168        fn with_partition(self, _partition: Partition) -> Self {
169            self
170        }
171    }
172
173    fn _accepts_partitionable<T, C>(_sub: T)
174    where
175        T: PartitionableSubscription<C>,
176    {
177    }
178
179    #[test]
180    fn partitionable_subscription_is_super_trait_of_startable() {
181        _accepts_partitionable(PartitionableStub);
182    }
183}
eventuary_core/io/position.rs

eventuary_core/io/
position.rs
