clawft_kernel/cog_ingest/
udp.rs

1//! Optional UDP forwarder: the ESP32 CSI feed from the node's LAN into a
2//! container's feed port. A relay of whole datagrams, nothing parsed.
3
4use std::net::{IpAddr, SocketAddr};
5use std::sync::Arc;
6use std::sync::atomic::{AtomicU64, Ordering};
7
8use tokio::net::UdpSocket;
9use tokio::task::JoinHandle;
10
11/// Largest datagram forwarded (the feed packets are 32 and 48 bytes).
12pub const MAX_DATAGRAM: usize = 512;
13
14/// Forwarder settings.
15#[derive(Debug, Clone)]
16pub struct UdpForwardConfig {
17    /// Address to receive the feed on.
18    pub listen: SocketAddr,
19    /// The container's feed port.
20    pub dest: SocketAddr,
21    /// Only datagrams from this source address are relayed (the ESP32);
22    /// `None` accepts any source.
23    pub allowed_source: Option<IpAddr>,
24}
25
26/// A running forwarder.
27pub struct UdpForwarder {
28    local: SocketAddr,
29    forwarded: Arc<AtomicU64>,
30    dropped: Arc<AtomicU64>,
31    task: JoinHandle<()>,
32}
33
34impl UdpForwarder {
35    /// Bind and start relaying.
36    pub async fn spawn(cfg: UdpForwardConfig) -> std::io::Result<Self> {
37        let sock = UdpSocket::bind(cfg.listen).await?;
38        let local = sock.local_addr()?;
39        let out = UdpSocket::bind(if cfg.dest.is_ipv4() { "0.0.0.0:0" } else { "[::]:0" }).await?;
40        let forwarded = Arc::new(AtomicU64::new(0));
41        let dropped = Arc::new(AtomicU64::new(0));
42        let (f, d) = (forwarded.clone(), dropped.clone());
43        let task = tokio::spawn(async move {
44            // One byte over the cap, so an oversize datagram is detected.
45            let mut buf = [0u8; MAX_DATAGRAM + 1];
46            loop {
47                let Ok((n, src)) = sock.recv_from(&mut buf).await else {
48                    return;
49                };
50                let source_ok = cfg.allowed_source.is_none_or(|a| a == src.ip());
51                if n > MAX_DATAGRAM || n == 0 || !source_ok {
52                    d.fetch_add(1, Ordering::Relaxed);
53                    continue;
54                }
55                if out.send_to(&buf[..n], cfg.dest).await.is_ok() {
56                    f.fetch_add(1, Ordering::Relaxed);
57                }
58            }
59        });
60        Ok(Self {
61            local,
62            forwarded,
63            dropped,
64            task,
65        })
66    }
67
68    /// Address the feed is received on.
69    pub fn local_addr(&self) -> SocketAddr {
70        self.local
71    }
72
73    /// Datagrams relayed.
74    pub fn forwarded(&self) -> u64 {
75        self.forwarded.load(Ordering::Relaxed)
76    }
77
78    /// Datagrams dropped (oversize, empty, wrong source).
79    pub fn dropped(&self) -> u64 {
80        self.dropped.load(Ordering::Relaxed)
81    }
82}
83
84impl Drop for UdpForwarder {
85    fn drop(&mut self) {
86        self.task.abort();
87    }
88}
clawft_kernel/cog_ingest/udp.rs

clawft_kernel/cog_ingest/
udp.rs
