ebus_mqtt/
lib.rs

1//! MQTT vocabulary for eBus.
2//!
3//! The messages and commands the Homie state machines exchange with an MQTT
4//! client, topic filter matching, and, with `alloc`, the `FilterLedger` that
5//! tracks which subscriptions the client holds and who may remove them, the
6//! [`Role`] trait those state machines implement, and [`Pair`], which runs two
7//! of them on one link.
8//!
9//! The crate performs no I/O and reads no clock: time arrives as an [`Instant`]
10//! the caller supplies, received messages arrive as [`Message`]s, and work for
11//! the MQTT client leaves as [`Command`]s.
12//!
13//! ```
14//! use ebus_mqtt::filter;
15//!
16//! assert!(filter::matches("ebus/5/+/$state", "ebus/5/meter-1/$state"));
17//! assert!(filter::matches("ebus/5/meter-1/#", "ebus/5/meter-1/power/watts"));
18//! // A leading wildcard does not match a topic that starts with `$`.
19//! assert!(!filter::matches("+/5/meter-1/$state", "$SYS/5/meter-1/$state"));
20//! ```
21//!
22//! # Features
23//!
24//! - `std` (default): links `std`; implies `alloc`.
25//! - `alloc`: [`Command`], [`LastWill`], [`FilterLedger`], [`Role`] and [`Pair`].
26
27#![cfg_attr(not(feature = "std"), no_std)]
28#![cfg_attr(docsrs, feature(doc_cfg))]
29
30#[cfg(feature = "alloc")]
31extern crate alloc;
32
33#[cfg(feature = "alloc")]
34use alloc::{string::String, vec::Vec};
35
36pub mod filter;
37#[cfg(feature = "alloc")]
38mod ledger;
39#[cfg(feature = "alloc")]
40mod role;
41
42#[cfg(feature = "alloc")]
43pub use ledger::FilterLedger;
44#[cfg(feature = "alloc")]
45pub use role::{Pair, Receipt, Role};
46
47/// Whether a role took a received message.
48#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
49pub enum Handled {
50    /// The message addressed something the role tracks.
51    Consumed,
52    /// Not for the role.
53    Ignored,
54}
55
56/// MQTT quality of service level.
57///
58/// The discriminants are the wire values, so `qos as u8` is the QoS field of a
59/// packet.
60#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
61#[repr(u8)]
62pub enum QoS {
63    /// QoS 0: delivered at most once, with no acknowledgment.
64    AtMostOnce = 0,
65    /// QoS 1: delivered at least once.
66    AtLeastOnce = 1,
67    /// QoS 2: delivered exactly once.
68    ExactlyOnce = 2,
69}
70
71impl QoS {
72    /// The QoS level for a wire value, or `None` for a value above 2.
73    pub const fn from_u8(value: u8) -> Option<Self> {
74        match value {
75            0 => Some(Self::AtMostOnce),
76            1 => Some(Self::AtLeastOnce),
77            2 => Some(Self::ExactlyOnce),
78            _ => None,
79        }
80    }
81}
82
83/// Milliseconds on the caller's monotonic clock.
84///
85/// The origin is the caller's choice; only differences between two instants
86/// from the same clock carry meaning.
87#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
88pub struct Instant(pub u64);
89
90impl Instant {
91    /// Milliseconds elapsed from `earlier` to `self`, or 0 when `earlier` is later.
92    pub const fn saturating_since(self, earlier: Instant) -> u64 {
93        self.0.saturating_sub(earlier.0)
94    }
95
96    /// The instant `ms` milliseconds after `self`, saturating at `u64::MAX`.
97    #[must_use]
98    pub const fn add_ms(self, ms: u64) -> Instant {
99        Instant(self.0.saturating_add(ms))
100    }
101}
102
103/// A message received from the broker, borrowed from the client's buffer.
104///
105/// Built with [`Message::new`] outside this crate.
106#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
107#[non_exhaustive]
108pub struct Message<'a> {
109    /// Topic name the message was published to.
110    pub topic: &'a str,
111    /// Application payload, possibly empty.
112    pub payload: &'a [u8],
113    /// MQTT retain bit; `None` when the client does not surface it.
114    pub retain: Option<bool>,
115}
116
117impl<'a> Message<'a> {
118    /// A received message.
119    pub const fn new(topic: &'a str, payload: &'a [u8], retain: Option<bool>) -> Self {
120        Self {
121            topic,
122            payload,
123            retain,
124        }
125    }
126}
127
128/// A change in the state of the connection to the broker.
129///
130/// `#[non_exhaustive]`: a `match` on it outside this crate needs a `_` arm.
131#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
132#[non_exhaustive]
133pub enum Link {
134    /// The broker accepted a connection (CONNACK with return code 0).
135    ///
136    /// Built with [`Link::up`] outside this crate.
137    #[non_exhaustive]
138    Up {
139        /// The CONNACK session present flag: the broker resumed a stored
140        /// session, including its subscriptions.
141        session_present: bool,
142    },
143    /// The connection was lost or closed.
144    Down,
145}
146
147impl Link {
148    /// [`Link::Up`] with the CONNACK session present flag.
149    pub const fn up(session_present: bool) -> Self {
150        Self::Up { session_present }
151    }
152}
153
154/// Work for the MQTT client.
155///
156/// Each variant is `#[non_exhaustive]`: match it with `..`, and build it with
157/// [`Command::subscribe`], [`Command::unsubscribe`] or [`Command::publish`]
158/// outside this crate.
159#[cfg(feature = "alloc")]
160#[derive(Clone, Debug, PartialEq, Eq, Hash)]
161#[non_exhaustive]
162pub enum Command {
163    /// Subscribe to every filter in one SUBSCRIBE packet.
164    #[non_exhaustive]
165    Subscribe {
166        /// Topic filters with their requested maximum QoS.
167        filters: Vec<(String, QoS)>,
168    },
169    /// Unsubscribe from every filter in one UNSUBSCRIBE packet.
170    #[non_exhaustive]
171    Unsubscribe {
172        /// Topic filters to remove.
173        filters: Vec<String>,
174    },
175    /// Publish one message.
176    #[non_exhaustive]
177    Publish {
178        /// Topic name.
179        topic: String,
180        /// Application payload.
181        payload: Vec<u8>,
182        /// QoS to publish with.
183        qos: QoS,
184        /// MQTT retain flag.
185        retain: bool,
186    },
187}
188
189#[cfg(feature = "alloc")]
190impl Command {
191    /// Subscribe to every filter in one SUBSCRIBE packet.
192    pub fn subscribe(filters: impl Into<Vec<(String, QoS)>>) -> Self {
193        Self::Subscribe {
194            filters: filters.into(),
195        }
196    }
197
198    /// Unsubscribe from every filter in one UNSUBSCRIBE packet.
199    pub fn unsubscribe(filters: impl Into<Vec<String>>) -> Self {
200        Self::Unsubscribe {
201            filters: filters.into(),
202        }
203    }
204
205    /// Publish one message.
206    pub fn publish(
207        topic: impl Into<String>,
208        payload: impl Into<Vec<u8>>,
209        qos: QoS,
210        retain: bool,
211    ) -> Self {
212        Self::Publish {
213            topic: topic.into(),
214            payload: payload.into(),
215            qos,
216            retain,
217        }
218    }
219}
220
221/// The Last Will and Testament a client registers when it connects.
222///
223/// Built with [`LastWill::new`] outside this crate.
224#[cfg(feature = "alloc")]
225#[derive(Clone, Debug, PartialEq, Eq, Hash)]
226#[non_exhaustive]
227pub struct LastWill {
228    /// Topic the broker publishes the will to.
229    pub topic: String,
230    /// Will payload.
231    pub payload: Vec<u8>,
232    /// Will QoS.
233    pub qos: QoS,
234    /// Will retain flag.
235    pub retain: bool,
236}
237
238#[cfg(feature = "alloc")]
239impl LastWill {
240    /// A will the broker publishes to `topic` when the client drops.
241    pub fn new(
242        topic: impl Into<String>,
243        payload: impl Into<Vec<u8>>,
244        qos: QoS,
245        retain: bool,
246    ) -> Self {
247        Self {
248            topic: topic.into(),
249            payload: payload.into(),
250            qos,
251            retain,
252        }
253    }
254}
255
256#[cfg(test)]
257mod tests {
258    use super::*;
259
260    #[test]
261    fn qos_wire_values_round_trip() {
262        for q in [QoS::AtMostOnce, QoS::AtLeastOnce, QoS::ExactlyOnce] {
263            assert_eq!(QoS::from_u8(q as u8), Some(q));
264        }
265        assert_eq!(QoS::from_u8(3), None);
266        assert_eq!(QoS::from_u8(u8::MAX), None);
267        assert!(QoS::AtMostOnce < QoS::AtLeastOnce);
268        assert!(QoS::AtLeastOnce < QoS::ExactlyOnce);
269    }
270
271    #[test]
272    fn instant_saturates() {
273        let a = Instant(1_000);
274        assert_eq!(a.add_ms(500), Instant(1_500));
275        assert_eq!(a.add_ms(500).saturating_since(a), 500);
276        assert_eq!(a.saturating_since(a.add_ms(500)), 0);
277        assert_eq!(a.saturating_since(a), 0);
278        assert_eq!(Instant(u64::MAX - 1).add_ms(10), Instant(u64::MAX));
279        assert_eq!(Instant(u64::MAX).saturating_since(Instant(0)), u64::MAX);
280        assert_eq!(Instant::default(), Instant(0));
281    }
282
283    #[cfg(feature = "alloc")]
284    #[test]
285    fn constructors_fill_every_field() {
286        use alloc::{string::ToString, vec};
287        assert_eq!(
288            Command::publish("t", b"p".to_vec(), QoS::AtLeastOnce, true),
289            Command::Publish {
290                topic: "t".to_string(),
291                payload: b"p".to_vec(),
292                qos: QoS::AtLeastOnce,
293                retain: true,
294            }
295        );
296        let filters = vec![("a/#".to_string(), QoS::ExactlyOnce)];
297        assert_eq!(
298            Command::subscribe(filters.clone()),
299            Command::Subscribe { filters }
300        );
301        assert_eq!(
302            Command::unsubscribe(vec!["a/#".to_string()]),
303            Command::Unsubscribe {
304                filters: vec!["a/#".to_string()],
305            }
306        );
307        let will = LastWill::new("w", b"lost".to_vec(), QoS::AtMostOnce, false);
308        assert_eq!(
309            (
310                will.topic.as_str(),
311                will.payload.as_slice(),
312                will.qos,
313                will.retain
314            ),
315            ("w", &b"lost"[..], QoS::AtMostOnce, false)
316        );
317        assert_eq!(
318            Link::up(true),
319            Link::Up {
320                session_present: true
321            }
322        );
323        assert_eq!(
324            Message::new("t", b"p", Some(false)),
325            Message {
326                topic: "t",
327                payload: b"p",
328                retain: Some(false),
329            }
330        );
331    }
332
333    #[test]
334    fn message_is_copy() {
335        let m = Message {
336            topic: "ebus/5/d/$state",
337            payload: b"ready",
338            retain: None,
339        };
340        let n = m;
341        assert_eq!(m, n);
342    }
343}
344
345#[cfg(doctest)]
346#[doc = include_str!("../README.md")]
347struct ReadmeDoctests;
ebus_mqtt/lib.rs

ebus_mqtt/
lib.rs
