pamoja_radios/
mesh.rs

1//! A pamoja transport over a LoRa radio, carrying topics in pamoja-mesh frames.
2//!
3//! [`MeshRadio`] turns a radio into a [`Transport`] and a [`Receive`]. A message goes out
4//! as a broadcast [`Frame`] whose payload is the topic's length in one byte, the topic, and
5//! the payload. Every node that hears it drops copies it has already seen, delivers what
6//! its subscriptions match, and relays the frame onward while hops remain, so a message
7//! crosses a mesh of radios that each hear only their neighbors. A [`DutyCycle`] holds the
8//! radio silent for the off time the region requires after each transmission, its own
9//! messages and its relays alike.
10//!
11//! The radio is driven from the task that awaits the transport. It is read every [`POLL`],
12//! with the tokio timer sleeping in between, so neither a frame's airtime nor a quiet
13//! channel blocks the runtime. Topic filters follow MQTT, through [`topic_matches`].
14//!
15//! # Examples
16//!
17//! Two nodes whose air is a queue each, with the frame carried across by hand:
18//!
19//! ```
20//! use std::collections::VecDeque;
21//! use std::convert::Infallible;
22//!
23//! use pamoja_core::{Receive, Transport};
24//! use pamoja_lora::LinkSettings;
25//! use pamoja_radios::mesh::{LoraRadio, MeshRadio};
26//!
27//! #[derive(Default)]
28//! struct Air(VecDeque<Vec<u8>>);
29//!
30//! impl LoraRadio for Air {
31//!     type Error = Infallible;
32//!
33//!     fn link(&self) -> Option<LinkSettings> {
34//!         Some(LinkSettings::new(7, 125_000))
35//!     }
36//!
37//!     fn start_transmit(&mut self, frame: &[u8]) -> Result<u64, Infallible> {
38//!         self.0.push_back(frame.to_vec());
39//!         Ok(LinkSettings::new(7, 125_000).airtime_us(frame.len()))
40//!     }
41//!
42//!     fn finish_transmit(&mut self) -> Result<bool, Infallible> {
43//!         Ok(true)
44//!     }
45//!
46//!     fn listen(&mut self) -> Result<(), Infallible> {
47//!         Ok(())
48//!     }
49//!
50//!     fn take_frame(&mut self, buffer: &mut [u8]) -> Result<Option<usize>, Infallible> {
51//!         Ok(self.0.pop_front().map(|frame| {
52//!             buffer[..frame.len()].copy_from_slice(&frame);
53//!             frame.len()
54//!         }))
55//!     }
56//! }
57//!
58//! # let runtime = tokio::runtime::Builder::new_current_thread().enable_time().build().unwrap();
59//! # runtime.block_on(async {
60//! // A soil sensor and a gateway, nodes 10 and 11, each on a 1% duty cycle, and the
61//! // gateway listening without relaying.
62//! let (sensor_node, gateway_node, one_percent) = (10, 11, 10);
63//! let mut sensor = MeshRadio::new(Air::default(), sensor_node, one_percent);
64//! let mut gateway = MeshRadio::new(Air::default(), gateway_node, one_percent).without_relaying();
65//! sensor.connect().await?;
66//! gateway.connect().await?;
67//! gateway.subscribe("garden/+/moisture").await?;
68//!
69//! sensor.send_text("garden/bed-1/moisture", "28.5").await?;
70//! let frame = sensor.radio_mut().0.pop_front().expect("the sensor transmitted");
71//! gateway.radio_mut().0.push_back(frame);
72//!
73//! let reading = gateway.recv().await?.expect("the gateway heard it");
74//! assert_eq!(reading.topic, "garden/bed-1/moisture");
75//! assert_eq!(reading.number()?, 28.5);
76//! # Ok::<(), pamoja_core::Error>(())
77//! # }).unwrap();
78//! ```
79
80use std::time::Duration;
81
82use embedded_hal::delay::DelayNs;
83use embedded_hal::digital::{InputPin, OutputPin};
84use embedded_hal::spi::SpiDevice;
85use pamoja_core::{topic_matches, Error, Message, Receive, Result, Transport};
86use pamoja_lora::LinkSettings;
87use pamoja_mesh::{DynamicSeenCache, Frame};
88use tokio::time::{sleep, Instant};
89
90use crate::duty::DutyCycle;
91use crate::radio::{self, Radio};
92use crate::sx126x::{RadioError, Reception, Sx126x};
93use crate::sx127x::{self, Sx127x};
94
95/// The most topic and payload bytes one message carries together: a frame's payload less
96/// the byte that holds the topic's length.
97pub const MAX_MESSAGE: usize = Frame::MAX_PAYLOAD - 1;
98
99/// How often the transport reads the radio for a received frame or a finished transmission.
100pub const POLL: Duration = Duration::from_millis(1);
101
102/// How long past a frame's airtime the transport waits for the radio to report it sent.
103pub const TRANSMIT_GRACE: Duration = Duration::from_secs(2);
104
105/// How many recent frames a node remembers, so copies arriving by other paths are dropped.
106pub const SEEN_CAPACITY: usize = 64;
107
108/// A radio a [`MeshRadio`] carries frames over.
109///
110/// Each method returns at once, so the transport can sleep on the runtime's timer between
111/// calls rather than block it. [`Sx126x`], [`Sx127x`], and [`Radio`] implement it, and a
112/// simulated radio implements the same five methods.
113pub trait LoraRadio {
114    /// What the radio reports when it or the bus under it fails.
115    type Error: core::fmt::Debug;
116
117    /// Returns the link settings frames go out with, which the duty cycle's arithmetic
118    /// needs.
119    ///
120    /// # Returns
121    ///
122    /// The settings, or `None` while the radio is unconfigured.
123    fn link(&self) -> Option<LinkSettings>;
124
125    /// Starts sending one frame, without waiting for it to leave.
126    ///
127    /// # Arguments
128    ///
129    /// * `frame` - the bytes to put on the air.
130    ///
131    /// # Returns
132    ///
133    /// The frame's airtime in microseconds.
134    ///
135    /// # Errors
136    ///
137    /// Whatever the radio reports.
138    fn start_transmit(&mut self, frame: &[u8]) -> std::result::Result<u64, Self::Error>;
139
140    /// Reports whether the frame [`start_transmit`](LoraRadio::start_transmit) began has
141    /// left.
142    ///
143    /// # Returns
144    ///
145    /// `true` once it has been sent.
146    ///
147    /// # Errors
148    ///
149    /// Whatever the radio reports, including a transmission that timed out.
150    fn finish_transmit(&mut self) -> std::result::Result<bool, Self::Error>;
151
152    /// Starts listening, and keeps listening frame after frame.
153    ///
154    /// # Errors
155    ///
156    /// Whatever the radio reports.
157    fn listen(&mut self) -> std::result::Result<(), Self::Error>;
158
159    /// Takes a frame the radio has received since the last call, dropping one whose CRC
160    /// failed.
161    ///
162    /// # Arguments
163    ///
164    /// * `buffer` - where the frame goes, 255 bytes long.
165    ///
166    /// # Returns
167    ///
168    /// The frame's length, or `None` when no good frame has arrived.
169    ///
170    /// # Errors
171    ///
172    /// Whatever the radio reports.
173    fn take_frame(&mut self, buffer: &mut [u8]) -> std::result::Result<Option<usize>, Self::Error>;
174}
175
176impl<SPI, BUSY, RESET, D> LoraRadio for Sx126x<SPI, BUSY, RESET, D>
177where
178    SPI: SpiDevice,
179    BUSY: InputPin,
180    RESET: OutputPin,
181    D: DelayNs,
182{
183    type Error = RadioError<SPI::Error>;
184
185    fn link(&self) -> Option<LinkSettings> {
186        self.config().map(|config| config.link)
187    }
188
189    fn start_transmit(&mut self, frame: &[u8]) -> std::result::Result<u64, Self::Error> {
190        Sx126x::start_transmit(self, frame)
191    }
192
193    fn finish_transmit(&mut self) -> std::result::Result<bool, Self::Error> {
194        Sx126x::finish_transmit(self)
195    }
196
197    fn listen(&mut self) -> std::result::Result<(), Self::Error> {
198        Sx126x::listen(self)
199    }
200
201    fn take_frame(&mut self, buffer: &mut [u8]) -> std::result::Result<Option<usize>, Self::Error> {
202        match Sx126x::take_frame(self, buffer)? {
203            Some(Reception::Frame { len, .. }) => Ok(Some(len)),
204            _ => Ok(None),
205        }
206    }
207}
208
209impl<SPI, RESET, D> LoraRadio for Sx127x<SPI, RESET, D>
210where
211    SPI: SpiDevice,
212    RESET: OutputPin,
213    D: DelayNs,
214{
215    type Error = sx127x::RadioError<SPI::Error>;
216
217    fn link(&self) -> Option<LinkSettings> {
218        self.config().map(|config| config.link)
219    }
220
221    fn start_transmit(&mut self, frame: &[u8]) -> std::result::Result<u64, Self::Error> {
222        Sx127x::start_transmit(self, frame)
223    }
224
225    fn finish_transmit(&mut self) -> std::result::Result<bool, Self::Error> {
226        Sx127x::finish_transmit(self)
227    }
228
229    fn listen(&mut self) -> std::result::Result<(), Self::Error> {
230        Sx127x::listen(self)
231    }
232
233    fn take_frame(&mut self, buffer: &mut [u8]) -> std::result::Result<Option<usize>, Self::Error> {
234        match Sx127x::take_frame(self, buffer)? {
235            Some(sx127x::Reception::Frame { len, .. }) => Ok(Some(len)),
236            _ => Ok(None),
237        }
238    }
239}
240
241impl<SPI, BUSY, RESET, D> LoraRadio for Radio<SPI, BUSY, RESET, D>
242where
243    SPI: SpiDevice,
244    BUSY: InputPin,
245    RESET: OutputPin,
246    D: DelayNs,
247{
248    type Error = radio::RadioError<SPI::Error>;
249
250    fn link(&self) -> Option<LinkSettings> {
251        Radio::link(self)
252    }
253
254    fn start_transmit(&mut self, frame: &[u8]) -> std::result::Result<u64, Self::Error> {
255        Radio::start_transmit(self, frame)
256    }
257
258    fn finish_transmit(&mut self) -> std::result::Result<bool, Self::Error> {
259        Radio::finish_transmit(self)
260    }
261
262    fn listen(&mut self) -> std::result::Result<(), Self::Error> {
263        Radio::listen(self)
264    }
265
266    fn take_frame(&mut self, buffer: &mut [u8]) -> std::result::Result<Option<usize>, Self::Error> {
267        match Radio::take_frame(self, buffer)? {
268            Some(radio::Reception::Frame { len, .. }) => Ok(Some(len)),
269            _ => Ok(None),
270        }
271    }
272}
273
274/// A node on a LoRa mesh: a radio, the node's address, and the rules it keeps.
275///
276/// Build it around a configured radio, [`connect`](Transport::connect) to start listening,
277/// then send and receive as over any other pamoja transport.
278pub struct MeshRadio<R> {
279    radio: R,
280    node: u32,
281    next_id: u16,
282    hop_limit: u8,
283    relay: bool,
284    duty: DutyCycle,
285    epoch: Instant,
286    filters: Vec<String>,
287    seen: DynamicSeenCache,
288    pending: Option<Message>,
289    connected: bool,
290    buffer: [u8; 255],
291}
292
293impl<R: LoraRadio> MeshRadio<R> {
294    /// Builds a node around a radio that is already configured.
295    ///
296    /// Frames start with [`Frame::DEFAULT_HOP_LIMIT`] hops, and the node relays what it
297    /// hears.
298    ///
299    /// # Arguments
300    ///
301    /// * `radio` - the radio, tuned to the channel the mesh uses.
302    /// * `node` - this node's address, which every frame it sends carries as its source.
303    /// * `duty_cycle_permille` - the region's duty-cycle limit in parts per thousand, such
304    ///   as `10` for 1%, or `1000` where the region sets none.
305    ///
306    /// # Returns
307    ///
308    /// The node, not yet connected.
309    pub fn new(radio: R, node: u32, duty_cycle_permille: u32) -> MeshRadio<R> {
310        MeshRadio {
311            radio,
312            node,
313            next_id: 0,
314            hop_limit: Frame::DEFAULT_HOP_LIMIT,
315            relay: true,
316            duty: DutyCycle::new(duty_cycle_permille),
317            epoch: Instant::now(),
318            filters: Vec::new(),
319            seen: DynamicSeenCache::new(SEEN_CAPACITY),
320            pending: None,
321            connected: false,
322            buffer: [0; 255],
323        }
324    }
325
326    /// Returns the node with another hop limit for the frames it sends.
327    ///
328    /// # Arguments
329    ///
330    /// * `hop_limit` - how many relays a frame may take; `0` keeps it to the nodes in range.
331    ///
332    /// # Returns
333    ///
334    /// The node.
335    pub fn with_hop_limit(mut self, hop_limit: u8) -> MeshRadio<R> {
336        self.hop_limit = hop_limit;
337        self
338    }
339
340    /// Returns the node with relaying off, so it only sends and receives its own traffic.
341    ///
342    /// # Returns
343    ///
344    /// The node.
345    pub fn without_relaying(mut self) -> MeshRadio<R> {
346        self.relay = false;
347        self
348    }
349
350    /// Returns this node's address.
351    ///
352    /// # Returns
353    ///
354    /// The address frames carry as their source.
355    pub fn node(&self) -> u32 {
356        self.node
357    }
358
359    /// Reports whether the node is connected and listening.
360    ///
361    /// # Returns
362    ///
363    /// `true` after [`connect`](Transport::connect).
364    pub fn is_connected(&self) -> bool {
365        self.connected
366    }
367
368    /// Returns the duty-cycle guard, to see how long the radio must still stay silent.
369    ///
370    /// # Returns
371    ///
372    /// The guard, whose clock is microseconds since the node was built.
373    pub fn duty_cycle(&self) -> &DutyCycle {
374        &self.duty
375    }
376
377    /// Returns the radio.
378    ///
379    /// # Returns
380    ///
381    /// A reference to the radio.
382    pub fn radio(&self) -> &R {
383        &self.radio
384    }
385
386    /// Returns the radio, to reach what the transport does not cover.
387    ///
388    /// # Returns
389    ///
390    /// A mutable reference to the radio.
391    pub fn radio_mut(&mut self) -> &mut R {
392        &mut self.radio
393    }
394
395    /// Gives back the radio.
396    ///
397    /// # Returns
398    ///
399    /// The radio.
400    pub fn release(self) -> R {
401        self.radio
402    }
403
404    fn now_us(&self) -> u64 {
405        u64::try_from(self.epoch.elapsed().as_micros()).unwrap_or(u64::MAX)
406    }
407
408    async fn transmit(&mut self, frame: &Frame) -> Result<()> {
409        let now_us = self.now_us();
410        let wait_us = self.duty.wait_us(now_us);
411        if wait_us > 0 {
412            return Err(Error::Transport(format!(
413                "the duty cycle keeps the radio silent for another {} ms",
414                wait_us.div_ceil(1000)
415            )));
416        }
417        let link = self
418            .radio
419            .link()
420            .ok_or_else(|| Error::Transport("the radio is not configured".to_owned()))?;
421        let bytes = frame.as_bytes();
422        let airtime_us = self.radio.start_transmit(bytes).map_err(radio_error)?;
423        self.duty.transmitted(now_us, &link, bytes.len());
424
425        let airtime = Duration::from_micros(airtime_us);
426        sleep(airtime).await;
427        let deadline = Instant::now() + TRANSMIT_GRACE;
428        while !self.radio.finish_transmit().map_err(radio_error)? {
429            if Instant::now() >= deadline {
430                return Err(Error::Transport(
431                    "the radio never reported the frame sent".to_owned(),
432                ));
433            }
434            sleep(POLL).await;
435        }
436        self.radio.listen().map_err(radio_error)
437    }
438
439    async fn accept(&mut self, len: usize) -> Result<()> {
440        let Ok(frame) = Frame::parse(&self.buffer[..len]) else {
441            return Ok(());
442        };
443        if frame.src() == self.node || !self.seen.record(frame.dedup_key()) {
444            return Ok(());
445        }
446        if let Some((topic, payload)) = decode(frame.payload()) {
447            if self
448                .filters
449                .iter()
450                .any(|filter| topic_matches(filter, topic))
451            {
452                self.pending = Some(Message::new(topic, payload));
453            }
454        }
455        if self.relay {
456            if let Some(onward) = frame.relayed() {
457                if self.duty.ready(self.now_us()) {
458                    self.transmit(&onward).await?;
459                }
460            }
461        }
462        Ok(())
463    }
464}
465
466impl<R: LoraRadio + Send> Transport for MeshRadio<R> {
467    async fn connect(&mut self) -> Result<()> {
468        self.radio.listen().map_err(radio_error)?;
469        self.connected = true;
470        Ok(())
471    }
472
473    async fn send(&mut self, topic: &str, payload: &[u8]) -> Result<()> {
474        if !self.connected {
475            return Err(Error::Closed);
476        }
477        let body = encode(topic, payload)?;
478        let frame = Frame::broadcast(self.node, self.next_id, &body)
479            .map_err(|error| Error::Transport(format!("mesh frame: {error:?}")))?
480            .with_hop_limit(self.hop_limit);
481        self.next_id = self.next_id.wrapping_add(1);
482        self.seen.record(frame.dedup_key());
483        self.transmit(&frame).await
484    }
485
486    async fn subscribe(&mut self, topic: &str) -> Result<()> {
487        if !self.connected {
488            return Err(Error::Closed);
489        }
490        self.filters.push(topic.to_owned());
491        Ok(())
492    }
493}
494
495impl<R: LoraRadio + Send> Receive for MeshRadio<R> {
496    /// Awaits the next message whose topic a subscription matches.
497    ///
498    /// A frame relayed on the way is sent before the message is handed up, and a message
499    /// already taken off the air waits in the node if the call is dropped meanwhile, so
500    /// nothing received is lost to a cancellation.
501    ///
502    /// # Returns
503    ///
504    /// The next message; the air never ends, so never `None`.
505    ///
506    /// # Errors
507    ///
508    /// Returns [`Error::Closed`] before [`connect`](Transport::connect), and
509    /// [`Error::Transport`] if the radio fails.
510    async fn recv(&mut self) -> Result<Option<Message>> {
511        if !self.connected {
512            return Err(Error::Closed);
513        }
514        loop {
515            if let Some(message) = self.pending.take() {
516                return Ok(Some(message));
517            }
518            let received = self
519                .radio
520                .take_frame(&mut self.buffer)
521                .map_err(radio_error)?;
522            match received {
523                Some(len) => self.accept(len).await?,
524                None => sleep(POLL).await,
525            }
526        }
527    }
528}
529
530fn radio_error<E: core::fmt::Debug>(error: E) -> Error {
531    Error::Transport(format!("radio: {error:?}"))
532}
533
534fn encode(topic: &str, payload: &[u8]) -> Result<Vec<u8>> {
535    let too_long = || {
536        Error::Transport(format!(
537            "a {} byte topic and a {} byte payload exceed the {MAX_MESSAGE} bytes a mesh frame carries",
538            topic.len(),
539            payload.len()
540        ))
541    };
542    if topic.len() + payload.len() > MAX_MESSAGE {
543        return Err(too_long());
544    }
545    let topic_len = u8::try_from(topic.len()).map_err(|_| too_long())?;
546    let mut body = Vec::with_capacity(1 + topic.len() + payload.len());
547    body.push(topic_len);
548    body.extend_from_slice(topic.as_bytes());
549    body.extend_from_slice(payload);
550    Ok(body)
551}
552
553fn decode(body: &[u8]) -> Option<(&str, &[u8])> {
554    let (&topic_len, rest) = body.split_first()?;
555    let topic_len = usize::from(topic_len);
556    if rest.len() < topic_len {
557        return None;
558    }
559    let (topic, payload) = rest.split_at(topic_len);
560    Some((core::str::from_utf8(topic).ok()?, payload))
561}
562
563#[cfg(test)]
564mod tests {
565    use super::*;
566    use std::collections::VecDeque;
567    use std::convert::Infallible;
568
569    fn link() -> LinkSettings {
570        LinkSettings::new(7, 125_000)
571    }
572
573    #[derive(Default)]
574    struct Air {
575        sent: Vec<Vec<u8>>,
576        heard: VecDeque<Vec<u8>>,
577        listens: usize,
578    }
579
580    impl LoraRadio for Air {
581        type Error = Infallible;
582
583        fn link(&self) -> Option<LinkSettings> {
584            Some(link())
585        }
586
587        fn start_transmit(&mut self, frame: &[u8]) -> std::result::Result<u64, Infallible> {
588            self.sent.push(frame.to_vec());
589            Ok(link().airtime_us(frame.len()))
590        }
591
592        fn finish_transmit(&mut self) -> std::result::Result<bool, Infallible> {
593            Ok(true)
594        }
595
596        fn listen(&mut self) -> std::result::Result<(), Infallible> {
597            self.listens += 1;
598            Ok(())
599        }
600
601        fn take_frame(
602            &mut self,
603            buffer: &mut [u8],
604        ) -> std::result::Result<Option<usize>, Infallible> {
605            Ok(self.heard.pop_front().map(|frame| {
606                buffer[..frame.len()].copy_from_slice(&frame);
607                frame.len()
608            }))
609        }
610    }
611
612    fn on_air(src: u32, id: u16, hops: u8, topic: &str, payload: &[u8]) -> Vec<u8> {
613        let body = encode(topic, payload).unwrap();
614        Frame::broadcast(src, id, &body)
615            .unwrap()
616            .with_hop_limit(hops)
617            .as_bytes()
618            .to_vec()
619    }
620
621    #[tokio::test(start_paused = true)]
622    async fn nothing_goes_out_or_comes_in_before_connect() {
623        let mut node = MeshRadio::new(Air::default(), 0x0A, 1000);
624        assert!(matches!(node.send("a", b"1").await, Err(Error::Closed)));
625        assert!(matches!(node.subscribe("a").await, Err(Error::Closed)));
626        assert!(matches!(node.recv().await, Err(Error::Closed)));
627        assert!(!node.is_connected());
628    }
629
630    #[tokio::test(start_paused = true)]
631    async fn a_message_leaves_as_a_broadcast_frame_carrying_its_topic() {
632        let mut node = MeshRadio::new(Air::default(), 0x0A, 1000);
633        node.connect().await.unwrap();
634        node.send("garden/moisture", b"28.5").await.unwrap();
635
636        let air = node.release();
637        assert_eq!(air.listens, 2);
638        let sent = Frame::parse(&air.sent[0]).unwrap();
639        assert_eq!(sent.src(), 0x0A);
640        assert!(sent.is_broadcast());
641        assert_eq!(sent.id(), 0);
642        assert_eq!(sent.hop_limit(), Frame::DEFAULT_HOP_LIMIT);
643        let mut body = vec![15];
644        body.extend_from_slice(b"garden/moisture28.5");
645        assert_eq!(sent.payload(), body.as_slice());
646    }
647
648    #[tokio::test(start_paused = true)]
649    async fn the_duty_cycle_holds_back_the_next_message_for_its_off_time() {
650        let mut node = MeshRadio::new(Air::default(), 0x0A, 10);
651        node.connect().await.unwrap();
652        node.send("t", b"1").await.unwrap();
653
654        let refused = node.send("t", b"2").await;
655        assert!(
656            matches!(&refused, Err(Error::Transport(reason)) if reason.contains("duty cycle")),
657            "{refused:?}"
658        );
659        let len = node.radio().sent[0].len();
660        sleep(Duration::from_micros(link().min_off_time_us(len, 10))).await;
661        node.send("t", b"2").await.unwrap();
662        assert_eq!(node.radio().sent.len(), 2);
663    }
664
665    #[tokio::test(start_paused = true)]
666    async fn recv_delivers_each_matching_topic_once_and_ignores_its_own_frames() {
667        let mut air = Air::default();
668        air.heard.extend([
669            on_air(0x0B, 7, 0, "kitchen/light", b"on"),
670            on_air(0x0B, 8, 0, "garden/moisture", b"28.5"),
671            on_air(0x0B, 8, 0, "garden/moisture", b"28.5"),
672            on_air(0x0A, 1, 0, "garden/valve", b"open"),
673            on_air(0x0C, 1, 0, "garden/valve", b"shut"),
674        ]);
675        let mut node = MeshRadio::new(air, 0x0A, 1000);
676        node.connect().await.unwrap();
677        node.subscribe("garden/+").await.unwrap();
678
679        let first = node.recv().await.unwrap().unwrap();
680        assert_eq!(first, Message::new("garden/moisture", b"28.5"));
681        let second = node.recv().await.unwrap().unwrap();
682        assert_eq!(second, Message::new("garden/valve", b"shut"));
683        assert!(node.radio().heard.is_empty());
684        assert!(node.radio().sent.is_empty());
685    }
686
687    #[tokio::test(start_paused = true)]
688    async fn a_frame_with_hops_left_is_relayed_with_one_hop_spent() {
689        let heard = on_air(0x0B, 3, 2, "alerts/flood", b"high");
690        let mut air = Air::default();
691        air.heard.push_back(heard.clone());
692        let mut node = MeshRadio::new(air, 0x0A, 1000);
693        node.connect().await.unwrap();
694        node.subscribe("alerts/#").await.unwrap();
695
696        let alert = node.recv().await.unwrap().unwrap();
697        assert_eq!(alert, Message::new("alerts/flood", b"high"));
698        let relayed = Frame::parse(&node.radio().sent[0]).unwrap();
699        let original = Frame::parse(&heard).unwrap();
700        assert_eq!(relayed.src(), 0x0B);
701        assert_eq!(relayed.id(), 3);
702        assert_eq!(relayed.hop_limit(), 1);
703        assert_eq!(relayed.payload(), original.payload());
704    }
705
706    #[tokio::test(start_paused = true)]
707    async fn a_node_that_does_not_relay_only_listens() {
708        let mut air = Air::default();
709        air.heard
710            .push_back(on_air(0x0B, 3, 2, "alerts/flood", b"high"));
711        let mut node = MeshRadio::new(air, 0x0A, 1000).without_relaying();
712        node.connect().await.unwrap();
713        node.subscribe("#").await.unwrap();
714
715        assert!(node.recv().await.unwrap().is_some());
716        assert!(node.radio().sent.is_empty());
717    }
718
719    #[tokio::test(start_paused = true)]
720    async fn a_message_too_long_for_one_frame_is_refused() {
721        let mut node = MeshRadio::new(Air::default(), 0x0A, 1000);
722        node.connect().await.unwrap();
723        let most = [0u8; MAX_MESSAGE];
724        assert!(matches!(
725            node.send("t", &most).await,
726            Err(Error::Transport(_))
727        ));
728        node.send("", &most).await.unwrap();
729        assert_eq!(node.radio().sent[0].len(), Frame::MAX_LEN);
730    }
731
732    #[test]
733    fn a_body_that_claims_a_longer_topic_than_it_holds_is_not_a_message() {
734        assert_eq!(decode(&[5, b'a', b'b']), None);
735        assert_eq!(decode(&[]), None);
736        assert_eq!(decode(&[1, 0xFF, b'x']), None);
737        assert_eq!(decode(&[1, b't', b'x']), Some(("t", &b"x"[..])));
738    }
739}
pamoja_radios/mesh.rs

pamoja_radios/
mesh.rs
