pamoja_coap/
lib.rs

1//! CoAP transport for the pamoja SDK.
2//!
3//! [`CoapTransport`] implements the core [`Transport`]
4//! trait on top of the pure-Rust [`coap_lite`] message codec and a UDP socket, so
5//! an application can talk to constrained RESTful devices through the same
6//! protocol-agnostic surface it uses for every other transport.
7//!
8//! CoAP is connectionless: [`connect`](Transport::connect) binds a local UDP
9//! socket and points it at the server, then spawns a background task that decodes
10//! inbound datagrams. A [`send`](Transport::send) is a CoAP `PUT` to a resource
11//! path, and a [`subscribe`](Transport::subscribe) registers an RFC 7641 observe on
12//! a resource so the server's notifications are forwarded to an internal queue that
13//! [`recv`](CoapTransport::recv) drains.
14//!
15//! Delivery follows the configured [`Reliability`]: [`Reliability::Confirmable`]
16//! messages are acknowledged with retransmission (at-least-once), while
17//! [`Reliability::NonConfirmable`] messages are fire-and-forget (at-most-once),
18//! which suits the cheapest, most power-constrained devices.
19//!
20//! # Examples
21//!
22//! ```no_run
23//! use pamoja_core::Transport;
24//! use pamoja_coap::{CoapConfig, CoapTransport};
25//!
26//! # async fn run() -> pamoja_core::Result<()> {
27//! let mut transport = CoapTransport::new(CoapConfig::new("localhost", 5683));
28//! transport.connect().await?;
29//! transport.subscribe("sensors/temperature").await?;
30//! transport.send("actuators/valve", b"open").await?;
31//!
32//! if let Some(message) = transport.recv().await? {
33//!     println!("{}: {} bytes", message.topic, message.payload.len());
34//! }
35//! # Ok(())
36//! # }
37//! ```
38
39use std::collections::HashMap;
40use std::sync::{Arc, Mutex};
41use std::time::Duration;
42
43use coap_lite::{CoapOption, MessageClass, MessageType, Packet, RequestType};
44use pamoja_core::{Error, Result, Transport};
45use tokio::net::UdpSocket;
46use tokio::sync::{mpsc, oneshot};
47use tokio::task::JoinHandle;
48
49/// Outstanding confirmable requests keyed by message id, each awaiting its ACK.
50type PendingAcks = Arc<Mutex<HashMap<u16, oneshot::Sender<()>>>>;
51
52/// The delivery guarantee applied to published and subscribed messages.
53///
54/// These map onto the CoAP message types defined in RFC 7252.
55#[derive(Clone, Copy, Debug, PartialEq, Eq)]
56pub enum Reliability {
57    /// Fire and forget: the request is sent once and not acknowledged.
58    NonConfirmable,
59    /// The request is acknowledged, and retransmitted until an ACK arrives.
60    Confirmable,
61}
62
63/// Connection settings for a [`CoapTransport`].
64///
65/// Construct with [`CoapConfig::new`] and refine with the chained setters; every
66/// field has a sensible default so only the server address is required.
67#[derive(Clone, Debug)]
68pub struct CoapConfig {
69    host: String,
70    port: u16,
71    bind: String,
72    reliability: Reliability,
73    ack_timeout: Duration,
74    max_retransmits: u32,
75}
76
77impl CoapConfig {
78    /// Creates a configuration pointing at the given CoAP server.
79    ///
80    /// # Arguments
81    ///
82    /// * `host` - the server hostname or IP address.
83    /// * `port` - the server UDP port, conventionally `5683` for plaintext CoAP.
84    ///
85    /// # Returns
86    ///
87    /// A configuration that binds an ephemeral local port, uses confirmable
88    /// delivery, waits two seconds for the first acknowledgement, and retransmits
89    /// up to four times.
90    pub fn new(host: impl Into<String>, port: u16) -> Self {
91        Self {
92            host: host.into(),
93            port,
94            bind: "0.0.0.0:0".to_owned(),
95            reliability: Reliability::Confirmable,
96            ack_timeout: Duration::from_secs(2),
97            max_retransmits: 4,
98        }
99    }
100
101    /// Sets the local socket address the transport binds to.
102    ///
103    /// # Arguments
104    ///
105    /// * `addr` - the local `host:port` to bind, for example `"0.0.0.0:0"` to let
106    ///   the operating system choose a free port.
107    ///
108    /// # Returns
109    ///
110    /// The updated configuration, for chaining.
111    pub fn bind(mut self, addr: impl Into<String>) -> Self {
112        self.bind = addr.into();
113        self
114    }
115
116    /// Sets the delivery guarantee applied to sends and subscriptions.
117    ///
118    /// # Arguments
119    ///
120    /// * `reliability` - confirmable (acknowledged) or non-confirmable delivery.
121    ///
122    /// # Returns
123    ///
124    /// The updated configuration, for chaining.
125    pub fn reliability(mut self, reliability: Reliability) -> Self {
126        self.reliability = reliability;
127        self
128    }
129
130    /// Sets how long to wait for the first acknowledgement of a confirmable request.
131    ///
132    /// The wait doubles for each retransmission, following the CoAP backoff.
133    ///
134    /// # Arguments
135    ///
136    /// * `timeout` - the initial acknowledgement timeout.
137    ///
138    /// # Returns
139    ///
140    /// The updated configuration, for chaining.
141    pub fn ack_timeout(mut self, timeout: Duration) -> Self {
142        self.ack_timeout = timeout;
143        self
144    }
145
146    /// Sets how many times a confirmable request is retransmitted before failing.
147    ///
148    /// # Arguments
149    ///
150    /// * `count` - the maximum number of retransmissions after the first send.
151    ///
152    /// # Returns
153    ///
154    /// The updated configuration, for chaining.
155    pub fn max_retransmits(mut self, count: u32) -> Self {
156        self.max_retransmits = count;
157        self
158    }
159}
160
161/// A message received from an observed resource.
162#[derive(Clone, Debug, PartialEq, Eq)]
163pub struct Message {
164    /// The resource path the message was published to.
165    pub topic: String,
166    /// The raw payload bytes.
167    pub payload: Vec<u8>,
168}
169
170/// A CoAP client that implements the core [`Transport`] trait.
171///
172/// A transport is created disconnected; [`connect`](Transport::connect) binds the
173/// socket and spawns the background task that decodes inbound datagrams for the
174/// life of the connection. Observe notifications are queued and read with
175/// [`recv`](CoapTransport::recv).
176pub struct CoapTransport {
177    config: CoapConfig,
178    socket: Option<Arc<UdpSocket>>,
179    incoming: Option<mpsc::UnboundedReceiver<Message>>,
180    pending: PendingAcks,
181    pump: Option<JoinHandle<()>>,
182    next_id: u16,
183    next_token: u16,
184}
185
186impl CoapTransport {
187    /// Creates a transport from the given configuration without connecting.
188    ///
189    /// # Arguments
190    ///
191    /// * `config` - the server connection settings.
192    ///
193    /// # Returns
194    ///
195    /// A disconnected transport ready for [`connect`](Transport::connect).
196    pub fn new(config: CoapConfig) -> Self {
197        Self {
198            config,
199            socket: None,
200            incoming: None,
201            pending: Arc::new(Mutex::new(HashMap::new())),
202            pump: None,
203            next_id: 0,
204            next_token: 0,
205        }
206    }
207
208    /// Reports whether the transport currently holds a bound socket.
209    ///
210    /// # Returns
211    ///
212    /// `true` once [`connect`](Transport::connect) has succeeded and before
213    /// [`disconnect`](CoapTransport::disconnect) is called.
214    pub fn is_connected(&self) -> bool {
215        self.socket.is_some()
216    }
217
218    /// Awaits the next notification from an observed resource.
219    ///
220    /// # Returns
221    ///
222    /// `Some(message)` for the next queued notification, or `None` once the
223    /// background task has stopped and no further messages will arrive.
224    ///
225    /// # Errors
226    ///
227    /// Returns [`Error::Closed`] if the transport is
228    /// not connected.
229    pub async fn recv(&mut self) -> Result<Option<Message>> {
230        let incoming = self.incoming.as_mut().ok_or(Error::Closed)?;
231        Ok(incoming.recv().await)
232    }
233
234    /// Closes the socket and stops the background task.
235    ///
236    /// Calling this on a transport that is not connected is a no-op.
237    ///
238    /// # Returns
239    ///
240    /// `Ok(())` once the background task has been stopped and the socket released.
241    ///
242    /// # Errors
243    ///
244    /// This call is best-effort and currently always returns `Ok(())`.
245    pub async fn disconnect(&mut self) -> Result<()> {
246        if let Some(pump) = self.pump.take() {
247            pump.abort();
248        }
249        self.socket = None;
250        self.incoming = None;
251        Ok(())
252    }
253
254    /// Returns the next message id and advances the counter.
255    fn next_message_id(&mut self) -> u16 {
256        let id = self.next_id;
257        self.next_id = self.next_id.wrapping_add(1);
258        id
259    }
260
261    /// Returns a fresh request token and advances the counter.
262    fn next_request_token(&mut self) -> Vec<u8> {
263        let token = self.next_token;
264        self.next_token = self.next_token.wrapping_add(1);
265        token.to_be_bytes().to_vec()
266    }
267
268    /// Transmits a confirmable datagram and waits for its acknowledgement,
269    /// retransmitting with a doubling timeout up to the configured limit.
270    async fn send_confirmable(&mut self, id: u16, bytes: &[u8], socket: &UdpSocket) -> Result<()> {
271        let mut timeout = self.config.ack_timeout;
272        for _ in 0..=self.config.max_retransmits {
273            let (tx, rx) = oneshot::channel();
274            self.pending.lock().expect("pending lock").insert(id, tx);
275            socket
276                .send(bytes)
277                .await
278                .map_err(|err| Error::Transport(err.to_string()))?;
279            match tokio::time::timeout(timeout, rx).await {
280                Ok(Ok(())) => return Ok(()),
281                Ok(Err(_)) => return Err(Error::Closed),
282                Err(_) => {
283                    self.pending.lock().expect("pending lock").remove(&id);
284                    timeout = timeout.saturating_mul(2);
285                }
286            }
287        }
288        Err(Error::Transport(format!(
289            "no acknowledgement for message {id}"
290        )))
291    }
292}
293
294impl Transport for CoapTransport {
295    async fn connect(&mut self) -> Result<()> {
296        let server = tokio::net::lookup_host((self.config.host.as_str(), self.config.port))
297            .await
298            .map_err(|err| Error::Transport(err.to_string()))?
299            .next()
300            .ok_or_else(|| Error::Transport(format!("could not resolve {}", self.config.host)))?;
301
302        let socket = UdpSocket::bind(&self.config.bind)
303            .await
304            .map_err(|err| Error::Transport(err.to_string()))?;
305        socket
306            .connect(server)
307            .await
308            .map_err(|err| Error::Transport(err.to_string()))?;
309        let socket = Arc::new(socket);
310
311        let (tx, rx) = mpsc::unbounded_channel();
312        let pending = Arc::clone(&self.pending);
313        let pump_socket = Arc::clone(&socket);
314        let pump = tokio::spawn(async move {
315            let mut buf = vec![0u8; 1500];
316            while let Ok(len) = pump_socket.recv(&mut buf).await {
317                let Ok(packet) = Packet::from_bytes(&buf[..len]) else {
318                    continue;
319                };
320                if !dispatch(packet, &pending, &tx, &pump_socket).await {
321                    break;
322                }
323            }
324        });
325
326        self.socket = Some(socket);
327        self.incoming = Some(rx);
328        self.pump = Some(pump);
329        Ok(())
330    }
331
332    async fn send(&mut self, topic: &str, payload: &[u8]) -> Result<()> {
333        let socket = self.socket.clone().ok_or(Error::Closed)?;
334        let id = self.next_message_id();
335        let token = self.next_request_token();
336
337        let mut packet = Packet::new();
338        packet.header.set_version(1);
339        packet
340            .header
341            .set_type(message_type(self.config.reliability));
342        packet.header.code = MessageClass::Request(RequestType::Put);
343        packet.header.message_id = id;
344        packet.set_token(token);
345        add_path(&mut packet, topic);
346        packet.payload = payload.to_vec();
347
348        let bytes = packet
349            .to_bytes()
350            .map_err(|err| Error::Codec(err.to_string()))?;
351
352        match self.config.reliability {
353            Reliability::NonConfirmable => socket
354                .send(&bytes)
355                .await
356                .map(|_| ())
357                .map_err(|err| Error::Transport(err.to_string())),
358            Reliability::Confirmable => self.send_confirmable(id, &bytes, &socket).await,
359        }
360    }
361
362    async fn subscribe(&mut self, topic: &str) -> Result<()> {
363        let socket = self.socket.clone().ok_or(Error::Closed)?;
364        let id = self.next_message_id();
365        let token = self.next_request_token();
366
367        let mut packet = Packet::new();
368        packet.header.set_version(1);
369        packet.header.set_type(MessageType::Confirmable);
370        packet.header.code = MessageClass::Request(RequestType::Get);
371        packet.header.message_id = id;
372        packet.set_token(token);
373        // An empty observe option value registers the observation (RFC 7641).
374        packet.add_option(CoapOption::Observe, Vec::new());
375        add_path(&mut packet, topic);
376
377        let bytes = packet
378            .to_bytes()
379            .map_err(|err| Error::Codec(err.to_string()))?;
380
381        self.send_confirmable(id, &bytes, &socket).await
382    }
383}
384
385/// Maps a [`Reliability`] onto the CoAP message type used on the wire.
386fn message_type(reliability: Reliability) -> MessageType {
387    match reliability {
388        Reliability::NonConfirmable => MessageType::NonConfirmable,
389        Reliability::Confirmable => MessageType::Confirmable,
390    }
391}
392
393/// Adds one `Uri-Path` option per non-empty segment of `topic`.
394fn add_path(packet: &mut Packet, topic: &str) {
395    for segment in topic.split('/').filter(|segment| !segment.is_empty()) {
396        packet.add_option(CoapOption::UriPath, segment.as_bytes().to_vec());
397    }
398}
399
400/// Reconstructs a resource path from a packet's `Uri-Path` options.
401fn path_from_packet(packet: &Packet) -> String {
402    match packet.get_option(CoapOption::UriPath) {
403        Some(segments) => segments
404            .iter()
405            .map(|segment| String::from_utf8_lossy(segment).into_owned())
406            .collect::<Vec<_>>()
407            .join("/"),
408        None => String::new(),
409    }
410}
411
412/// Routes one decoded packet, returning `false` when the inbound queue is gone.
413async fn dispatch(
414    packet: Packet,
415    pending: &PendingAcks,
416    tx: &mpsc::UnboundedSender<Message>,
417    socket: &UdpSocket,
418) -> bool {
419    match packet.header.get_type() {
420        MessageType::Acknowledgement => {
421            if let Some(waiter) = pending
422                .lock()
423                .expect("pending lock")
424                .remove(&packet.header.message_id)
425            {
426                let _ = waiter.send(());
427            }
428            // A piggybacked observe notification rides in on the ACK.
429            if packet.get_option(CoapOption::Observe).is_some() {
430                return enqueue(packet, tx);
431            }
432            true
433        }
434        MessageType::Confirmable => {
435            acknowledge(&packet, socket).await;
436            enqueue(packet, tx)
437        }
438        MessageType::NonConfirmable => enqueue(packet, tx),
439        MessageType::Reset => {
440            if let Some(waiter) = pending
441                .lock()
442                .expect("pending lock")
443                .remove(&packet.header.message_id)
444            {
445                let _ = waiter.send(());
446            }
447            true
448        }
449    }
450}
451
452/// Sends an empty acknowledgement for a confirmable notification.
453async fn acknowledge(packet: &Packet, socket: &UdpSocket) {
454    let mut ack = Packet::new();
455    ack.header.set_version(1);
456    ack.header.set_type(MessageType::Acknowledgement);
457    ack.header.code = MessageClass::Empty;
458    ack.header.message_id = packet.header.message_id;
459    if let Ok(bytes) = ack.to_bytes() {
460        let _ = socket.send(&bytes).await;
461    }
462}
463
464/// Queues a notification, returning `false` once the receiver has been dropped.
465fn enqueue(packet: Packet, tx: &mpsc::UnboundedSender<Message>) -> bool {
466    let message = Message {
467        topic: path_from_packet(&packet),
468        payload: packet.payload,
469    };
470    tx.send(message).is_ok()
471}
472
473#[cfg(test)]
474mod tests {
475    use super::*;
476
477    #[test]
478    fn reliability_defaults_to_confirmable() {
479        let config = CoapConfig::new("localhost", 5683);
480        assert_eq!(config.reliability, Reliability::Confirmable);
481    }
482
483    #[test]
484    fn setters_update_the_configuration() {
485        let config = CoapConfig::new("localhost", 5683)
486            .reliability(Reliability::NonConfirmable)
487            .ack_timeout(Duration::from_millis(250))
488            .max_retransmits(1)
489            .bind("127.0.0.1:0");
490        assert_eq!(config.reliability, Reliability::NonConfirmable);
491        assert_eq!(config.ack_timeout, Duration::from_millis(250));
492        assert_eq!(config.max_retransmits, 1);
493        assert_eq!(config.bind, "127.0.0.1:0");
494    }
495
496    #[test]
497    fn path_round_trips_through_uri_path_options() {
498        let mut packet = Packet::new();
499        add_path(&mut packet, "sensors/1/temperature");
500        assert_eq!(path_from_packet(&packet), "sensors/1/temperature");
501    }
502
503    #[test]
504    fn leading_and_repeated_slashes_are_ignored() {
505        let mut packet = Packet::new();
506        add_path(&mut packet, "/sensors//1/");
507        assert_eq!(path_from_packet(&packet), "sensors/1");
508    }
509
510    #[tokio::test]
511    async fn send_before_connect_reports_closed() {
512        let mut transport = CoapTransport::new(CoapConfig::new("localhost", 5683));
513        assert!(matches!(
514            transport.send("t", b"x").await,
515            Err(Error::Closed)
516        ));
517    }
518
519    #[tokio::test]
520    async fn subscribe_before_connect_reports_closed() {
521        let mut transport = CoapTransport::new(CoapConfig::new("localhost", 5683));
522        assert!(matches!(transport.subscribe("t").await, Err(Error::Closed)));
523    }
524
525    #[tokio::test]
526    async fn recv_before_connect_reports_closed() {
527        let mut transport = CoapTransport::new(CoapConfig::new("localhost", 5683));
528        assert!(matches!(transport.recv().await, Err(Error::Closed)));
529    }
530
531    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
532    async fn confirmable_send_without_a_server_times_out() {
533        let config = CoapConfig::new("127.0.0.1", 1)
534            .ack_timeout(Duration::from_millis(20))
535            .max_retransmits(1);
536        let mut transport = CoapTransport::new(config);
537        transport.connect().await.expect("bind socket");
538        assert!(transport.send("sensors/1", b"x").await.is_err());
539    }
540}
pamoja_coap/lib.rs

pamoja_coap/
lib.rs
