sedsnet/
relay.rs

1//! Compatibility facade for the forwarding-only role of the shared router.
2//! All queue ownership, transport, discovery and reliability live in `Router`.
3#[cfg(feature = "discovery")]
4use crate::discovery::{ClientStatsSnapshot, TopologySnapshot};
5use crate::{
6    RouteSelectionMode, TelemetryError, TelemetryResult,
7    config::RuntimeMemoryConfig,
8    diagnostics::RuntimeStatsSnapshot,
9    packet::Packet,
10    router::{Clock, Router},
11};
12use alloc::{boxed::Box, string::String, sync::Arc};
13
14pub type RelaySideId = usize;
15pub const IPV4_LIKE_COMPACT_HEADER_TARGET_BYTES: usize =
16    crate::router::IPV4_LIKE_COMPACT_HEADER_TARGET_BYTES;
17pub const IPV6_LIKE_COMPACT_HEADER_TARGET_BYTES: usize =
18    crate::router::IPV6_LIKE_COMPACT_HEADER_TARGET_BYTES;
19pub const DEFAULT_SIDE_TRANSPORT_TEMPLATE_LIMIT: usize =
20    crate::router::DEFAULT_SIDE_TRANSPORT_TEMPLATE_LIMIT;
21/// Packet Handler function type
22type PacketHandlerFn = dyn Fn(&Packet) -> TelemetryResult<()> + Send + Sync + 'static;
23
24/// Packed Handler function type
25type PackedHandlerFn = dyn Fn(&[u8]) -> TelemetryResult<()> + Send + Sync + 'static;
26
27/// TX handler for a relay side: either packed or packet-based.
28#[derive(Clone)]
29pub enum RelayTxHandlerFn {
30    Packed(Arc<PackedHandlerFn>),
31    Packet(Arc<PacketHandlerFn>),
32}
33
34pub use crate::routing_config::SideOptions as RelaySideOptions;
35
36/// One side of the relay – a name + TX handler.
37#[derive(Clone)]
38pub struct RelaySide {
39    pub name: Arc<str>,
40    pub tx_handler: RelayTxHandlerFn,
41    pub opts: RelaySideOptions,
42}
43
44#[derive(Clone, Debug, PartialEq, Eq)]
45pub enum RelayItem {
46    Packed(crate::SharedBytes),
47    Packet(Arc<Packet>),
48}
49
50/// Compatibility builder backed by the same configuration as Router.
51#[derive(Clone, Debug)]
52pub struct RelayConfig {
53    core: crate::router::RouterConfig,
54}
55
56impl RelayConfig {
57    pub fn new() -> Self {
58        Self::default()
59    }
60
61    pub fn with_sender<S: AsRef<str>>(mut self, sender: S) -> Self {
62        self.core = self.core.with_sender(sender);
63        self
64    }
65
66    pub fn with_memory_config(mut self, memory: RuntimeMemoryConfig) -> TelemetryResult<Self> {
67        self.core = self.core.with_memory_config(memory)?;
68        Ok(self)
69    }
70}
71
72impl Default for RelayConfig {
73    fn default() -> Self {
74        Self {
75            core: crate::router::RouterConfig::relay(),
76        }
77    }
78}
79
80/// A forwarding-only view of the same engine used by Router.
81/// It exposes no application publishing or local endpoint registration API.
82pub struct Relay {
83    pub(crate) core: Router,
84}
85
86impl Relay {
87    pub fn new(clock: Box<dyn Clock + Send + Sync>) -> Self {
88        Self::new_with_config(RelayConfig::default(), clock)
89    }
90
91    pub fn new_with_config(cfg: RelayConfig, clock: Box<dyn Clock + Send + Sync>) -> Self {
92        Self {
93            core: Router::new_with_clock(cfg.core, clock),
94        }
95    }
96
97    pub fn sender(&self) -> Arc<str> {
98        self.core.sender()
99    }
100
101    pub fn set_sender<S: AsRef<str>>(&self, sender: S) {
102        self.core.set_sender(sender)
103    }
104
105    pub fn note_side_link_probe_sample(
106        &self,
107        side: RelaySideId,
108        bytes: usize,
109        duration_ms: u64,
110    ) -> TelemetryResult<()> {
111        self.core
112            .note_side_link_probe_sample(side, bytes, duration_ms)
113    }
114
115    pub fn add_side_packed<N, F>(&self, name: N, tx: F) -> RelaySideId
116    where
117        N: AsRef<str>,
118        F: Fn(&[u8]) -> TelemetryResult<()> + Send + Sync + 'static,
119    {
120        self.core.add_side_packed(name, tx)
121    }
122
123    pub fn add_side_packed_small_packets<N, F>(
124        &self,
125        name: N,
126        tx: F,
127        max_frame_bytes: usize,
128    ) -> RelaySideId
129    where
130        N: AsRef<str>,
131        F: Fn(&[u8]) -> TelemetryResult<()> + Send + Sync + 'static,
132    {
133        self.core
134            .add_side_packed_small_packets(name, tx, max_frame_bytes)
135    }
136
137    pub fn add_side_packed_with_options<N, F>(
138        &self,
139        name: N,
140        tx: F,
141        opts: RelaySideOptions,
142    ) -> RelaySideId
143    where
144        N: AsRef<str>,
145        F: Fn(&[u8]) -> TelemetryResult<()> + Send + Sync + 'static,
146    {
147        self.core
148            .add_side_packed_with_options(name, tx, legacy_chunk_options(opts))
149    }
150
151    pub fn add_side_packet<N, F>(&self, name: N, tx: F) -> RelaySideId
152    where
153        N: AsRef<str>,
154        F: Fn(&Packet) -> TelemetryResult<()> + Send + Sync + 'static,
155    {
156        self.core.add_side_packet(name, tx)
157    }
158
159    pub fn add_side_packet_with_options<N, F>(
160        &self,
161        name: N,
162        tx: F,
163        opts: RelaySideOptions,
164    ) -> RelaySideId
165    where
166        N: AsRef<str>,
167        F: Fn(&Packet) -> TelemetryResult<()> + Send + Sync + 'static,
168    {
169        self.core.add_side_packet_with_options(name, tx, opts)
170    }
171
172    pub fn remove_side(&self, side: RelaySideId) -> TelemetryResult<()> {
173        self.core.remove_side(side)
174    }
175
176    pub fn set_side_ingress_enabled(
177        &self,
178        side: RelaySideId,
179        enabled: bool,
180    ) -> TelemetryResult<()> {
181        self.core.set_side_ingress_enabled(side, enabled)
182    }
183
184    /// Opt one packed egress into never-attempted newest-pending best-effort data.
185    /// See [`Router::set_latest_value_side`] for eligibility and timestamp semantics.
186    #[cfg(feature = "latest-value-egress")]
187    pub fn set_latest_value_side(&self, side: Option<RelaySideId>) -> TelemetryResult<()> {
188        self.core.set_latest_value_side(side)
189    }
190
191    pub fn set_side_egress_enabled(&self, side: RelaySideId, enabled: bool) -> TelemetryResult<()> {
192        self.core.set_side_egress_enabled(side, enabled)
193    }
194
195    pub fn set_source_route_mode(
196        &self,
197        src: Option<RelaySideId>,
198        mode: RouteSelectionMode,
199    ) -> TelemetryResult<()> {
200        self.core.set_source_route_mode(src, mode)
201    }
202
203    pub fn clear_source_route_mode(&self, src: Option<RelaySideId>) -> TelemetryResult<()> {
204        self.core.clear_source_route_mode(src)
205    }
206
207    pub fn set_route_weight(
208        &self,
209        src: Option<RelaySideId>,
210        dst: RelaySideId,
211        weight: u32,
212    ) -> TelemetryResult<()> {
213        self.core.set_route_weight(src, dst, weight)
214    }
215
216    pub fn clear_route_weight(
217        &self,
218        src: Option<RelaySideId>,
219        dst: RelaySideId,
220    ) -> TelemetryResult<()> {
221        self.core.clear_route_weight(src, dst)
222    }
223
224    pub fn set_route_priority(
225        &self,
226        src: Option<RelaySideId>,
227        dst: RelaySideId,
228        priority: u32,
229    ) -> TelemetryResult<()> {
230        self.core.set_route_priority(src, dst, priority)
231    }
232
233    pub fn clear_route_priority(
234        &self,
235        src: Option<RelaySideId>,
236        dst: RelaySideId,
237    ) -> TelemetryResult<()> {
238        self.core.clear_route_priority(src, dst)
239    }
240
241    pub fn set_route(
242        &self,
243        src: Option<RelaySideId>,
244        dst: RelaySideId,
245        enabled: bool,
246    ) -> TelemetryResult<()> {
247        self.core.set_route(src, dst, enabled)
248    }
249
250    pub fn set_typed_route(
251        &self,
252        src: Option<RelaySideId>,
253        ty: crate::DataType,
254        dst: RelaySideId,
255        enabled: bool,
256    ) -> TelemetryResult<()> {
257        self.core.set_typed_route(src, ty, dst, enabled)
258    }
259
260    pub fn clear_typed_route(
261        &self,
262        src: Option<RelaySideId>,
263        ty: crate::DataType,
264        dst: RelaySideId,
265    ) -> TelemetryResult<()> {
266        self.core.clear_typed_route(src, ty, dst)
267    }
268
269    pub fn clear_route(&self, src: Option<RelaySideId>, dst: RelaySideId) -> TelemetryResult<()> {
270        self.core.clear_route(src, dst)
271    }
272
273    #[cfg(feature = "discovery")]
274    pub fn announce_discovery(&self) -> TelemetryResult<()> {
275        self.core.announce_discovery()
276    }
277
278    #[cfg(feature = "discovery")]
279    pub fn announce_leave(&self) -> TelemetryResult<()> {
280        self.core.announce_leave()
281    }
282
283    #[cfg(feature = "discovery")]
284    pub fn poll_discovery(&self) -> TelemetryResult<bool> {
285        self.core.poll_discovery()
286    }
287
288    #[cfg(feature = "discovery")]
289    pub fn export_topology(&self) -> TopologySnapshot {
290        self.core.export_topology()
291    }
292
293    #[cfg(feature = "discovery")]
294    pub fn client_stats(&self, sender_id: &str) -> Option<ClientStatsSnapshot> {
295        self.core.client_stats(sender_id)
296    }
297
298    pub fn export_runtime_stats(&self) -> RuntimeStatsSnapshot {
299        self.core.export_runtime_stats()
300    }
301
302    pub fn export_memory_layout_json(&self) -> String {
303        self.core.export_memory_layout_json()
304    }
305
306    pub fn rx_packed_from_side(&self, src: RelaySideId, bytes: &[u8]) -> TelemetryResult<()> {
307        self.core
308            .rx_packed_queue_from_side(bytes, src)
309            .map_err(legacy_ingress_error)
310    }
311
312    pub fn rx_from_side(&self, src: RelaySideId, packet: Packet) -> TelemetryResult<()> {
313        self.core
314            .rx_queue_from_side(packet, src)
315            .map_err(legacy_ingress_error)
316    }
317
318    pub fn clear_queues(&self) {
319        self.core.clear_queues()
320    }
321
322    pub fn clear_rx_queue(&self) {
323        self.core.clear_rx_queue()
324    }
325
326    pub fn clear_tx_queue(&self) {
327        self.core.clear_tx_queue()
328    }
329
330    pub fn process_rx_queue(&self) -> TelemetryResult<()> {
331        self.core.process_rx_queue()
332    }
333
334    pub fn process_tx_queue(&self) -> TelemetryResult<()> {
335        self.core.process_tx_queue()
336    }
337
338    pub fn process_all_queues(&self) -> TelemetryResult<()> {
339        self.core.process_all_queues()
340    }
341
342    pub fn process_tx_queue_with_timeout(&self, timeout_ms: u32) -> TelemetryResult<()> {
343        self.core.process_tx_queue_with_timeout(timeout_ms)
344    }
345
346    pub fn process_rx_queue_with_timeout(&self, timeout_ms: u32) -> TelemetryResult<()> {
347        self.core.process_rx_queue_with_timeout(timeout_ms)
348    }
349
350    pub fn process_all_queues_with_timeout(&self, timeout_ms: u32) -> TelemetryResult<()> {
351        self.core.process_all_queues_with_timeout(timeout_ms)
352    }
353
354    pub fn drain_queues_with_timeout(&self, timeout_ms: u32) -> TelemetryResult<()> {
355        self.core.drain_queues_with_timeout(timeout_ms)
356    }
357
358    pub fn periodic(&self, timeout_ms: u32) -> TelemetryResult<()> {
359        self.core.periodic(timeout_ms)
360    }
361}
362
363// Internal legacy tests can inspect the actual shared engine. This is not a
364// public delegation escape hatch: production Relay exposes only its facade.
365#[cfg(test)]
366impl core::ops::Deref for Relay {
367    type Target = Router;
368    fn deref(&self) -> &Router {
369        &self.core
370    }
371}
372
373// Legacy Relay treated a finite packed-frame size as implicit template mode,
374// even when the caller left the old header flag false. Normalize that preset
375// once at the facade boundary; the common codec/configuration remains singular.
376fn legacy_chunk_options(mut opts: RelaySideOptions) -> RelaySideOptions {
377    if opts.max_frame_bytes != 0 {
378        opts.header_template_enabled = true;
379    }
380    opts
381}
382fn legacy_ingress_error(error: TelemetryError) -> TelemetryError {
383    match error {
384        TelemetryError::HandlerError("router: invalid side id") => {
385            TelemetryError::HandlerError("relay: invalid side id")
386        }
387        other => other,
388    }
389}
sedsnet/relay.rs

sedsnet/
relay.rs
