ebus_rumqttc/
device.rs

1//! The device role, alone or with a controller, on a bundled connection.
2
3use std::fmt;
4use std::time::Duration;
5
6use ebus_homie_tokio::device::{
7    AttachError, BuildError, DeviceDriver, DeviceHandle, DeviceOptions, PairDriver, TreeBuilder,
8    attach_device, attach_pair,
9};
10use ebus_homie_tokio::ebus_homie::controller::Config;
11use ebus_homie_tokio::ebus_mqtt::LastWill;
12use ebus_homie_tokio::{ControllerHandle, EventReceiver};
13use rumqttc::{AsyncClient, MqttOptions};
14
15use crate::Error;
16use crate::conn::{self, Conn, DeviceConf, REQUEST_CAPACITY, Redirect, Reresolve};
17
18/// mosquitto's default receive quota: it discards QoS 2 messages past 20
19/// in flight from an MQTT 3.1.1 client (client.py:549-552).
20const MAX_INFLIGHT: u16 = 20;
21
22/// Room for a publish's fixed header, topic and packet id beyond its payload.
23const PACKET_OVERHEAD: usize = 1024;
24
25/// A device tree with its own connection.
26#[derive(Debug)]
27#[non_exhaustive]
28pub struct RunningDevice {
29    /// Presence, alerts, reconfiguration and shutdown.
30    pub handle: DeviceHandle,
31    /// The future the host spawns.
32    pub task: DeviceTask,
33}
34
35/// A device tree and a controller sharing one connection.
36#[derive(Debug)]
37#[non_exhaustive]
38pub struct RunningPair {
39    /// The device's handle.
40    pub device: DeviceHandle,
41    /// The controller's handle.
42    pub controller: ControllerHandle,
43    /// The controller's events.
44    pub events: EventReceiver,
45    /// The future the host spawns.
46    pub task: PairTask,
47}
48
49/// The device driver, the rumqttc event loop and the command executor, as
50/// one future.
51pub struct DeviceTask {
52    driver: DeviceDriver,
53    conn: Conn,
54    /// What [`device`] set on the options, for a [`Reresolve`] hook.
55    device: DeviceConf,
56}
57
58impl fmt::Debug for DeviceTask {
59    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
60        f.debug_struct("DeviceTask")
61            .field("driver", &self.driver)
62            .field("broker", &self.conn.eventloop.mqtt_options.broker_address())
63            .finish_non_exhaustive()
64    }
65}
66
67/// The pair driver, the rumqttc event loop and the command executor, as one
68/// future.
69pub struct PairTask {
70    driver: PairDriver,
71    conn: Conn,
72    /// What [`pair`] set on the options, for a [`Reresolve`] hook.
73    device: DeviceConf,
74}
75
76impl fmt::Debug for PairTask {
77    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
78        f.debug_struct("PairTask")
79            .field("driver", &self.driver)
80            .field("broker", &self.conn.eventloop.mqtt_options.broker_address())
81            .finish_non_exhaustive()
82    }
83}
84
85impl DeviceTask {
86    /// Looks the broker up again while its address may have changed, as
87    /// [`ControllerTask::reresolve`](crate::ControllerTask::reresolve) does.
88    /// Options `f` returns get the will, clean session, inflight cap and
89    /// outgoing packet limit [`device`] sets.
90    pub fn reresolve(mut self, f: Reresolve) -> Self {
91        self.conn.redirect = Some(Redirect::new(f, Some(self.device.clone())));
92        self
93    }
94
95    /// Runs until the device stops; see [`DeviceDriver::run`]. After
96    /// `Shutdown::Announce` or `Shutdown::Remove`, the connection waits until
97    /// every publish handed to rumqttc was written and acknowledged, at most
98    /// [`DeviceOptions::flush_timeout_ms`], then sends DISCONNECT. After
99    /// `Shutdown::Silent` or [`DeviceHandle::abort`] it drops the connection
100    /// without DISCONNECT, so the broker publishes the will. The future is
101    /// `Send`.
102    pub async fn run(self) -> Result<(), Error> {
103        conn::run(self.driver.run(), self.conn)
104            .await
105            .map_err(Error::Driver)
106    }
107}
108
109impl PairTask {
110    /// Looks the broker up again while its address may have changed, as
111    /// [`DeviceTask::reresolve`] does.
112    pub fn reresolve(mut self, f: Reresolve) -> Self {
113        self.conn.redirect = Some(Redirect::new(f, Some(self.device.clone())));
114        self
115    }
116
117    /// Runs until the device stops, as [`DeviceTask::run`] does. The future
118    /// is `Send`.
119    pub async fn run(self) -> Result<(), Error> {
120        conn::run(self.driver.run(), self.conn)
121            .await
122            .map_err(Error::Driver)
123    }
124}
125
126/// Sets what a device's connection needs on the caller's options: `will`, a
127/// clean session, at most 20 messages in flight, and an outgoing packet
128/// limit that fits the largest `$description`.
129pub(crate) fn configure(mqtt: &mut MqttOptions, will: &LastWill, max_description: usize) {
130    mqtt.set_last_will(rumqttc::LastWill::new(
131        will.topic.clone(),
132        will.payload.clone(),
133        conn::qos(will.qos),
134        will.retain,
135    ));
136    mqtt.set_clean_session(true);
137    let inflight = mqtt.inflight().min(MAX_INFLIGHT);
138    mqtt.set_inflight(inflight);
139    let incoming = mqtt.max_packet_size();
140    mqtt.set_max_packet_size(incoming, max_description.saturating_add(PACKET_OVERHEAD));
141}
142
143fn options(tree: &mut TreeBuilder, mut opts: DeviceOptions) -> DeviceOptions {
144    tree.set_receipts(true);
145    opts.link_up_discards_queued = true;
146    opts
147}
148
149/// A device tree that connects with `mqtt`. The broker coordinates, the
150/// client id (by convention the root's id, homie.py:3379), credentials and
151/// TLS are the caller's; the root's will, a clean session, the inflight cap
152/// and the outgoing packet limit are set here, along with
153/// [`TreeConfig::receipts`](ebus_homie_tokio::device::TreeConfig::receipts)
154/// and [`DeviceOptions::link_up_discards_queued`]. Nothing runs until the
155/// host spawns [`DeviceTask::run`].
156pub fn device(
157    mut mqtt: MqttOptions,
158    mut tree: TreeBuilder,
159    opts: DeviceOptions,
160) -> Result<RunningDevice, BuildError> {
161    let opts = options(&mut tree, opts);
162    let max = tree.config().max_description_bytes;
163    let a = attach_device(tree, opts)?;
164    configure(&mut mqtt, &a.last_will, max);
165    let (client, eventloop) = AsyncClient::new(mqtt, REQUEST_CAPACITY);
166    Ok(RunningDevice {
167        handle: a.handle,
168        task: DeviceTask {
169            driver: a.driver,
170            conn: Conn {
171                inbox: a.inbox,
172                commands: a.commands,
173                client,
174                eventloop,
175                receipts: true,
176                flush_timeout: Duration::from_millis(opts.flush_timeout_ms),
177                redirect: None,
178            },
179            device: DeviceConf {
180                will: a.last_will,
181                max_description: max,
182            },
183        },
184    })
185}
186
187/// A device tree and a controller for `config` on one connection, configured
188/// as [`device`] configures it. Nothing runs until the host spawns
189/// [`PairTask::run`].
190pub fn pair(
191    mut mqtt: MqttOptions,
192    mut tree: TreeBuilder,
193    config: Config,
194    opts: DeviceOptions,
195) -> Result<RunningPair, AttachError> {
196    let opts = options(&mut tree, opts);
197    let max = tree.config().max_description_bytes;
198    let a = attach_pair(tree, config, opts)?;
199    configure(&mut mqtt, &a.last_will, max);
200    let (client, eventloop) = AsyncClient::new(mqtt, REQUEST_CAPACITY);
201    Ok(RunningPair {
202        device: a.device,
203        controller: a.controller,
204        events: a.events,
205        task: PairTask {
206            driver: a.driver,
207            conn: Conn {
208                inbox: a.inbox,
209                commands: a.commands,
210                client,
211                eventloop,
212                receipts: true,
213                flush_timeout: Duration::from_millis(opts.flush_timeout_ms),
214                redirect: None,
215            },
216            device: DeviceConf {
217                will: a.last_will,
218                max_description: max,
219            },
220        },
221    })
222}
223
224#[cfg(test)]
225mod tests {
226    use super::*;
227
228    use ebus_homie_tokio::device::{DeviceSpec, TreeConfig};
229
230    fn send_check<T: Send>(_: &T) {}
231
232    fn tree() -> TreeBuilder {
233        TreeBuilder::new(TreeConfig::default(), "r", DeviceSpec::default()).expect("tree")
234    }
235
236    #[test]
237    fn device_connections_carry_the_will_and_the_limits() {
238        let mut mqtt = MqttOptions::new("r", "localhost", 1883);
239        mqtt.set_clean_session(false);
240        let will = LastWill::new(
241            "ebus/5/r/$state",
242            b"lost".to_vec(),
243            ebus_homie_tokio::ebus_mqtt::QoS::ExactlyOnce,
244            true,
245        );
246        configure(&mut mqtt, &will, 10_000);
247        let w = mqtt.last_will().expect("will");
248        assert_eq!(w.topic, "ebus/5/r/$state");
249        assert_eq!(&w.message[..], b"lost");
250        assert_eq!(w.qos, rumqttc::QoS::ExactlyOnce);
251        assert!(w.retain);
252        assert!(mqtt.clean_session());
253        assert_eq!(mqtt.inflight(), 20);
254        mqtt.set_inflight(5);
255        configure(&mut mqtt, &will, 10_000);
256        assert_eq!(mqtt.inflight(), 5, "a lower cap is kept");
257    }
258
259    /// True when a PUBLISH with a `payload`-byte body reaches a broker over
260    /// options set up as a device's with a `max_description` of `payload`,
261    /// or left as built when `device` is false.
262    async fn publish_reaches_broker(device: bool, payload: usize) -> bool {
263        use tokio::io::{AsyncReadExt, AsyncWriteExt};
264        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
265            .await
266            .expect("bind");
267        let port = listener.local_addr().expect("address").port();
268        let broker = tokio::spawn(async move {
269            let (mut s, _) = listener.accept().await.expect("accept");
270            let mut buf = vec![0u8; 4096];
271            let _ = s.read(&mut buf).await;
272            s.write_all(&[0x20, 2, 0, 0]).await.expect("CONNACK");
273            let mut total = 0;
274            while total < payload {
275                match s.read(&mut buf).await {
276                    Ok(0) | Err(_) => return false,
277                    Ok(n) => total += n,
278                }
279            }
280            true
281        });
282        let mut mqtt = MqttOptions::new("r", "127.0.0.1", port);
283        if device {
284            let will = LastWill::new(
285                "ebus/5/r/$state",
286                b"lost".to_vec(),
287                ebus_homie_tokio::ebus_mqtt::QoS::AtLeastOnce,
288                true,
289            );
290            configure(&mut mqtt, &will, payload);
291        }
292        let (client, mut eventloop) = AsyncClient::new(mqtt, 4);
293        client
294            .publish("t", rumqttc::QoS::AtMostOnce, false, vec![b'x'; payload])
295            .await
296            .expect("queued");
297        let poller = tokio::spawn(async move { while eventloop.poll().await.is_ok() {} });
298        let reached = tokio::time::timeout(Duration::from_secs(5), broker)
299            .await
300            .is_ok_and(|r| r.expect("broker task"));
301        poller.abort();
302        reached
303    }
304
305    #[tokio::test]
306    async fn device_connections_send_a_description_of_the_configured_size() {
307        // Over rumqttc's default 10 KiB outgoing limit.
308        assert!(!publish_reaches_broker(false, 12_000).await);
309        assert!(publish_reaches_broker(true, 12_000).await);
310    }
311
312    #[test]
313    fn tasks_are_send() {
314        let mqtt = MqttOptions::new("r", "localhost", 1883);
315        let run = device(mqtt.clone(), tree(), DeviceOptions::default()).expect("device");
316        send_check(&run.task.run());
317        let run = device(mqtt.clone(), tree(), DeviceOptions::default()).expect("device");
318        send_check(
319            &run.task
320                .reresolve(Box::new(|| Box::pin(async { None })))
321                .run(),
322        );
323        let run = pair(mqtt, tree(), Config::wildcard(), DeviceOptions::default()).expect("pair");
324        send_check(
325            &run.task
326                .reresolve(Box::new(|| Box::pin(async { None })))
327                .run(),
328        );
329    }
330
331    #[tokio::test]
332    async fn abort_while_the_broker_is_down_ends_the_task() {
333        let port = std::net::TcpListener::bind("127.0.0.1:0")
334            .and_then(|l| l.local_addr())
335            .expect("free port")
336            .port();
337        let mqtt = MqttOptions::new("r", "127.0.0.1", port);
338        let run = device(mqtt, tree(), DeviceOptions::default()).expect("device");
339        let task = tokio::spawn(run.task.run());
340        tokio::time::sleep(Duration::from_millis(100)).await;
341        run.handle.abort();
342        let ended = tokio::time::timeout(Duration::from_secs(10), task).await;
343        assert!(matches!(ended, Ok(Ok(Ok(())))), "{ended:?}");
344    }
345
346    #[tokio::test]
347    async fn shutdown_while_the_broker_is_down_ends_the_task() {
348        let port = std::net::TcpListener::bind("127.0.0.1:0")
349            .and_then(|l| l.local_addr())
350            .expect("free port")
351            .port();
352        let mqtt = MqttOptions::new("r", "127.0.0.1", port);
353        let run = device(mqtt, tree(), DeviceOptions::default()).expect("device");
354        let task = tokio::spawn(run.task.run());
355        tokio::time::sleep(Duration::from_millis(100)).await;
356        run.handle
357            .shutdown(ebus_homie_tokio::device::Shutdown::Announce)
358            .await;
359        let ended = tokio::time::timeout(Duration::from_secs(10), task).await;
360        assert!(matches!(ended, Ok(Ok(Ok(())))), "{ended:?}");
361    }
362}
ebus_rumqttc/device.rs

ebus_rumqttc/
device.rs
