bambu_rs/server/
live.rs

1//! The live printer source: bridges the blocking LAN MQTT [`monitor`] loop into
2//! a [`watch`] channel so the dashboard can stream the *real* device status the
3//! same way it streams the fake one.
4//!
5//! [`monitor`]: crate::client::LanMqttClient::monitor
6
7use std::thread;
8use std::time::Duration;
9
10use tokio::sync::watch;
11
12use super::api::PrinterSource;
13use crate::client::{LanMqttClient, WatchStep};
14use crate::config::ResolvedTarget;
15use crate::core::status::PrinterStatus;
16
17/// Wait before reconnecting after `monitor` returns (a stall or connect error).
18const RECONNECT_DELAY: Duration = Duration::from_secs(3);
19/// Stall window handed to `monitor`: it returns if no report arrives within this,
20/// then our outer loop reconnects. Generous so a momentarily-quiet printer isn't
21/// dropped mid-print.
22const STALL: Duration = Duration::from_secs(120);
23
24/// A [`PrinterSource`] backed by a live LAN MQTT connection.
25///
26/// A dedicated OS thread runs the blocking [`LanMqttClient::monitor`] loop (which
27/// owns its own current-thread runtime and auto-reconnects internally), turning
28/// each merged report into a [`PrinterStatus`] pushed onto a [`watch`] channel.
29/// The thread reconnects forever until the source and all its subscribers drop.
30pub struct LiveSource {
31    tx: watch::Sender<PrinterStatus>,
32    // Held so the channel keeps a receiver while the source is alive; when the
33    // source (and any WS subscribers) drop, the worker's reconnect loop stops.
34    _keepalive: watch::Receiver<PrinterStatus>,
35}
36
37impl LiveSource {
38    /// Connect to `target` and start streaming live status. `interval` is the
39    /// `pushall` poll period; `None` relies on the printer's autonomous ~2 s push
40    /// (the gentlest option, the default for an always-on dashboard).
41    pub fn connect(target: ResolvedTarget, interval: Option<Duration>) -> Self {
42        Self::spawn(move |tx| {
43            loop {
44                let client = LanMqttClient::new(target.clone()).with_timeout(STALL);
45                let _ = client.monitor(interval, |rs| {
46                    // Stop promptly when the source + all subscribers are gone,
47                    // instead of running until the next stall/transport break.
48                    if tx.send(PrinterStatus::from_state(rs.get())).is_err() {
49                        WatchStep::Stop
50                    } else {
51                        WatchStep::Continue
52                    }
53                });
54                if tx.receiver_count() == 0 {
55                    break; // the source and all subscribers are gone
56                }
57                thread::sleep(RECONNECT_DELAY);
58            }
59        })
60    }
61
62    /// Spawn the bridge thread around `run`, handing it the channel sender. The
63    /// real MQTT loop lives in [`LiveSource::connect`]; this seam lets tests drive
64    /// the channel without a printer.
65    fn spawn<R>(run: R) -> Self
66    where
67        R: FnOnce(watch::Sender<PrinterStatus>) + Send + 'static,
68    {
69        let (tx, rx) = watch::channel(PrinterStatus::default());
70        let worker = tx.clone();
71        thread::spawn(move || run(worker));
72        Self { tx, _keepalive: rx }
73    }
74}
75
76impl PrinterSource for LiveSource {
77    fn current(&self) -> PrinterStatus {
78        self.tx.borrow().clone()
79    }
80    fn subscribe(&self) -> watch::Receiver<PrinterStatus> {
81        self.tx.subscribe()
82    }
83}
84
85#[cfg(test)]
86mod tests {
87    use super::*;
88
89    #[test]
90    fn bridges_published_status_to_current_and_subscribers() {
91        let src = LiveSource::spawn(|tx| {
92            let _ = tx.send(PrinterStatus {
93                gcode_state: Some("RUNNING".to_string()),
94                mc_percent: Some(42),
95                ..Default::default()
96            });
97        });
98        // The worker sends near-instantly; poll up to a generous deadline so the
99        // test is reliable without timing assumptions.
100        let mut got = None;
101        for _ in 0..200 {
102            let s = src.current();
103            if s.gcode_state.is_some() {
104                got = Some(s);
105                break;
106            }
107            thread::sleep(Duration::from_millis(10));
108        }
109        let s = got.expect("worker thread should publish a status");
110        assert_eq!(s.gcode_state.as_deref(), Some("RUNNING"));
111        assert_eq!(s.mc_percent, Some(42));
112    }
113}
bambu_rs/server/live.rs

bambu_rs/server/
live.rs
