matc/
transport.rs

1// Simple UDP transport abstraction that multiplexes datagrams by remote address
2// into per-connection mpsc channels. Each Connection is a logical association
3// identified solely by the peer's socket address string.
4
5/// Returned by [`Connection::receive`] when the underlying mpsc channel has been
6/// closed (e.g. because the same remote address was re-registered via
7/// [`Transport::create_connection`]). Callers can detect this via
8/// `anyhow::Error::downcast_ref::<ConnectionClosed>()` and bail immediately
9/// instead of spinning on retransmit.
10#[derive(Debug)]
11pub struct ConnectionClosed;
12
13impl std::fmt::Display for ConnectionClosed {
14    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
15        write!(f, "connection closed")
16    }
17}
18
19impl std::error::Error for ConnectionClosed {}
20
21use anyhow::{Context, Result};
22use std::{
23    collections::HashMap,
24    net::{IpAddr, SocketAddr},
25    sync::{
26        atomic::{AtomicU64, Ordering},
27        Arc,
28    },
29    time::Duration,
30};
31use tokio::{net::UdpSocket, sync::Mutex};
32
33/// Normalize a peer address string so that its address family matches the
34/// local socket. This canonical form is used both as the connection map key
35/// and as the destination for `send_to`, so inbound and outbound paths agree
36/// without per-packet conversion.
37///
38/// * V6 socket + V4 peer  -> V4-mapped V6 (`[::ffff:a.b.c.d]:port`), needed
39///   because the kernel rejects cross-family `sendto` on AF_INET6.
40/// * V4 socket + V4-mapped V6 peer -> plain V4.
41/// * Other combinations are returned unchanged (real V6 on a V4 socket will
42///   fail at `send_to`, which is the correct behavior).
43fn normalize_remote_for_socket(socket: &UdpSocket, remote: &str) -> String {
44    let Ok(parsed) = remote.parse::<SocketAddr>() else {
45        return remote.to_owned();
46    };
47    let Ok(local) = socket.local_addr() else {
48        return parsed.to_string();
49    };
50    let normalized = match (local.is_ipv6(), parsed) {
51        (true, SocketAddr::V4(v4)) => {
52            let mapped = v4.ip().to_ipv6_mapped();
53            SocketAddr::new(IpAddr::V6(mapped), v4.port())
54        }
55        (false, SocketAddr::V6(v6)) => {
56            if let Some(v4) = v6.ip().to_ipv4_mapped() {
57                SocketAddr::new(IpAddr::V4(v4), v6.port())
58            } else {
59                parsed
60            }
61        }
62        _ => parsed,
63    };
64    normalized.to_string()
65}
66
67/// Transport-agnostic connection: send and receive raw Matter messages.
68///
69/// Implement this for UDP ([`Connection`]) and BTP ([`crate::btp::BtpConnection`]).
70#[async_trait::async_trait]
71pub trait ConnectionTrait: Send + Sync {
72    async fn send(&self, data: &[u8]) -> Result<()>;
73    async fn receive(&self, timeout: Duration) -> Result<Vec<u8>>;
74    /// True for transports (BTP) that guarantee delivery so Matter-layer MRP
75    /// retransmit should be suppressed.  Default: false (UDP).
76    fn is_reliable(&self) -> bool { false }
77    /// Peer MRP intervals used for retransmission timing. Defaults to spec
78    /// defaults unless overridden via [`ConnectionTrait::set_mrp_params`].
79    fn mrp_params(&self) -> crate::mrp::MrpParameters { Default::default() }
80    /// Set peer MRP intervals (typically from its mDNS SII/SAI/SAT TXT records).
81    fn set_mrp_params(&self, _params: crate::mrp::MrpParameters) {}
82    /// Time since the last message was received from the peer, if any.
83    /// Used to select the active vs idle retransmission interval.
84    fn last_received_elapsed(&self) -> Option<Duration> { None }
85}
86
87#[derive(Debug, Clone)]
88struct ConnectionInfo {
89    sender: tokio::sync::mpsc::Sender<Vec<u8>>,
90    generation: u64,
91}
92
93/// Shared transport holding:
94/// * a single UDP socket
95/// * a map of remote_addr -> channel sender
96/// * a task to read incoming datagrams and dispatch them
97/// * a task to remove connection entries when Connections drop
98pub struct Transport {
99    socket: Arc<UdpSocket>,
100    connections: Mutex<HashMap<String, ConnectionInfo>>,
101    remove_channel_sender: tokio::sync::mpsc::UnboundedSender<(String, u64)>,
102    next_generation: AtomicU64,
103    stop_receive_token: tokio_util::sync::CancellationToken,
104}
105
106/// Logical connection bound to a remote UDP address. Receiving is done by
107/// reading from an mpsc channel populated by the Transport reader task.
108pub struct Connection {
109    transport: Arc<Transport>,
110    remote_address: String,
111    /// scope_id (interface zone) for a link-local IPv6 peer, from mDNS.
112    scope_id: Option<u32>,
113    receiver: Mutex<tokio::sync::mpsc::Receiver<Vec<u8>>>,
114    generation: u64,
115    mrp: std::sync::Mutex<crate::mrp::MrpParameters>,
116    created: tokio::time::Instant,
117    /// Milliseconds since `created` of the last received datagram; u64::MAX = never.
118    last_rx_ms: AtomicU64,
119}
120
121impl Transport {
122    async fn read_from_socket_loop(
123        socket: Arc<UdpSocket>,
124        stop_receive_token: tokio_util::sync::CancellationToken,
125        self_weak: std::sync::Weak<Transport>,
126    ) -> Result<()> {
127        loop {
128            let mut buf = vec![0u8; 2048];
129            let recv_result = {
130                tokio::select! {
131                    recv_resp = socket.recv_from(&mut buf) => recv_resp,
132                    _ = stop_receive_token.cancelled() => break
133                }
134            };
135            let (n, addr) = match recv_result {
136                Ok(r) => r,
137                Err(e) => {
138                    log::debug!("transport recv error (ignored): {:?}", e);
139                    continue;
140                }
141            };
142            buf.resize(n, 0);
143            let self_strong = self_weak
144                .upgrade()
145                .context("weakpointer to self is gone - just stop")?;
146            let cons = self_strong.connections.lock().await;
147            if let Some(c) = cons.get(&scopeless_key(addr)) {
148                _ = c.sender.send(buf).await;
149            }
150        }
151        Ok(())
152    }
153
154    async fn read_from_delete_queue_loop(
155        mut remove_channel_receiver: tokio::sync::mpsc::UnboundedReceiver<(String, u64)>,
156        self_weak: std::sync::Weak<Transport>,
157    ) -> Result<()> {
158        loop {
159            let to_remove = remove_channel_receiver.recv().await;
160            match to_remove {
161                Some((addr, _gen)) if addr.is_empty() => {
162                    // Empty address is the shutdown sentinel.
163                    break;
164                }
165                Some((addr, gen)) => {
166                    let self_strong = self_weak
167                        .upgrade()
168                        .context("weak to self is gone - just stop")?;
169                    let mut cons = self_strong.connections.lock().await;
170                    // Only remove if the entry still belongs to this Connection.
171                    // A concurrent create_connection for the same address inserts a
172                    // newer generation, so the stale remove becomes a no-op.
173                    if cons.get(&addr).map(|c| c.generation) == Some(gen) {
174                        cons.remove(&addr);
175                    }
176                }
177                None => break, // Sender dropped => shutdown
178            }
179        }
180        Ok(())
181    }
182
183    /// Bind a UDP socket and spawn background tasks.
184    pub async fn new(local: &str) -> Result<Arc<Self>> {
185        let socket = UdpSocket::bind(local).await?;
186        let (remove_channel_sender, remove_channel_receiver) =
187            tokio::sync::mpsc::unbounded_channel();
188        let stop_receive_token = tokio_util::sync::CancellationToken::new();
189        let stop_receive_token_child = stop_receive_token.child_token();
190        let o = Arc::new(Self {
191            socket: Arc::new(socket),
192            connections: Mutex::new(HashMap::new()),
193            remove_channel_sender,
194            next_generation: AtomicU64::new(1),
195            stop_receive_token,
196        });
197        let self_weak = Arc::downgrade(&o.clone());
198        let socket = o.socket.clone();
199        tokio::spawn(async move {
200            _ = Self::read_from_socket_loop(socket, stop_receive_token_child, self_weak).await;
201        });
202        let self_weak = Arc::downgrade(&o.clone());
203        tokio::spawn(async move {
204            _ = Self::read_from_delete_queue_loop(remove_channel_receiver, self_weak).await;
205        });
206        Ok(o)
207    }
208
209    /// Create (or replace) a logical connection entry for the given remote address.
210    pub async fn create_connection(self: &Arc<Self>, remote: &str) -> Arc<dyn ConnectionTrait> {
211        let (remote, scope_id) = split_scope(remote);
212        let remote = normalize_remote_for_socket(&self.socket, &remote);
213        let mut clock = self.connections.lock().await;
214        let generation = self.next_generation.fetch_add(1, Ordering::Relaxed);
215        let (sender, receiver) = tokio::sync::mpsc::channel(32);
216        clock.insert(remote.to_owned(), ConnectionInfo { sender, generation });
217        Arc::new(Connection {
218            transport: self.clone(),
219            remote_address: remote,
220            scope_id,
221            receiver: Mutex::new(receiver),
222            generation,
223            mrp: std::sync::Mutex::new(Default::default()),
224            created: tokio::time::Instant::now(),
225            last_rx_ms: AtomicU64::new(u64::MAX),
226        })
227    }
228}
229
230/// True for IPv6 link-local (fe80::/10).
231fn is_link_local_v6(ip: &std::net::Ipv6Addr) -> bool {
232    (ip.segments()[0] & 0xffc0) == 0xfe80
233}
234
235/// Split the zone out of `[fe80::...%<idx>]:port`: returns the zone-less address and the scope_id.
236fn split_scope(remote: &str) -> (String, Option<u32>) {
237    if let Some(pct) = remote.find('%') {
238        let (head, tail) = remote.split_at(pct);
239        let after = &tail[1..];
240        let digits: String = after.chars().take_while(|c| c.is_ascii_digit()).collect();
241        let rest = &after[digits.len()..];
242        if let Ok(idx) = digits.parse::<u32>() {
243            return (format!("{head}{rest}"), Some(idx));
244        }
245    }
246    (remote.to_owned(), None)
247}
248
249/// Connection key without the zone.
250fn scopeless_key(addr: SocketAddr) -> String {
251    match addr {
252        // SocketAddrV6 is Copy: `mut v6` copies, so the original addr is untouched.
253        SocketAddr::V6(mut v6) if v6.scope_id() != 0 => {
254            v6.set_scope_id(0);
255            v6.to_string()
256        }
257        _ => addr.to_string(),
258    }
259}
260
261impl Connection {
262    /// Send a datagram to the remote address.
263    pub async fn send(&self, data: &[u8]) -> Result<()> {
264        let socket = &self.transport.socket;
265
266        // For link-local IPv6 attach the scope_id (interface zone from mDNS).
267        if let (Some(scope), Ok(SocketAddr::V6(v6))) =
268            (self.scope_id, self.remote_address.parse::<SocketAddr>())
269        {
270            if is_link_local_v6(v6.ip()) {
271                let target = std::net::SocketAddrV6::new(*v6.ip(), v6.port(), v6.flowinfo(), scope);
272                socket.send_to(data, SocketAddr::V6(target)).await?;
273                return Ok(());
274            }
275        }
276
277        socket.send_to(data, &self.remote_address).await?;
278        Ok(())
279    }
280    /// Receive the next datagram for this connection (with timeout).
281    ///
282    /// Returns `Err(ConnectionClosed)` (detectable via `downcast_ref`) when the
283    /// channel is permanently closed, distinct from a normal receive timeout.
284    pub async fn receive(&self, timeout: Duration) -> Result<Vec<u8>> {
285        let mut ch = self.receiver.lock().await;
286        let rec_future = ch.recv();
287        let with_timeout = tokio::time::timeout(timeout, rec_future);
288        match with_timeout.await {
289            Err(_elapsed) => Err(anyhow::anyhow!("receive timeout")),
290            Ok(None) => Err(anyhow::Error::new(ConnectionClosed)),
291            Ok(Some(v)) => {
292                self.last_rx_ms
293                    .store(self.created.elapsed().as_millis() as u64, Ordering::Relaxed);
294                Ok(v)
295            }
296        }
297    }
298}
299
300impl Drop for Transport {
301    fn drop(&mut self) {
302        _ = self.remove_channel_sender.send(("".to_owned(), 0));
303        self.stop_receive_token.cancel();
304    }
305}
306
307#[async_trait::async_trait]
308impl ConnectionTrait for Connection {
309    async fn send(&self, data: &[u8]) -> Result<()> {
310        self.send(data).await
311    }
312    async fn receive(&self, timeout: Duration) -> Result<Vec<u8>> {
313        self.receive(timeout).await
314    }
315    fn mrp_params(&self) -> crate::mrp::MrpParameters {
316        *self.mrp.lock().unwrap()
317    }
318    fn set_mrp_params(&self, params: crate::mrp::MrpParameters) {
319        *self.mrp.lock().unwrap() = params;
320    }
321    fn last_received_elapsed(&self) -> Option<Duration> {
322        let ms = self.last_rx_ms.load(Ordering::Relaxed);
323        if ms == u64::MAX {
324            return None;
325        }
326        Some(self.created.elapsed().saturating_sub(Duration::from_millis(ms)))
327    }
328}
329
330impl Drop for Connection {
331    fn drop(&mut self) {
332        _ = self
333            .transport
334            .remove_channel_sender
335            .send((self.remote_address.clone(), self.generation));
336    }
337}
338
matc/transport.rs

matc/
transport.rs
