rmux_core/events/
cursor.rs

1use super::ring::{OutputEvent, RecentOutputSnapshot};
2use std::ops::Range;
3
4/// Independent read position for one pane-output subscriber.
5#[derive(Debug, Clone, PartialEq, Eq)]
6pub struct OutputCursor {
7    next_sequence: u64,
8    missed_events: u64,
9}
10
11impl OutputCursor {
12    /// Creates a cursor that will next read `next_sequence`.
13    #[must_use]
14    pub const fn new(next_sequence: u64) -> Self {
15        Self {
16            next_sequence,
17            missed_events: 0,
18        }
19    }
20
21    /// Returns the next sequence this cursor expects to read.
22    #[must_use]
23    pub const fn next_sequence(&self) -> u64 {
24        self.next_sequence
25    }
26
27    /// Returns the total number of events this cursor has explicitly missed.
28    #[must_use]
29    pub const fn missed_events(&self) -> u64 {
30        self.missed_events
31    }
32
33    pub(super) fn advance_to(&mut self, next_sequence: u64) {
34        self.next_sequence = next_sequence;
35    }
36
37    pub(super) fn record_gap(&mut self, missed: u64, resume_sequence: u64) {
38        self.missed_events = self.missed_events.saturating_add(missed);
39        self.next_sequence = resume_sequence;
40    }
41}
42
43/// One cursor poll result from an [`OutputRing`](super::ring::OutputRing).
44#[derive(Debug, Clone, PartialEq, Eq)]
45pub enum OutputCursorItem {
46    /// A retained output event.
47    Event(OutputEvent),
48    /// The cursor fell behind the oldest retained event.
49    Gap(OutputGap),
50}
51
52/// Explicit report for output events that no longer fit in the ring.
53#[derive(Debug, Clone, PartialEq, Eq)]
54pub struct OutputGap {
55    expected_sequence: u64,
56    resume_sequence: u64,
57    missed_events: u64,
58    newest_sequence: u64,
59    recent_snapshot: RecentOutputSnapshot,
60}
61
62impl OutputGap {
63    pub(super) const fn new(
64        expected_sequence: u64,
65        resume_sequence: u64,
66        missed_events: u64,
67        newest_sequence: u64,
68        recent_snapshot: RecentOutputSnapshot,
69    ) -> Self {
70        Self {
71            expected_sequence,
72            resume_sequence,
73            missed_events,
74            newest_sequence,
75            recent_snapshot,
76        }
77    }
78
79    /// Returns the sequence the cursor expected before lag was detected.
80    #[must_use]
81    pub const fn expected_sequence(&self) -> u64 {
82        self.expected_sequence
83    }
84
85    /// Returns the oldest retained sequence the cursor can resume from.
86    #[must_use]
87    pub const fn resume_sequence(&self) -> u64 {
88        self.resume_sequence
89    }
90
91    /// Returns the number of events skipped by this gap.
92    #[must_use]
93    pub const fn missed_events(&self) -> u64 {
94        self.missed_events
95    }
96
97    /// Returns the half-open output sequence range skipped by this gap.
98    #[must_use]
99    pub fn missed_range(&self) -> Range<u64> {
100        self.expected_sequence..self.resume_sequence
101    }
102
103    /// Returns the newest appended sequence when the gap was reported.
104    #[must_use]
105    pub const fn newest_sequence(&self) -> u64 {
106        self.newest_sequence
107    }
108
109    /// Returns the bounded recent live bytes available at gap detection time.
110    #[must_use]
111    pub const fn recent_snapshot(&self) -> &RecentOutputSnapshot {
112        &self.recent_snapshot
113    }
114}
115
116#[cfg(test)]
117mod tests {
118    use super::{OutputCursor, OutputCursorItem};
119    use crate::events::OutputRing;
120
121    #[test]
122    fn cursor_advances_independently_through_retained_events() {
123        let mut ring = OutputRing::new(8, 64);
124        ring.push(b"one".to_vec());
125        ring.push(b"two".to_vec());
126        let mut first = ring.cursor_from_oldest();
127        let mut second = ring.cursor_from_oldest();
128
129        assert_eq!(
130            ring.poll_cursor(&mut first),
131            Some(OutputCursorItem::Event(ring.retained_events()[0].clone()))
132        );
133        assert_eq!(first.next_sequence(), 1);
134        assert_eq!(second.next_sequence(), 0);
135
136        assert_eq!(
137            ring.poll_cursor(&mut first),
138            Some(OutputCursorItem::Event(ring.retained_events()[1].clone()))
139        );
140        assert_eq!(ring.poll_cursor(&mut first), None);
141        assert_eq!(first.next_sequence(), ring.next_sequence());
142
143        assert_eq!(
144            ring.poll_cursor(&mut second),
145            Some(OutputCursorItem::Event(ring.retained_events()[0].clone()))
146        );
147        assert_eq!(second.next_sequence(), 1);
148    }
149
150    #[test]
151    fn lagged_cursor_reports_explicit_gap_and_resumes_at_oldest_event() {
152        let mut ring = OutputRing::new(2, 64);
153        let mut cursor = OutputCursor::new(0);
154        for bytes in [b"zero".as_slice(), b"one".as_slice(), b"two".as_slice()] {
155            ring.push(bytes.to_vec());
156        }
157
158        let Some(OutputCursorItem::Gap(gap)) = ring.poll_cursor(&mut cursor) else {
159            panic!("cursor should report lag");
160        };
161        assert_eq!(gap.expected_sequence(), 0);
162        assert_eq!(gap.resume_sequence(), 1);
163        assert_eq!(gap.missed_events(), 1);
164        assert_eq!(gap.missed_range(), 0..1);
165        assert_eq!(gap.newest_sequence(), 2);
166        assert_eq!(gap.recent_snapshot().oldest_sequence(), Some(0));
167        assert_eq!(gap.recent_snapshot().newest_sequence(), Some(2));
168        assert_eq!(cursor.missed_events(), 1);
169        assert_eq!(cursor.next_sequence(), 1);
170
171        let Some(OutputCursorItem::Event(event)) = ring.poll_cursor(&mut cursor) else {
172            panic!("cursor should resume with oldest retained event");
173        };
174        assert_eq!(event.sequence(), 1);
175        assert_eq!(event.bytes(), b"one");
176    }
177}
rmux_core/events/cursor.rs

rmux_core/events/
cursor.rs
