deepstrike_core/signals/
queue.rs

1use std::cmp::Ordering;
2use std::collections::BinaryHeap;
3
4use crate::types::signal::{RuntimeSignal, Urgency};
5
6/// Wrapper for priority ordering: higher urgency first, then older timestamp first.
7struct PrioritizedSignal {
8    urgency: Urgency,
9    timestamp_ms: u64,
10    signal: RuntimeSignal,
11}
12
13impl PartialEq for PrioritizedSignal {
14    fn eq(&self, other: &Self) -> bool {
15        self.signal.id == other.signal.id
16    }
17}
18impl Eq for PrioritizedSignal {}
19
20impl PartialOrd for PrioritizedSignal {
21    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
22        Some(self.cmp(other))
23    }
24}
25
26impl Ord for PrioritizedSignal {
27    fn cmp(&self, other: &Self) -> Ordering {
28        self.urgency
29            .cmp(&other.urgency)
30            .then_with(|| other.timestamp_ms.cmp(&self.timestamp_ms))
31            .then_with(|| self.signal.id.cmp(&other.signal.id))
32    }
33}
34
35/// Priority queue for runtime signals. Internal to the signals module.
36pub(super) struct SignalQueue {
37    heap: BinaryHeap<PrioritizedSignal>,
38    max_size: usize,
39}
40
41impl SignalQueue {
42    pub(super) fn new(max_size: usize) -> Self {
43        Self {
44            heap: BinaryHeap::new(),
45            max_size,
46        }
47    }
48
49    /// Returns false if the queue is full (signal is dropped).
50    pub(super) fn push(&mut self, signal: RuntimeSignal) -> bool {
51        if self.heap.len() >= self.max_size {
52            return false;
53        }
54        let urgency = signal.urgency;
55        let timestamp_ms = signal.timestamp_ms;
56        self.heap.push(PrioritizedSignal {
57            urgency,
58            timestamp_ms,
59            signal,
60        });
61        true
62    }
63
64    pub(super) fn pop(&mut self) -> Option<RuntimeSignal> {
65        self.heap.pop().map(|ps| ps.signal)
66    }
67
68    pub(super) fn len(&self) -> usize {
69        self.heap.len()
70    }
71}
72
73#[cfg(test)]
74mod tests {
75    use super::*;
76    use crate::types::signal::{SignalSource, SignalType};
77
78    #[test]
79    fn higher_urgency_dequeued_first() {
80        let mut q = SignalQueue::new(10);
81        q.push(
82            RuntimeSignal::new(SignalSource::Cron, SignalType::Event, Urgency::Low, "low")
83                .with_timestamp(1),
84        );
85        q.push(
86            RuntimeSignal::new(
87                SignalSource::Gateway,
88                SignalType::Alert,
89                Urgency::Critical,
90                "crit",
91            )
92            .with_timestamp(2),
93        );
94        q.push(
95            RuntimeSignal::new(
96                SignalSource::Cron,
97                SignalType::Event,
98                Urgency::Normal,
99                "norm",
100            )
101            .with_timestamp(3),
102        );
103
104        assert_eq!(q.pop().unwrap().urgency, Urgency::Critical);
105        assert_eq!(q.pop().unwrap().urgency, Urgency::Normal);
106        assert_eq!(q.pop().unwrap().urgency, Urgency::Low);
107    }
108
109    #[test]
110    fn respects_max_size() {
111        let mut q = SignalQueue::new(1);
112        assert!(
113            q.push(
114                RuntimeSignal::new(SignalSource::Cron, SignalType::Event, Urgency::Low, "a")
115                    .with_timestamp(1)
116            )
117        );
118        assert!(
119            !q.push(
120                RuntimeSignal::new(SignalSource::Cron, SignalType::Event, Urgency::Low, "b")
121                    .with_timestamp(2)
122            )
123        );
124    }
125
126    #[test]
127    fn same_urgency_older_first() {
128        let mut q = SignalQueue::new(10);
129        q.push(
130            RuntimeSignal::new(
131                SignalSource::Cron,
132                SignalType::Event,
133                Urgency::Normal,
134                "newer",
135            )
136            .with_timestamp(100),
137        );
138        q.push(
139            RuntimeSignal::new(
140                SignalSource::Cron,
141                SignalType::Event,
142                Urgency::Normal,
143                "older",
144            )
145            .with_timestamp(1),
146        );
147
148        assert_eq!(q.pop().unwrap().summary.as_str(), "older");
149    }
150}
deepstrike_core/signals/queue.rs

deepstrike_core/signals/
queue.rs
