pamoja_coap/
server.rs

1//! A CoAP server: the end that nodes report their readings to and take their
2//! commands from.
3
4use std::collections::HashMap;
5use std::net::SocketAddr;
6use std::sync::{Arc, Mutex};
7use std::time::{Duration, Instant};
8
9use coap_lite::{CoapOption, MessageClass, MessageType, Packet, RequestType, ResponseType};
10use pamoja_core::{topic_matches, Error, Message, Receive, Result, Transport};
11use tokio::net::UdpSocket;
12use tokio::sync::mpsc;
13use tokio::task::JoinHandle;
14
15use crate::{observe_value, path_from_packet, random};
16
17/// How long the server remembers the answer to a confirmable request, so that a
18/// retransmission of it is answered again rather than acted on twice. It is
19/// EXCHANGE_LIFETIME, RFC 7252 section 4.8.2.
20const EXCHANGE_LIFETIME: Duration = Duration::from_secs(247);
21
22/// The largest datagram the server reads; a larger one is cut short and fails to
23/// decode.
24const DATAGRAM: usize = 1500;
25
26/// A CoAP server that implements the core [`Transport`] and [`Receive`] traits.
27///
28/// It is the far end of a [`CoapTransport`](crate::CoapTransport): a gateway that
29/// nodes send their readings to and observe their commands on. A PUT or POST to a
30/// path matching one of its [`subscribe`](Transport::subscribe) filters is answered
31/// 2.04 Changed and delivered to [`recv`](Receive::recv), and one to any other path
32/// is answered 4.04 Not Found. [`send`](Transport::send) sets the state of a
33/// resource, which a GET reads and every observer of the path is notified of, as
34/// RFC 7641 describes.
35///
36/// # Examples
37///
38/// ```no_run
39/// use pamoja_coap::CoapServer;
40/// use pamoja_core::{Receive, Transport};
41///
42/// # async fn run() -> pamoja_core::Result<()> {
43/// let mut gateway = CoapServer::new("0.0.0.0:5683");
44/// gateway.connect().await?;
45/// gateway.subscribe("sensors/#").await?;
46/// gateway.send_text("commands/valve", "closed").await?;
47///
48/// while let Some(reading) = gateway.recv().await? {
49///     println!("{}: {}", reading.topic, reading.text()?);
50/// }
51/// # Ok(())
52/// # }
53/// ```
54pub struct CoapServer {
55    bind: String,
56    shared: Arc<Shared>,
57    incoming: Option<mpsc::UnboundedReceiver<Message>>,
58    pump: Option<JoinHandle<()>>,
59}
60
61/// A handle that sets the state of a server's resources from anywhere, while
62/// another task waits on the server's [`recv`](Receive::recv).
63///
64/// A gateway that takes readings in one loop and sends commands from another holds
65/// one of these for the commands, since a receive holds the server itself.
66#[derive(Clone)]
67pub struct CoapPublisher {
68    shared: Arc<Shared>,
69}
70
71/// What a server and its publishers share.
72struct Shared {
73    /// The bound socket, while the server is connected.
74    socket: Mutex<Option<Arc<UdpSocket>>>,
75    /// The resources, the observers, and the rest of the server's state.
76    state: Mutex<State>,
77}
78
79/// What the server holds between datagrams.
80struct State {
81    /// The filters a PUT or POST path has to match to be taken.
82    filters: Vec<String>,
83    /// The state of each resource, by path.
84    resources: HashMap<String, Vec<u8>>,
85    /// The observers of each resource, by path.
86    observers: HashMap<String, Vec<Observer>>,
87    /// The answer sent to each recent confirmable request, by sender and message id.
88    answered: HashMap<(SocketAddr, u16), (Instant, Vec<u8>)>,
89    /// The next message id for a message the server starts.
90    next_id: u16,
91    /// The number of the next notification, of which RFC 7641 section 4.4 sends the
92    /// low 24 bits.
93    sequence: u32,
94}
95
96/// One client observing one resource.
97struct Observer {
98    /// Where its notifications go.
99    peer: SocketAddr,
100    /// The token of its registration, which every notification carries back.
101    token: Vec<u8>,
102    /// The message id of the last notification sent to it, which a Reset names.
103    last_id: u16,
104}
105
106impl CoapServer {
107    /// Creates a server that will listen on a local address, without binding it.
108    ///
109    /// # Arguments
110    ///
111    /// * `bind` - the local `host:port`, such as `"0.0.0.0:5683"` for the plaintext
112    ///   CoAP port on every interface, or port `0` for a free one.
113    ///
114    /// # Returns
115    ///
116    /// A server ready for [`connect`](Transport::connect).
117    pub fn new(bind: impl Into<String>) -> Self {
118        Self {
119            bind: bind.into(),
120            shared: Arc::new(Shared {
121                socket: Mutex::new(None),
122                state: Mutex::new(State {
123                    filters: Vec::new(),
124                    resources: HashMap::new(),
125                    observers: HashMap::new(),
126                    answered: HashMap::new(),
127                    next_id: random() as u16,
128                    sequence: 0,
129                }),
130            }),
131            incoming: None,
132            pump: None,
133        }
134    }
135
136    /// Takes a handle that publishes resource states while a receive waits.
137    ///
138    /// # Returns
139    ///
140    /// A publisher sharing this server's resources and socket.
141    pub fn publisher(&self) -> CoapPublisher {
142        CoapPublisher {
143            shared: Arc::clone(&self.shared),
144        }
145    }
146
147    /// Reports whether the server holds a bound socket.
148    ///
149    /// # Returns
150    ///
151    /// `true` from a successful [`connect`](Transport::connect) until
152    /// [`disconnect`](CoapServer::disconnect).
153    pub fn is_connected(&self) -> bool {
154        self.shared.socket().is_some()
155    }
156
157    /// The address the server listens on, which names the port the system chose
158    /// when it bound port `0`.
159    ///
160    /// # Returns
161    ///
162    /// The local address, or `None` before [`connect`](Transport::connect).
163    pub fn local_addr(&self) -> Option<SocketAddr> {
164        self.shared.socket()?.local_addr().ok()
165    }
166
167    /// Counts the clients observing a resource.
168    ///
169    /// # Arguments
170    ///
171    /// * `path` - the resource's path.
172    ///
173    /// # Returns
174    ///
175    /// How many observers the resource has.
176    pub fn observers(&self, path: &str) -> usize {
177        self.shared.observers(path)
178    }
179
180    /// Closes the socket and stops the background task.
181    ///
182    /// The resources and filters are kept for the next
183    /// [`connect`](Transport::connect), and the observers are dropped, since their
184    /// registrations were made to the socket that closed.
185    ///
186    /// # Returns
187    ///
188    /// `Ok(())` once the socket is released.
189    ///
190    /// # Errors
191    ///
192    /// This call is best-effort and always returns `Ok(())`.
193    pub async fn disconnect(&mut self) -> Result<()> {
194        if let Some(pump) = self.pump.take() {
195            pump.abort();
196        }
197        *self.shared.socket.lock().expect("socket lock") = None;
198        self.incoming = None;
199        self.shared.state().observers.clear();
200        Ok(())
201    }
202}
203
204impl CoapPublisher {
205    /// Sets the state of a resource and notifies every observer of it.
206    ///
207    /// # Arguments
208    ///
209    /// * `path` - the resource's path.
210    /// * `payload` - its new state, which a GET now reads.
211    ///
212    /// # Returns
213    ///
214    /// `Ok(())` once each observer's notification has left. An observer that has
215    /// gone does not stop the others' notifications.
216    ///
217    /// # Errors
218    ///
219    /// Returns [`Error::Closed`] if the server is not connected, or
220    /// [`Error::Transport`] if the socket failed for a reason other than an observer
221    /// that has gone.
222    pub async fn publish(&self, path: &str, payload: &[u8]) -> Result<()> {
223        let socket = self.shared.socket().ok_or(Error::Closed)?;
224        let out = {
225            let mut state = self.shared.state();
226            let path = path.trim_matches('/').to_owned();
227            state.resources.insert(path.clone(), payload.to_vec());
228            state.notify(&path, payload)
229        };
230        let mut failure = None;
231        for (to, bytes) in out {
232            if let Err(error) = socket.send_to(&bytes, to).await {
233                if !crate::refused(&error) {
234                    failure.get_or_insert(Error::Transport(error.to_string()));
235                }
236            }
237        }
238        failure.map_or(Ok(()), Err)
239    }
240
241    /// Counts the clients observing a resource.
242    ///
243    /// # Arguments
244    ///
245    /// * `path` - the resource's path.
246    ///
247    /// # Returns
248    ///
249    /// How many observers the resource has.
250    pub fn observers(&self, path: &str) -> usize {
251        self.shared.observers(path)
252    }
253
254    /// Reports whether the server holds a bound socket.
255    ///
256    /// # Returns
257    ///
258    /// `true` while the server is connected.
259    pub fn is_connected(&self) -> bool {
260        self.shared.socket().is_some()
261    }
262
263    /// The address the server listens on.
264    ///
265    /// # Returns
266    ///
267    /// The local address, or `None` while the server is not connected.
268    pub fn local_addr(&self) -> Option<SocketAddr> {
269        self.shared.socket()?.local_addr().ok()
270    }
271}
272
273impl Shared {
274    /// The bound socket, while the server is connected.
275    fn socket(&self) -> Option<Arc<UdpSocket>> {
276        self.socket.lock().expect("socket lock").clone()
277    }
278
279    /// The server's state, locked.
280    fn state(&self) -> std::sync::MutexGuard<'_, State> {
281        self.state.lock().expect("server lock")
282    }
283
284    /// Counts the observers of a resource.
285    fn observers(&self, path: &str) -> usize {
286        self.state()
287            .observers
288            .get(path.trim_matches('/'))
289            .map_or(0, Vec::len)
290    }
291}
292
293impl Transport for CoapServer {
294    /// Binds the server's socket and starts answering requests.
295    ///
296    /// # Returns
297    ///
298    /// `Ok(())` once the socket is bound.
299    ///
300    /// # Errors
301    ///
302    /// Returns [`Error::Transport`] if the address cannot be bound, such as a port
303    /// another program holds.
304    async fn connect(&mut self) -> Result<()> {
305        self.disconnect().await?;
306        let socket = Arc::new(
307            UdpSocket::bind(&self.bind)
308                .await
309                .map_err(|err| Error::Transport(format!("could not bind {}: {err}", self.bind)))?,
310        );
311        let (tx, rx) = mpsc::unbounded_channel();
312        let shared = Arc::clone(&self.shared);
313        let pump_socket = Arc::clone(&socket);
314        let pump = tokio::spawn(async move {
315            let mut buf = vec![0u8; DATAGRAM];
316            loop {
317                let (len, peer) = match pump_socket.recv_from(&mut buf).await {
318                    Ok(received) => received,
319                    Err(error) if crate::refused(&error) => continue,
320                    Err(_) => break,
321                };
322                let Ok(packet) = Packet::from_bytes(&buf[..len]) else {
323                    continue;
324                };
325                let out = handle(&packet, peer, &shared.state, &tx);
326                for (to, bytes) in out {
327                    let _ = pump_socket.send_to(&bytes, to).await;
328                }
329            }
330        });
331        *self.shared.socket.lock().expect("socket lock") = Some(socket);
332        self.incoming = Some(rx);
333        self.pump = Some(pump);
334        Ok(())
335    }
336
337    /// Sets the state of a resource and notifies every observer of it, as
338    /// [`CoapPublisher::publish`] does.
339    ///
340    /// # Arguments
341    ///
342    /// * `topic` - the resource's path.
343    /// * `payload` - its new state, which a GET now reads.
344    ///
345    /// # Returns
346    ///
347    /// `Ok(())` once each observer's notification has left.
348    ///
349    /// # Errors
350    ///
351    /// Returns [`Error::Closed`] if the server is not connected.
352    async fn send(&mut self, topic: &str, payload: &[u8]) -> Result<()> {
353        self.publisher().publish(topic, payload).await
354    }
355
356    /// Takes the readings sent to the paths a filter matches.
357    ///
358    /// # Arguments
359    ///
360    /// * `topic` - the filter, with `+` for one level and `#` for the rest, as
361    ///   [`topic_matches`] reads them.
362    ///
363    /// # Returns
364    ///
365    /// `Ok(())` once the filter is in place. A server takes readings for its filters
366    /// whether or not it is connected.
367    async fn subscribe(&mut self, topic: &str) -> Result<()> {
368        self.shared
369            .state()
370            .filters
371            .push(topic.trim_matches('/').to_owned());
372        Ok(())
373    }
374}
375
376impl Receive for CoapServer {
377    /// Awaits the next reading sent to a path the server takes.
378    ///
379    /// # Returns
380    ///
381    /// `Some(message)` for the next reading, carrying the path it was sent to.
382    ///
383    /// # Errors
384    ///
385    /// Returns [`Error::Closed`] if the server is not connected.
386    async fn recv(&mut self) -> Result<Option<Message>> {
387        let incoming = self.incoming.as_mut().ok_or(Error::Closed)?;
388        Ok(incoming.recv().await)
389    }
390}
391
392impl State {
393    /// Takes the next message id for a message the server starts.
394    fn message_id(&mut self) -> u16 {
395        let id = self.next_id;
396        self.next_id = self.next_id.wrapping_add(1);
397        id
398    }
399
400    /// Takes the value of the next notification's Observe option.
401    fn next_sequence(&mut self) -> u32 {
402        self.sequence = (self.sequence + 1) & 0x00FF_FFFF;
403        self.sequence
404    }
405
406    /// Builds a notification of a resource's new state for each of its observers.
407    fn notify(&mut self, path: &str, payload: &[u8]) -> Vec<(SocketAddr, Vec<u8>)> {
408        let sequence = self.next_sequence();
409        let count = self.observers.get(path).map_or(0, Vec::len);
410        let ids: Vec<u16> = (0..count).map(|_| self.message_id()).collect();
411        let Some(observers) = self.observers.get_mut(path) else {
412            return Vec::new();
413        };
414        observers
415            .iter_mut()
416            .zip(ids)
417            .filter_map(|(observer, id)| {
418                observer.last_id = id;
419                let mut packet = response(ResponseType::Content, payload);
420                packet.header.set_type(MessageType::NonConfirmable);
421                packet.header.message_id = id;
422                packet.set_token(observer.token.clone());
423                packet.add_option(CoapOption::Observe, observe_bytes(sequence));
424                Some((observer.peer, packet.to_bytes().ok()?))
425            })
426            .collect()
427    }
428}
429
430/// Answers one datagram, returning what to send and where.
431fn handle(
432    packet: &Packet,
433    peer: SocketAddr,
434    state: &Mutex<State>,
435    tx: &mpsc::UnboundedSender<Message>,
436) -> Vec<(SocketAddr, Vec<u8>)> {
437    let mut state = state.lock().expect("server lock");
438    let id = packet.header.message_id;
439    match (packet.header.get_type(), packet.header.code) {
440        (MessageType::Reset, _) => {
441            for observers in state.observers.values_mut() {
442                observers.retain(|observer| !(observer.peer == peer && observer.last_id == id));
443            }
444            Vec::new()
445        }
446        (MessageType::Confirmable, MessageClass::Empty) => {
447            vec![(peer, empty(MessageType::Reset, id))]
448        }
449        (
450            kind @ (MessageType::Confirmable | MessageType::NonConfirmable),
451            MessageClass::Request(method),
452        ) => {
453            let confirmable = kind == MessageType::Confirmable;
454            if confirmable {
455                let now = Instant::now();
456                state
457                    .answered
458                    .retain(|_, (at, _)| now.duration_since(*at) < EXCHANGE_LIFETIME);
459                if let Some((_, bytes)) = state.answered.get(&(peer, id)) {
460                    return vec![(peer, bytes.clone())];
461                }
462            }
463            let (mut answer, mut out) = request(&mut state, packet, method, peer, tx);
464            answer.set_token(packet.get_token().to_vec());
465            if confirmable {
466                answer.header.set_type(MessageType::Acknowledgement);
467                answer.header.message_id = id;
468            } else {
469                answer.header.set_type(MessageType::NonConfirmable);
470                answer.header.message_id = state.message_id();
471            }
472            if let Ok(bytes) = answer.to_bytes() {
473                if confirmable {
474                    state
475                        .answered
476                        .insert((peer, id), (Instant::now(), bytes.clone()));
477                }
478                out.insert(0, (peer, bytes));
479            }
480            out
481        }
482        (MessageType::Confirmable, _) => vec![(peer, empty(MessageType::Reset, id))],
483        _ => Vec::new(),
484    }
485}
486
487/// Carries out a request, returning the response and any notifications it causes.
488fn request(
489    state: &mut State,
490    packet: &Packet,
491    method: RequestType,
492    peer: SocketAddr,
493    tx: &mpsc::UnboundedSender<Message>,
494) -> (Packet, Vec<(SocketAddr, Vec<u8>)>) {
495    let path = path_from_packet(packet);
496    match method {
497        RequestType::Get => {
498            let Some(current) = state.resources.get(&path).cloned() else {
499                return (response(ResponseType::NotFound, &[]), Vec::new());
500            };
501            let token = packet.get_token().to_vec();
502            let mut answer = response(ResponseType::Content, &current);
503            match observe_value(packet) {
504                Some(0) => {
505                    let observers = state.observers.entry(path).or_default();
506                    observers
507                        .retain(|observer| !(observer.peer == peer && observer.token == token));
508                    observers.push(Observer {
509                        peer,
510                        token,
511                        last_id: 0,
512                    });
513                    answer.add_option(CoapOption::Observe, observe_bytes(state.sequence));
514                }
515                Some(1) => {
516                    if let Some(observers) = state.observers.get_mut(&path) {
517                        observers
518                            .retain(|observer| !(observer.peer == peer && observer.token == token));
519                    }
520                }
521                _ => {}
522            }
523            (answer, Vec::new())
524        }
525        RequestType::Put | RequestType::Post => {
526            if !state
527                .filters
528                .iter()
529                .any(|filter| topic_matches(filter, &path))
530            {
531                return (response(ResponseType::NotFound, &[]), Vec::new());
532            }
533            let _ = tx.send(Message::new(path.clone(), packet.payload.clone()));
534            state.resources.insert(path.clone(), packet.payload.clone());
535            let notifications = state.notify(&path, &packet.payload);
536            (response(ResponseType::Changed, &[]), notifications)
537        }
538        _ => (response(ResponseType::MethodNotAllowed, &[]), Vec::new()),
539    }
540}
541
542/// Builds a response with a code and a payload, its type, id, and token still to set.
543fn response(code: ResponseType, payload: &[u8]) -> Packet {
544    let mut packet = Packet::new();
545    packet.header.set_version(1);
546    packet.header.code = MessageClass::Response(code);
547    packet.payload = payload.to_vec();
548    packet
549}
550
551/// Encodes an empty message of a type, answering a message id.
552fn empty(kind: MessageType, message_id: u16) -> Vec<u8> {
553    let mut packet = Packet::new();
554    packet.header.set_version(1);
555    packet.header.set_type(kind);
556    packet.header.code = MessageClass::Empty;
557    packet.header.message_id = message_id;
558    packet.to_bytes().unwrap_or_default()
559}
560
561/// Encodes an Observe value as the shortest big-endian integer that holds it, with
562/// zero as no bytes at all (RFC 7252 section 3.2).
563fn observe_bytes(value: u32) -> Vec<u8> {
564    let bytes = value.to_be_bytes();
565    let first = bytes
566        .iter()
567        .position(|byte| *byte != 0)
568        .unwrap_or(bytes.len());
569    bytes[first..].to_vec()
570}
571
572#[cfg(test)]
573mod tests {
574    use super::*;
575
576    #[test]
577    fn an_observe_value_takes_the_fewest_bytes_that_hold_it() {
578        assert_eq!(observe_bytes(0), Vec::<u8>::new());
579        assert_eq!(observe_bytes(1), vec![1]);
580        assert_eq!(observe_bytes(0x0100), vec![1, 0]);
581        assert_eq!(observe_bytes(0x00FF_FFFF), vec![0xFF, 0xFF, 0xFF]);
582    }
583
584    #[test]
585    fn the_notification_number_wraps_at_24_bits() {
586        let mut state = State {
587            filters: Vec::new(),
588            resources: HashMap::new(),
589            observers: HashMap::new(),
590            answered: HashMap::new(),
591            next_id: 0,
592            sequence: 0x00FF_FFFF,
593        };
594        assert_eq!(state.next_sequence(), 0);
595        assert_eq!(state.next_sequence(), 1);
596    }
597}
pamoja_coap/server.rs

pamoja_coap/
server.rs
