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

pamoja_radios/
mesh.rs
