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    pub fn set_side_egress_enabled(&self, side: RelaySideId, enabled: bool) -> TelemetryResult<()> {
185        self.core.set_side_egress_enabled(side, enabled)
186    }
187
188    pub fn set_source_route_mode(
189        &self,
190        src: Option<RelaySideId>,
191        mode: RouteSelectionMode,
192    ) -> TelemetryResult<()> {
193        self.core.set_source_route_mode(src, mode)
194    }
195
196    pub fn clear_source_route_mode(&self, src: Option<RelaySideId>) -> TelemetryResult<()> {
197        self.core.clear_source_route_mode(src)
198    }
199
200    pub fn set_route_weight(
201        &self,
202        src: Option<RelaySideId>,
203        dst: RelaySideId,
204        weight: u32,
205    ) -> TelemetryResult<()> {
206        self.core.set_route_weight(src, dst, weight)
207    }
208
209    pub fn clear_route_weight(
210        &self,
211        src: Option<RelaySideId>,
212        dst: RelaySideId,
213    ) -> TelemetryResult<()> {
214        self.core.clear_route_weight(src, dst)
215    }
216
217    pub fn set_route_priority(
218        &self,
219        src: Option<RelaySideId>,
220        dst: RelaySideId,
221        priority: u32,
222    ) -> TelemetryResult<()> {
223        self.core.set_route_priority(src, dst, priority)
224    }
225
226    pub fn clear_route_priority(
227        &self,
228        src: Option<RelaySideId>,
229        dst: RelaySideId,
230    ) -> TelemetryResult<()> {
231        self.core.clear_route_priority(src, dst)
232    }
233
234    pub fn set_route(
235        &self,
236        src: Option<RelaySideId>,
237        dst: RelaySideId,
238        enabled: bool,
239    ) -> TelemetryResult<()> {
240        self.core.set_route(src, dst, enabled)
241    }
242
243    pub fn set_typed_route(
244        &self,
245        src: Option<RelaySideId>,
246        ty: crate::DataType,
247        dst: RelaySideId,
248        enabled: bool,
249    ) -> TelemetryResult<()> {
250        self.core.set_typed_route(src, ty, dst, enabled)
251    }
252
253    pub fn clear_typed_route(
254        &self,
255        src: Option<RelaySideId>,
256        ty: crate::DataType,
257        dst: RelaySideId,
258    ) -> TelemetryResult<()> {
259        self.core.clear_typed_route(src, ty, dst)
260    }
261
262    pub fn clear_route(&self, src: Option<RelaySideId>, dst: RelaySideId) -> TelemetryResult<()> {
263        self.core.clear_route(src, dst)
264    }
265
266    #[cfg(feature = "discovery")]
267    pub fn announce_discovery(&self) -> TelemetryResult<()> {
268        self.core.announce_discovery()
269    }
270
271    #[cfg(feature = "discovery")]
272    pub fn announce_leave(&self) -> TelemetryResult<()> {
273        self.core.announce_leave()
274    }
275
276    #[cfg(feature = "discovery")]
277    pub fn poll_discovery(&self) -> TelemetryResult<bool> {
278        self.core.poll_discovery()
279    }
280
281    #[cfg(feature = "discovery")]
282    pub fn export_topology(&self) -> TopologySnapshot {
283        self.core.export_topology()
284    }
285
286    #[cfg(feature = "discovery")]
287    pub fn client_stats(&self, sender_id: &str) -> Option<ClientStatsSnapshot> {
288        self.core.client_stats(sender_id)
289    }
290
291    pub fn export_runtime_stats(&self) -> RuntimeStatsSnapshot {
292        self.core.export_runtime_stats()
293    }
294
295    pub fn export_memory_layout_json(&self) -> String {
296        self.core.export_memory_layout_json()
297    }
298
299    pub fn rx_packed_from_side(&self, src: RelaySideId, bytes: &[u8]) -> TelemetryResult<()> {
300        self.core
301            .rx_packed_queue_from_side(bytes, src)
302            .map_err(legacy_ingress_error)
303    }
304
305    pub fn rx_from_side(&self, src: RelaySideId, packet: Packet) -> TelemetryResult<()> {
306        self.core
307            .rx_queue_from_side(packet, src)
308            .map_err(legacy_ingress_error)
309    }
310
311    pub fn clear_queues(&self) {
312        self.core.clear_queues()
313    }
314
315    pub fn clear_rx_queue(&self) {
316        self.core.clear_rx_queue()
317    }
318
319    pub fn clear_tx_queue(&self) {
320        self.core.clear_tx_queue()
321    }
322
323    pub fn process_rx_queue(&self) -> TelemetryResult<()> {
324        self.core.process_rx_queue()
325    }
326
327    pub fn process_tx_queue(&self) -> TelemetryResult<()> {
328        self.core.process_tx_queue()
329    }
330
331    pub fn process_all_queues(&self) -> TelemetryResult<()> {
332        self.core.process_all_queues()
333    }
334
335    pub fn process_tx_queue_with_timeout(&self, timeout_ms: u32) -> TelemetryResult<()> {
336        self.core.process_tx_queue_with_timeout(timeout_ms)
337    }
338
339    pub fn process_rx_queue_with_timeout(&self, timeout_ms: u32) -> TelemetryResult<()> {
340        self.core.process_rx_queue_with_timeout(timeout_ms)
341    }
342
343    pub fn process_all_queues_with_timeout(&self, timeout_ms: u32) -> TelemetryResult<()> {
344        self.core.process_all_queues_with_timeout(timeout_ms)
345    }
346
347    pub fn drain_queues_with_timeout(&self, timeout_ms: u32) -> TelemetryResult<()> {
348        self.core.drain_queues_with_timeout(timeout_ms)
349    }
350
351    pub fn periodic(&self, timeout_ms: u32) -> TelemetryResult<()> {
352        self.core.periodic(timeout_ms)
353    }
354}
355
356// Internal legacy tests can inspect the actual shared engine. This is not a
357// public delegation escape hatch: production Relay exposes only its facade.
358#[cfg(test)]
359impl core::ops::Deref for Relay {
360    type Target = Router;
361    fn deref(&self) -> &Router {
362        &self.core
363    }
364}
365
366// Legacy Relay treated a finite packed-frame size as implicit template mode,
367// even when the caller left the old header flag false. Normalize that preset
368// once at the facade boundary; the common codec/configuration remains singular.
369fn legacy_chunk_options(mut opts: RelaySideOptions) -> RelaySideOptions {
370    if opts.max_frame_bytes != 0 {
371        opts.header_template_enabled = true;
372    }
373    opts
374}
375fn legacy_ingress_error(error: TelemetryError) -> TelemetryError {
376    match error {
377        TelemetryError::HandlerError("router: invalid side id") => {
378            TelemetryError::HandlerError("relay: invalid side id")
379        }
380        other => other,
381    }
382}
sedsnet/relay.rs

sedsnet/
relay.rs
