prns_runtime/manifold/kernel/
reactions.rs

1#[cfg(feature = "runtime-metrics")]
2use crate::engine::AnnounceOrigin;
3use crate::engine::{Directive, EngineReaction, FanTarget, Journaled};
4use crate::interfaces::{InterfaceId, InterfaceKind};
5
6pub struct AnnounceDirective<'a> {
7    bytes: &'a [u8],
8    hops: u8,
9    #[cfg(feature = "runtime-metrics")]
10    origin: AnnounceOrigin,
11}
12
13impl<'a> AnnounceDirective<'a> {
14    #[must_use]
15    pub fn bytes(&self) -> &'a [u8] {
16        self.bytes
17    }
18
19    #[must_use]
20    pub fn hops(&self) -> u8 {
21        self.hops
22    }
23
24    #[cfg(feature = "runtime-metrics")]
25    #[must_use]
26    pub fn origin(&self) -> AnnounceOrigin {
27        self.origin
28    }
29}
30
31pub trait DirectiveEgress {
32    fn send(&mut self, target: InterfaceId, bytes: &[u8]);
33
34    fn send_if_online(&mut self, target: InterfaceId, bytes: &[u8], on_send: &mut dyn FnMut()) {
35        on_send();
36        self.send(target, bytes);
37    }
38
39    fn send_announce(&mut self, target: InterfaceId, announce: AnnounceDirective<'_>);
40
41    fn send_to_fleet(&mut self, supervisor: InterfaceKind, fan: FanTarget, bytes: &[u8]);
42
43    fn send_announce_to_fleet(
44        &mut self,
45        supervisor: InterfaceKind,
46        fan: FanTarget,
47        announce: AnnounceDirective<'_>,
48    );
49
50    fn emit_frame(
51        &mut self,
52        target: InterfaceId,
53        size_hint: usize,
54        fill: &mut dyn FnMut(&mut [u8]) -> Option<usize>,
55    );
56
57    fn send_measured_local_announce(&mut self, target: InterfaceId, bytes: &[u8]) {
58        self.send(target, bytes);
59    }
60
61    fn send_measured_local_announce_to_fleet(
62        &mut self,
63        supervisor: InterfaceKind,
64        fan: FanTarget,
65        bytes: &[u8],
66    ) {
67        self.send_to_fleet(supervisor, fan, bytes);
68    }
69}
70
71pub fn route_reaction(
72    reaction: EngineReaction<'_>,
73    egress: &mut impl DirectiveEgress,
74    app: &mut impl FnMut(Journaled<'_>),
75) {
76    match reaction {
77        EngineReaction::Directive(Directive::Send { target, bytes }) => {
78            egress.send(target, bytes);
79        }
80        EngineReaction::Directive(Directive::SendIfOnline {
81            target,
82            bytes,
83            on_send,
84        }) => {
85            egress.send_if_online(target, bytes, on_send);
86        }
87        EngineReaction::Directive(Directive::SendAnnounce {
88            target,
89            bytes,
90            hops,
91            #[cfg(feature = "runtime-metrics")]
92            origin,
93        }) => {
94            egress.send_announce(
95                target,
96                AnnounceDirective {
97                    bytes,
98                    hops,
99                    #[cfg(feature = "runtime-metrics")]
100                    origin,
101                },
102            );
103        }
104        EngineReaction::Directive(Directive::SendToFleet {
105            supervisor,
106            fan,
107            bytes,
108        }) => {
109            egress.send_to_fleet(supervisor, fan, bytes);
110        }
111        EngineReaction::Directive(Directive::SendAnnounceToFleet {
112            supervisor,
113            fan,
114            bytes,
115            hops,
116            #[cfg(feature = "runtime-metrics")]
117            origin,
118        }) => {
119            egress.send_announce_to_fleet(
120                supervisor,
121                fan,
122                AnnounceDirective {
123                    bytes,
124                    hops,
125                    #[cfg(feature = "runtime-metrics")]
126                    origin,
127                },
128            );
129        }
130        EngineReaction::Directive(Directive::EmitFrame {
131            target,
132            size_hint,
133            fill,
134        }) => {
135            egress.emit_frame(target, size_hint, fill);
136        }
137        #[cfg(feature = "runtime-metrics")]
138        EngineReaction::Directive(Directive::SendMeasuredLocalAnnounce { target, bytes }) => {
139            egress.send_measured_local_announce(target, bytes);
140        }
141        #[cfg(feature = "runtime-metrics")]
142        EngineReaction::Directive(Directive::SendMeasuredLocalAnnounceToFleet {
143            supervisor,
144            fan,
145            bytes,
146        }) => {
147            egress.send_measured_local_announce_to_fleet(supervisor, fan, bytes);
148        }
149        EngineReaction::Journaled(journaled) => app(journaled),
150    }
151}
152
153#[cfg(test)]
154mod tests {
155    use super::*;
156
157    struct CountingEgress {
158        sends: usize,
159    }
160
161    impl DirectiveEgress for CountingEgress {
162        fn send(&mut self, _target: InterfaceId, _bytes: &[u8]) {
163            self.sends += 1;
164        }
165
166        fn send_announce(&mut self, _target: InterfaceId, _announce: AnnounceDirective<'_>) {}
167
168        fn send_to_fleet(&mut self, _supervisor: InterfaceKind, _fan: FanTarget, _bytes: &[u8]) {}
169
170        fn send_announce_to_fleet(
171            &mut self,
172            _supervisor: InterfaceKind,
173            _fan: FanTarget,
174            _announce: AnnounceDirective<'_>,
175        ) {
176        }
177
178        fn emit_frame(
179            &mut self,
180            _target: InterfaceId,
181            _size_hint: usize,
182            _fill: &mut dyn FnMut(&mut [u8]) -> Option<usize>,
183        ) {
184        }
185    }
186
187    #[test]
188    fn an_online_only_send_records_after_egress_accepts_it() {
189        let target = InterfaceId::new([0x6c; 8]);
190        let mut records = 0;
191        let mut on_send = || records += 1;
192        let mut egress = CountingEgress { sends: 0 };
193
194        route_reaction(
195            EngineReaction::Directive(Directive::SendIfOnline {
196                target,
197                bytes: b"path request",
198                on_send: &mut on_send,
199            }),
200            &mut egress,
201            &mut |_| {},
202        );
203
204        assert_eq!((records, egress.sends), (1, 1));
205    }
206}
prns_runtime/manifold/kernel/reactions.rs

prns_runtime/manifold/kernel/
reactions.rs
