pamoja_coap/
lib.rs

1//! CoAP transport for the pamoja SDK.
2//!
3//! [`CoapTransport`] implements the core [`Transport`] and [`Receive`] traits on
4//! top of the pure-Rust [`coap_lite`] message codec and a UDP socket, so an
5//! 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`](Receive::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. A confirmable request
19//! the server resets, or answers with a 4.xx or 5.xx code, fails rather than
20//! counting as delivered, and a notification is delivered under the path its
21//! observation named, since a notification carries only the registration's token.
22//!
23//! [`CoapServer`] is the other end: a gateway that takes the readings nodes send it
24//! and holds the resources they observe, through the same traits.
25//!
26//! # Examples
27//!
28//! ```no_run
29//! use pamoja_core::{Receive, Transport};
30//! use pamoja_coap::{CoapConfig, CoapTransport};
31//!
32//! # async fn run() -> pamoja_core::Result<()> {
33//! let mut transport = CoapTransport::new(CoapConfig::new("localhost", 5683));
34//! transport.connect().await?;
35//! transport.subscribe("sensors/temperature").await?;
36//! transport.send("actuators/valve", b"open").await?;
37//!
38//! if let Some(message) = transport.recv().await? {
39//!     println!("{}: {} bytes", message.topic, message.payload.len());
40//! }
41//! # Ok(())
42//! # }
43//! ```
44
45use std::collections::HashMap;
46use std::sync::{Arc, Mutex};
47use std::time::{Duration, Instant};
48
49use coap_lite::{CoapOption, MessageClass, MessageType, Packet, RequestType};
50pub use pamoja_core::Message;
51use pamoja_core::{Error, Receive, Result, Transport};
52use tokio::net::UdpSocket;
53use tokio::sync::{mpsc, oneshot};
54use tokio::task::JoinHandle;
55
56mod server;
57
58pub use server::{CoapPublisher, CoapServer};
59
60/// Outstanding confirmable requests keyed by message id, each awaiting its answer.
61type PendingAcks = Arc<Mutex<HashMap<u16, oneshot::Sender<Answer>>>>;
62
63/// The resources this client observes, keyed by the token each registration used.
64type Observations = Arc<Mutex<HashMap<Vec<u8>, Observation>>>;
65
66/// What came back for a confirmable message.
67enum Answer {
68    /// An acknowledgment and the response it may have carried.
69    Acknowledged(Reply),
70    /// A Reset: the server received the message but could not process it.
71    Reset,
72}
73
74/// A response piggybacked on an acknowledgment.
75struct Reply {
76    /// The response code, `Empty` when the response follows separately.
77    code: MessageClass,
78    /// The response payload, which for an error is the server's diagnostic.
79    payload: Vec<u8>,
80    /// The Observe option's value, present when the server registered an observation.
81    observe: Option<u32>,
82}
83
84/// One resource this client observes.
85struct Observation {
86    /// The path the registration named, which each notification is delivered under.
87    path: String,
88    /// The sequence number and arrival of the freshest notification so far.
89    freshest: Option<(u32, Instant)>,
90}
91
92/// The delivery guarantee applied to published and subscribed messages.
93///
94/// These map onto the CoAP message types defined in RFC 7252.
95#[derive(Clone, Copy, Debug, PartialEq, Eq)]
96pub enum Reliability {
97    /// Fire and forget: the request is sent once and not acknowledged.
98    NonConfirmable,
99    /// The request is acknowledged, and retransmitted until an ACK arrives.
100    Confirmable,
101}
102
103/// Connection settings for a [`CoapTransport`].
104///
105/// Construct with [`CoapConfig::new`] and refine with the chained setters; every
106/// field has a sensible default so only the server address is required.
107#[derive(Clone, Debug)]
108pub struct CoapConfig {
109    host: String,
110    port: u16,
111    bind: String,
112    reliability: Reliability,
113    ack_timeout: Duration,
114    max_retransmits: u32,
115}
116
117impl CoapConfig {
118    /// Creates a configuration pointing at the given CoAP server.
119    ///
120    /// # Arguments
121    ///
122    /// * `host` - the server hostname or IP address.
123    /// * `port` - the server UDP port, conventionally `5683` for plaintext CoAP.
124    ///
125    /// # Returns
126    ///
127    /// A configuration that binds an ephemeral local port, uses confirmable
128    /// delivery, waits two to three seconds for the first acknowledgment, and
129    /// retransmits up to four times: RFC 7252's defaults.
130    pub fn new(host: impl Into<String>, port: u16) -> Self {
131        Self {
132            host: host.into(),
133            port,
134            bind: "0.0.0.0:0".to_owned(),
135            reliability: Reliability::Confirmable,
136            ack_timeout: Duration::from_secs(2),
137            max_retransmits: 4,
138        }
139    }
140
141    /// Sets the local socket address the transport binds to.
142    ///
143    /// # Arguments
144    ///
145    /// * `addr` - the local `host:port` to bind, for example `"0.0.0.0:0"` to let
146    ///   the operating system choose a free port.
147    ///
148    /// # Returns
149    ///
150    /// The updated configuration, for chaining.
151    pub fn bind(mut self, addr: impl Into<String>) -> Self {
152        self.bind = addr.into();
153        self
154    }
155
156    /// Sets the delivery guarantee applied to sends and subscriptions.
157    ///
158    /// # Arguments
159    ///
160    /// * `reliability` - confirmable (acknowledged) or non-confirmable delivery.
161    ///
162    /// # Returns
163    ///
164    /// The updated configuration, for chaining.
165    pub fn reliability(mut self, reliability: Reliability) -> Self {
166        self.reliability = reliability;
167        self
168    }
169
170    /// Sets how long to wait for the first acknowledgment of a confirmable request.
171    ///
172    /// The first wait is drawn between this and one and a half times it, and each
173    /// wait after it doubles, as RFC 7252 section 4.2 describes. Section 4.8.1
174    /// forbids setting it below the default two seconds on a network without
175    /// congestion control; lower the retransmissions instead to give up sooner.
176    ///
177    /// # Arguments
178    ///
179    /// * `timeout` - the initial acknowledgment timeout.
180    ///
181    /// # Returns
182    ///
183    /// The updated configuration, for chaining.
184    pub fn ack_timeout(mut self, timeout: Duration) -> Self {
185        self.ack_timeout = timeout;
186        self
187    }
188
189    /// Sets how many times a confirmable request is retransmitted before failing.
190    ///
191    /// # Arguments
192    ///
193    /// * `count` - the maximum number of retransmissions after the first send.
194    ///
195    /// # Returns
196    ///
197    /// The updated configuration, for chaining.
198    pub fn max_retransmits(mut self, count: u32) -> Self {
199        self.max_retransmits = count;
200        self
201    }
202}
203
204/// A CoAP client that implements the core [`Transport`] trait.
205///
206/// A transport is created disconnected; [`connect`](Transport::connect) binds the
207/// socket and spawns the background task that decodes inbound datagrams for the
208/// life of the connection. Observe notifications are queued and read with
209/// [`recv`](CoapTransport::recv).
210pub struct CoapTransport {
211    config: CoapConfig,
212    socket: Option<Arc<UdpSocket>>,
213    incoming: Option<mpsc::UnboundedReceiver<Message>>,
214    pending: PendingAcks,
215    observations: Observations,
216    pump: Option<JoinHandle<()>>,
217    next_id: u16,
218}
219
220impl CoapTransport {
221    /// Creates a transport from the given configuration without connecting.
222    ///
223    /// # Arguments
224    ///
225    /// * `config` - the server connection settings.
226    ///
227    /// # Returns
228    ///
229    /// A disconnected transport ready for [`connect`](Transport::connect).
230    pub fn new(config: CoapConfig) -> Self {
231        Self {
232            config,
233            socket: None,
234            incoming: None,
235            pending: Arc::new(Mutex::new(HashMap::new())),
236            observations: Arc::new(Mutex::new(HashMap::new())),
237            pump: None,
238            next_id: random() as u16,
239        }
240    }
241
242    /// Reports whether the transport currently holds a bound socket.
243    ///
244    /// # Returns
245    ///
246    /// `true` once [`connect`](Transport::connect) has succeeded and before
247    /// [`disconnect`](CoapTransport::disconnect) is called.
248    pub fn is_connected(&self) -> bool {
249        self.socket.is_some()
250    }
251
252    /// Closes the socket and stops the background task.
253    ///
254    /// Calling this on a transport that is not connected is a no-op.
255    ///
256    /// # Returns
257    ///
258    /// `Ok(())` once the background task has been stopped and the socket released.
259    ///
260    /// # Errors
261    ///
262    /// This call is best-effort and currently always returns `Ok(())`.
263    pub async fn disconnect(&mut self) -> Result<()> {
264        if let Some(pump) = self.pump.take() {
265            pump.abort();
266        }
267        self.socket = None;
268        self.incoming = None;
269        self.observations.lock().expect("observations lock").clear();
270        Ok(())
271    }
272
273    /// Returns the next message id and advances the counter.
274    ///
275    /// The counter starts at a random value, as RFC 7252 section 4.4 strongly
276    /// recommends, so a node that restarts does not reuse the ids a server still
277    /// holds for spotting duplicates.
278    fn next_message_id(&mut self) -> u16 {
279        let id = self.next_id;
280        self.next_id = self.next_id.wrapping_add(1);
281        id
282    }
283
284    /// Returns a fresh, random request token.
285    ///
286    /// RFC 7252 section 5.3.1 asks a client without transport security for a
287    /// nontrivial, randomized token, so that a response cannot be spoofed by
288    /// guessing it.
289    fn next_request_token(&mut self) -> Vec<u8> {
290        random().to_be_bytes()[..TOKEN_LENGTH].to_vec()
291    }
292
293    /// Transmits a confirmable datagram and waits for its acknowledgment,
294    /// retransmitting with a doubling timeout up to the configured limit.
295    ///
296    /// The first wait is drawn between the configured timeout and one and a half
297    /// times it, and a Reset ends the attempt as a failure, both as RFC 7252
298    /// section 4.2 describes.
299    async fn exchange(&mut self, id: u16, bytes: &[u8], socket: &UdpSocket) -> Result<Reply> {
300        let spread = (random() >> 11) as f64 / (1u64 << 53) as f64;
301        let mut timeout = self
302            .config
303            .ack_timeout
304            .mul_f64(1.0 + spread * (ACK_RANDOM_FACTOR - 1.0));
305        for _ in 0..=self.config.max_retransmits {
306            let (tx, rx) = oneshot::channel();
307            self.pending.lock().expect("pending lock").insert(id, tx);
308            if let Err(error) = socket.send(bytes).await {
309                if !refused(&error) {
310                    return Err(Error::Transport(error.to_string()));
311                }
312            }
313            match tokio::time::timeout(timeout, rx).await {
314                Ok(Ok(Answer::Acknowledged(reply))) => return Ok(reply),
315                Ok(Ok(Answer::Reset)) => {
316                    return Err(Error::Transport(
317                        "the server reset the request: it arrived, but the server could not \
318                         process it"
319                            .into(),
320                    ))
321                }
322                Ok(Err(_)) => return Err(Error::Closed),
323                Err(_) => {
324                    self.pending.lock().expect("pending lock").remove(&id);
325                    timeout = timeout.saturating_mul(2);
326                }
327            }
328        }
329        let sent = self.config.max_retransmits + 1;
330        let times = if sent == 1 {
331            "transmission"
332        } else {
333            "transmissions"
334        };
335        Err(Error::Transport(format!(
336            "no acknowledgment after {sent} {times}"
337        )))
338    }
339}
340
341/// How long a random request token is, in bytes.
342const TOKEN_LENGTH: usize = 4;
343
344/// The factor RFC 7252 section 4.8 spreads the first acknowledgment wait over.
345const ACK_RANDOM_FACTOR: f64 = 1.5;
346
347/// Whether a receive failed only because an earlier datagram found no one
348/// listening.
349///
350/// The operating system reports the ICMP error for a datagram sent to a closed
351/// port on the socket's next receive: Linux as a refused connection on a
352/// connected socket, and Windows as a reset on any socket. Neither says anything
353/// about the datagram being waited for, so the socket keeps listening.
354fn refused(error: &std::io::Error) -> bool {
355    matches!(
356        error.kind(),
357        std::io::ErrorKind::ConnectionRefused | std::io::ErrorKind::ConnectionReset
358    )
359}
360
361/// A random value for message ids, tokens, and the spread of the first wait.
362///
363/// Each `RandomState` is keyed from the operating system's randomness, which is
364/// what these values need: to be unpredictable to someone off the path.
365fn random() -> u64 {
366    use std::hash::{BuildHasher, RandomState};
367    RandomState::new().hash_one(std::time::SystemTime::now())
368}
369
370/// Reads the code an acknowledgment carried, failing on an error response.
371///
372/// An empty acknowledgment, or one carrying a 2.xx success, means the request
373/// took. A 4.xx or 5.xx means it did not, and its payload is the server's
374/// diagnostic, which RFC 7252 section 5.5.2 makes human-readable text.
375fn answered(code: MessageClass, diagnostic: &[u8]) -> Result<()> {
376    let raw = u8::from(code);
377    let class = raw >> 5;
378    if class != 4 && class != 5 {
379        return Ok(());
380    }
381    let detail = raw & 0x1F;
382    let name = response_name(class, detail)
383        .map(|name| format!(" {name}"))
384        .unwrap_or_default();
385    let reason = match core::str::from_utf8(diagnostic) {
386        Ok(text) if !text.is_empty() => format!(": {text}"),
387        _ => String::new(),
388    };
389    Err(Error::Transport(format!(
390        "the server answered {class}.{detail:02}{name}{reason}"
391    )))
392}
393
394/// The name RFC 7252 table 6 gives an error response code.
395fn response_name(class: u8, detail: u8) -> Option<&'static str> {
396    Some(match (class, detail) {
397        (4, 0) => "Bad Request",
398        (4, 1) => "Unauthorized",
399        (4, 2) => "Bad Option",
400        (4, 3) => "Forbidden",
401        (4, 4) => "Not Found",
402        (4, 5) => "Method Not Allowed",
403        (4, 6) => "Not Acceptable",
404        (4, 12) => "Precondition Failed",
405        (4, 13) => "Request Entity Too Large",
406        (4, 15) => "Unsupported Content-Format",
407        (5, 0) => "Internal Server Error",
408        (5, 1) => "Not Implemented",
409        (5, 2) => "Bad Gateway",
410        (5, 3) => "Service Unavailable",
411        (5, 4) => "Gateway Timeout",
412        (5, 5) => "Proxying Not Supported",
413        _ => return None,
414    })
415}
416
417impl Transport for CoapTransport {
418    async fn connect(&mut self) -> Result<()> {
419        let server = tokio::net::lookup_host((self.config.host.as_str(), self.config.port))
420            .await
421            .map_err(|err| Error::Transport(err.to_string()))?
422            .next()
423            .ok_or_else(|| Error::Transport(format!("could not resolve {}", self.config.host)))?;
424
425        let socket = UdpSocket::bind(&self.config.bind)
426            .await
427            .map_err(|err| Error::Transport(err.to_string()))?;
428        socket
429            .connect(server)
430            .await
431            .map_err(|err| Error::Transport(err.to_string()))?;
432        let socket = Arc::new(socket);
433
434        let (tx, rx) = mpsc::unbounded_channel();
435        let pending = Arc::clone(&self.pending);
436        let observations = Arc::clone(&self.observations);
437        let pump_socket = Arc::clone(&socket);
438        let pump = tokio::spawn(async move {
439            let mut buf = vec![0u8; 1500];
440            loop {
441                let len = match pump_socket.recv(&mut buf).await {
442                    Ok(len) => len,
443                    Err(error) if refused(&error) => continue,
444                    Err(_) => break,
445                };
446                let Ok(packet) = Packet::from_bytes(&buf[..len]) else {
447                    continue;
448                };
449                if !dispatch(packet, &pending, &observations, &tx, &pump_socket).await {
450                    break;
451                }
452            }
453        });
454
455        self.socket = Some(socket);
456        self.incoming = Some(rx);
457        self.pump = Some(pump);
458        Ok(())
459    }
460
461    async fn send(&mut self, topic: &str, payload: &[u8]) -> Result<()> {
462        let socket = self.socket.clone().ok_or(Error::Closed)?;
463        let id = self.next_message_id();
464        let token = self.next_request_token();
465
466        let mut packet = Packet::new();
467        packet.header.set_version(1);
468        packet
469            .header
470            .set_type(message_type(self.config.reliability));
471        packet.header.code = MessageClass::Request(RequestType::Put);
472        packet.header.message_id = id;
473        packet.set_token(token);
474        add_path(&mut packet, topic);
475        packet.payload = payload.to_vec();
476
477        let bytes = packet
478            .to_bytes()
479            .map_err(|err| Error::Codec(err.to_string()))?;
480
481        match self.config.reliability {
482            Reliability::NonConfirmable => match socket.send(&bytes).await {
483                Err(error) if !refused(&error) => Err(Error::Transport(error.to_string())),
484                _ => Ok(()),
485            },
486            Reliability::Confirmable => {
487                let reply = self.exchange(id, &bytes, &socket).await?;
488                answered(reply.code, &reply.payload)
489            }
490        }
491    }
492
493    /// Registers an observation of a resource path, so its notifications arrive.
494    ///
495    /// A notification carries no path of its own, only the token of the request
496    /// that registered it (RFC 7641 section 3.2), so the client remembers which
497    /// path each token observes and delivers every notification under that path.
498    /// Registering a path again reuses its token, which a server takes as renewing
499    /// the observation rather than adding a second one (RFC 7641 section 4.1), so a
500    /// node can register again now and then in case the server has restarted.
501    ///
502    /// # Returns
503    ///
504    /// `Ok(())` once the server has registered the observation, or has acknowledged
505    /// the request with its response to follow.
506    ///
507    /// # Errors
508    ///
509    /// Returns [`Error::Transport`] if the server refuses the request, answers it
510    /// without registering an observation, or never acknowledges it, and
511    /// [`Error::Closed`] if the transport is not connected.
512    async fn subscribe(&mut self, topic: &str) -> Result<()> {
513        let socket = self.socket.clone().ok_or(Error::Closed)?;
514        let id = self.next_message_id();
515        let path = topic.trim_matches('/');
516        let observed = self
517            .observations
518            .lock()
519            .expect("observations lock")
520            .iter()
521            .find(|(_, observation)| observation.path == path)
522            .map(|(token, _)| token.clone());
523        let token = observed.unwrap_or_else(|| self.next_request_token());
524
525        let mut packet = Packet::new();
526        packet.header.set_version(1);
527        packet.header.set_type(MessageType::Confirmable);
528        packet.header.code = MessageClass::Request(RequestType::Get);
529        packet.header.message_id = id;
530        packet.set_token(token.clone());
531        // An empty observe option value registers the observation (RFC 7641).
532        packet.add_option(CoapOption::Observe, Vec::new());
533        add_path(&mut packet, topic);
534
535        let bytes = packet
536            .to_bytes()
537            .map_err(|err| Error::Codec(err.to_string()))?;
538
539        self.observations.lock().expect("observations lock").insert(
540            token.clone(),
541            Observation {
542                path: topic.trim_matches('/').to_owned(),
543                freshest: None,
544            },
545        );
546        let registered = match self.exchange(id, &bytes, &socket).await {
547            Ok(reply) => answered(reply.code, &reply.payload).and_then(|()| {
548                if reply.code == MessageClass::Empty || reply.observe.is_some() {
549                    Ok(())
550                } else {
551                    Err(Error::Transport(format!(
552                        "the server answered {topic} without registering an observation, so \
553                         no changes will follow"
554                    )))
555                }
556            }),
557            Err(error) => Err(error),
558        };
559        if registered.is_err() {
560            self.observations
561                .lock()
562                .expect("observations lock")
563                .remove(&token);
564        }
565        registered
566    }
567}
568
569impl Receive for CoapTransport {
570    /// Awaits the next notification from an observed resource.
571    ///
572    /// # Returns
573    ///
574    /// `Some(message)` for the next queued notification, or `None` once the
575    /// background task has stopped and no further messages will arrive.
576    ///
577    /// # Errors
578    ///
579    /// Returns [`Error::Closed`] if the transport is not connected.
580    async fn recv(&mut self) -> Result<Option<Message>> {
581        let incoming = self.incoming.as_mut().ok_or(Error::Closed)?;
582        Ok(incoming.recv().await)
583    }
584}
585
586/// Maps a [`Reliability`] onto the CoAP message type used on the wire.
587fn message_type(reliability: Reliability) -> MessageType {
588    match reliability {
589        Reliability::NonConfirmable => MessageType::NonConfirmable,
590        Reliability::Confirmable => MessageType::Confirmable,
591    }
592}
593
594/// Adds one `Uri-Path` option per non-empty segment of `topic`.
595fn add_path(packet: &mut Packet, topic: &str) {
596    for segment in topic.split('/').filter(|segment| !segment.is_empty()) {
597        packet.add_option(CoapOption::UriPath, segment.as_bytes().to_vec());
598    }
599}
600
601/// Reconstructs a resource path from a packet's `Uri-Path` options.
602fn path_from_packet(packet: &Packet) -> String {
603    match packet.get_option(CoapOption::UriPath) {
604        Some(segments) => segments
605            .iter()
606            .map(|segment| String::from_utf8_lossy(segment).into_owned())
607            .collect::<Vec<_>>()
608            .join("/"),
609        None => String::new(),
610    }
611}
612
613/// Routes one decoded packet, returning `false` when the inbound queue is gone.
614///
615/// An acknowledgment that registers an observation queues the resource's current
616/// state before it wakes the request, so the state is already waiting when
617/// `subscribe` returns, and a registration made straight after it starts from that
618/// state rather than racing it.
619async fn dispatch(
620    packet: Packet,
621    pending: &PendingAcks,
622    observations: &Observations,
623    tx: &mpsc::UnboundedSender<Message>,
624    socket: &UdpSocket,
625) -> bool {
626    let observe = observe_value(&packet);
627    match packet.header.get_type() {
628        MessageType::Acknowledgement => {
629            let message_id = packet.header.message_id;
630            let reply = Reply {
631                code: packet.header.code,
632                payload: packet.payload.clone(),
633                observe,
634            };
635            let delivered = match observe {
636                Some(sequence) if is_success(reply.code) => {
637                    notify(packet, sequence, observations, tx)
638                }
639                _ => Delivered::Taken,
640            };
641            if let Some(waiter) = pending.lock().expect("pending lock").remove(&message_id) {
642                let _ = waiter.send(Answer::Acknowledged(reply));
643            }
644            delivered != Delivered::Gone
645        }
646        MessageType::Reset => {
647            if let Some(waiter) = pending
648                .lock()
649                .expect("pending lock")
650                .remove(&packet.header.message_id)
651            {
652                let _ = waiter.send(Answer::Reset);
653            }
654            true
655        }
656        kind => {
657            let confirmable = kind == MessageType::Confirmable;
658            let message_id = packet.header.message_id;
659            let delivered = match (packet.header.code, observe) {
660                (MessageClass::Response(_), Some(sequence)) if is_success(packet.header.code) => {
661                    notify(packet, sequence, observations, tx)
662                }
663                (MessageClass::Response(_), _) => end(&packet, observations),
664                _ => Delivered::Unknown,
665            };
666            match delivered {
667                Delivered::Unknown if confirmable => reset(message_id, socket).await,
668                Delivered::Taken if confirmable => acknowledge(message_id, socket).await,
669                _ => {}
670            }
671            delivered != Delivered::Gone
672        }
673    }
674}
675
676/// Forgets the observation a response without a sequence number ends.
677///
678/// A server that can no longer notify, such as when the resource is deleted,
679/// answers with a response carrying no Observe option and removes the client from
680/// its observers (RFC 7641 section 3.2), so the client forgets the token too.
681fn end(packet: &Packet, observations: &Observations) -> Delivered {
682    match observations
683        .lock()
684        .expect("observations lock")
685        .remove(packet.get_token())
686    {
687        Some(_) => Delivered::Taken,
688        None => Delivered::Unknown,
689    }
690}
691
692/// What became of a notification.
693#[derive(PartialEq, Eq)]
694enum Delivered {
695    /// It was queued for a receive, or dropped as older than one already queued.
696    Taken,
697    /// Its token matches no observation, so the client does not want it.
698    Unknown,
699    /// The queue behind the receive is gone.
700    Gone,
701}
702
703/// Queues a notification under the path its token observes.
704///
705/// One no fresher than the last, by the rule of RFC 7641 section 3.4, is dropped
706/// rather than delivered out of order.
707fn notify(
708    packet: Packet,
709    sequence: u32,
710    observations: &Observations,
711    tx: &mpsc::UnboundedSender<Message>,
712) -> Delivered {
713    let path = {
714        let mut observations = observations.lock().expect("observations lock");
715        let Some(observation) = observations.get_mut(packet.get_token()) else {
716            return Delivered::Unknown;
717        };
718        let now = Instant::now();
719        if !fresher(observation.freshest, sequence, now) {
720            return Delivered::Taken;
721        }
722        observation.freshest = Some((sequence, now));
723        observation.path.clone()
724    };
725    match tx.send(Message::new(path, packet.payload)) {
726        Ok(()) => Delivered::Taken,
727        Err(_) => Delivered::Gone,
728    }
729}
730
731/// Whether a notification numbered `incoming`, arriving at `now`, is newer than
732/// the freshest so far, by the rule of RFC 7641 section 3.4.
733fn fresher(freshest: Option<(u32, Instant)>, incoming: u32, now: Instant) -> bool {
734    const HALF: u32 = 1 << 23;
735    let Some((known, at)) = freshest else {
736        return true;
737    };
738    (known < incoming && incoming - known < HALF)
739        || (known > incoming && known - incoming > HALF)
740        || now > at + Duration::from_secs(128)
741}
742
743/// Reads the Observe option's value, a big-endian integer of up to three bytes.
744fn observe_value(packet: &Packet) -> Option<u32> {
745    let value = packet.get_option(CoapOption::Observe)?.front()?;
746    Some(
747        value
748            .iter()
749            .fold(0u32, |sum, byte| (sum << 8) | u32::from(*byte)),
750    )
751}
752
753/// Whether a code is a 2.xx success response.
754fn is_success(code: MessageClass) -> bool {
755    u8::from(code) >> 5 == 2
756}
757
758/// Sends an empty acknowledgment for a confirmable message.
759async fn acknowledge(message_id: u16, socket: &UdpSocket) {
760    empty(MessageType::Acknowledgement, message_id, socket).await;
761}
762
763/// Rejects a message with a Reset, which RFC 7641 section 3.6 also makes the way a
764/// client says it no longer wants an observation's notifications.
765async fn reset(message_id: u16, socket: &UdpSocket) {
766    empty(MessageType::Reset, message_id, socket).await;
767}
768
769/// Sends an empty message of the given type, answering a message id.
770async fn empty(kind: MessageType, message_id: u16, socket: &UdpSocket) {
771    let mut packet = Packet::new();
772    packet.header.set_version(1);
773    packet.header.set_type(kind);
774    packet.header.code = MessageClass::Empty;
775    packet.header.message_id = message_id;
776    if let Ok(bytes) = packet.to_bytes() {
777        let _ = socket.send(&bytes).await;
778    }
779}
780
781#[cfg(test)]
782mod tests {
783    use super::*;
784
785    #[test]
786    fn reliability_defaults_to_confirmable() {
787        let config = CoapConfig::new("localhost", 5683);
788        assert_eq!(config.reliability, Reliability::Confirmable);
789    }
790
791    #[test]
792    fn setters_update_the_configuration() {
793        let config = CoapConfig::new("localhost", 5683)
794            .reliability(Reliability::NonConfirmable)
795            .ack_timeout(Duration::from_millis(250))
796            .max_retransmits(1)
797            .bind("127.0.0.1:0");
798        assert_eq!(config.reliability, Reliability::NonConfirmable);
799        assert_eq!(config.ack_timeout, Duration::from_millis(250));
800        assert_eq!(config.max_retransmits, 1);
801        assert_eq!(config.bind, "127.0.0.1:0");
802    }
803
804    #[test]
805    fn a_notification_is_fresher_by_the_rule_of_rfc_7641_section_3_4() {
806        let then = Instant::now();
807        let soon = then + Duration::from_secs(1);
808        assert!(
809            fresher(None, 0, then),
810            "the first notification is always fresh"
811        );
812        assert!(fresher(Some((1, then)), 2, soon));
813        assert!(
814            !fresher(Some((2, then)), 1, soon),
815            "an older number is stale"
816        );
817        assert!(!fresher(Some((2, then)), 2, soon), "a repeat is stale");
818        assert!(
819            fresher(Some((0x00FF_FFFF, then)), 0, soon),
820            "the 24-bit number wraps"
821        );
822        assert!(
823            !fresher(Some((0, then)), 0x0080_0001, soon),
824            "a jump past half the space reads as older"
825        );
826        assert!(
827            fresher(Some((5, then)), 1, then + Duration::from_secs(129)),
828            "after 128 seconds any notification is fresher"
829        );
830    }
831
832    #[test]
833    fn an_observe_value_is_a_big_endian_integer_of_up_to_three_bytes() {
834        let mut packet = Packet::new();
835        packet.add_option(CoapOption::Observe, vec![0x01, 0x02, 0x03]);
836        assert_eq!(observe_value(&packet), Some(0x0001_0203));
837        let mut empty = Packet::new();
838        empty.add_option(CoapOption::Observe, Vec::new());
839        assert_eq!(
840            observe_value(&empty),
841            Some(0),
842            "a zero-length value is zero"
843        );
844        assert_eq!(observe_value(&Packet::new()), None);
845    }
846
847    #[test]
848    fn path_round_trips_through_uri_path_options() {
849        let mut packet = Packet::new();
850        add_path(&mut packet, "sensors/1/temperature");
851        assert_eq!(path_from_packet(&packet), "sensors/1/temperature");
852    }
853
854    #[test]
855    fn leading_and_repeated_slashes_are_ignored() {
856        let mut packet = Packet::new();
857        add_path(&mut packet, "/sensors//1/");
858        assert_eq!(path_from_packet(&packet), "sensors/1");
859    }
860
861    #[tokio::test]
862    async fn send_before_connect_reports_closed() {
863        let mut transport = CoapTransport::new(CoapConfig::new("localhost", 5683));
864        assert!(matches!(
865            transport.send("t", b"x").await,
866            Err(Error::Closed)
867        ));
868    }
869
870    #[tokio::test]
871    async fn subscribe_before_connect_reports_closed() {
872        let mut transport = CoapTransport::new(CoapConfig::new("localhost", 5683));
873        assert!(matches!(transport.subscribe("t").await, Err(Error::Closed)));
874    }
875
876    #[tokio::test]
877    async fn recv_before_connect_reports_closed() {
878        let mut transport = CoapTransport::new(CoapConfig::new("localhost", 5683));
879        assert!(matches!(transport.recv().await, Err(Error::Closed)));
880    }
881
882    #[test]
883    fn a_registration_queues_its_state_before_the_request_wakes() {
884        let pending: PendingAcks = Arc::default();
885        let observations: Observations = Arc::default();
886        let (waiter, mut answer) = oneshot::channel();
887        pending.lock().expect("pending lock").insert(9, waiter);
888        observations.lock().expect("observations lock").insert(
889            vec![1, 2],
890            Observation {
891                path: "commands/valve".to_owned(),
892                freshest: None,
893            },
894        );
895        let mut ack = Packet::new();
896        ack.header.set_version(1);
897        ack.header.set_type(MessageType::Acknowledgement);
898        ack.header.code = MessageClass::Response(coap_lite::ResponseType::Content);
899        ack.header.message_id = 9;
900        ack.set_token(vec![1, 2]);
901        ack.add_option(CoapOption::Observe, vec![5]);
902        ack.payload = b"closed".to_vec();
903        let (tx, mut queue) = mpsc::unbounded_channel();
904
905        let held = observations.lock().expect("observations lock");
906        let pump = std::thread::spawn({
907            let pending = Arc::clone(&pending);
908            let observations = Arc::clone(&observations);
909            move || {
910                tokio::runtime::Builder::new_current_thread()
911                    .enable_all()
912                    .build()
913                    .expect("a runtime")
914                    .block_on(async {
915                        let socket = UdpSocket::bind("127.0.0.1:0").await.expect("bind");
916                        dispatch(ack, &pending, &observations, &tx, &socket).await
917                    })
918            }
919        });
920        std::thread::sleep(Duration::from_millis(100));
921        assert!(
922            matches!(answer.try_recv(), Err(oneshot::error::TryRecvError::Empty)),
923            "the request woke before the state it registered was queued"
924        );
925        drop(held);
926
927        assert!(pump.join().expect("the pump"), "the queue is still open");
928        assert!(matches!(answer.try_recv(), Ok(Answer::Acknowledged(_))));
929        let state = queue.try_recv().expect("the current state");
930        assert_eq!(state.topic, "commands/valve");
931        assert_eq!(state.payload, b"closed");
932    }
933
934    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
935    async fn confirmable_send_without_a_server_times_out() {
936        let config = CoapConfig::new("127.0.0.1", 1)
937            .ack_timeout(Duration::from_millis(20))
938            .max_retransmits(1);
939        let mut transport = CoapTransport::new(config);
940        transport.connect().await.expect("bind socket");
941        assert!(transport.send("sensors/1", b"x").await.is_err());
942    }
943}
pamoja_coap/lib.rs

pamoja_coap/
lib.rs
