Use broadcast for mqtt message so we can have a queue

src/devices.rs

@@ -95,40 +95,25 @@ pub fn start(mut mqtt_rx: mqtt::Receiver, mut presence_rx: presence::Receiver, m
     let (tx, mut rx) = mpsc::channel(100);
 
     tokio::spawn(async move {
+        // @TODO Handle error better
         loop {
             tokio::select! {
-                res = mqtt_rx.changed() => {
+                Ok(message) = mqtt_rx.recv() => {
-                    if !res.is_ok() {
+                    devices.on_mqtt(&message).await;
-                        break;
-                    }
-
-                    // @TODO Not ideal that we have to clone here, but not sure how to work around that
-                    let message = mqtt_rx.borrow().clone();
-                    if let Some(message) = message {
-                        devices.on_mqtt(&message).await;
-                    }
                 }
-                res = presence_rx.changed() => {
+                Ok(_) = presence_rx.changed() => {
-                    if !res.is_ok() {
-                        break;
-                    }
-
                     let presence = *presence_rx.borrow();
                     devices.on_presence(presence).await;
                 }
-                res = light_sensor_rx.changed() => {
+                Ok(_) = light_sensor_rx.changed() => {
-                    if !res.is_ok() {
-                        break;
-                    }
-
                     let darkness = *light_sensor_rx.borrow();
                     devices.on_darkness(darkness).await;
                 }
+                // @TODO Handle receiving None better, otherwise it might constantly run doing
+                // nothing
                 Some(cmd) = rx.recv() => devices.handle_cmd(cmd)
             }
         }
-
-        unreachable!("Did not expect this");
     });
 
     return DeviceHandle { tx };

src/light_sensor.rs

@@ -28,14 +28,12 @@ pub async fn start(mut mqtt_rx: mqtt::Receiver, config: LightSensorConfig, clien
     let mut light_sensor = LightSensor { is_dark: is_dark.clone(), mqtt: config.mqtt, min: config.min, max: config.max, tx };
 
     tokio::spawn(async move {
-        while mqtt_rx.changed().await.is_ok() {
+        loop {
-            let message = mqtt_rx.borrow().clone();
+            // @TODO Handle errors, warn if lagging
-            if let Some(message) = message {
+            if let Ok(message) = mqtt_rx.recv().await {
                 light_sensor.on_mqtt(&message).await;
             }
         }
-
-        unreachable!("Did not expect this");
     });
 
     return is_dark;

src/main.rs

@@ -8,7 +8,7 @@ use automation::{
     config::{Config, OpenIDConfig},
     devices,
     hue_bridge::HueBridge,
-    light_sensor, mqtt::{self, Mqtt},
+    light_sensor, mqtt::Mqtt,
     ntfy::Ntfy,
     presence,
 };

src/mqtt.rs

@@ -3,15 +3,15 @@ use serde::{Serialize, Deserialize};
 use tracing::{error, debug};
 
 use rumqttc::{Publish, Event, Incoming, EventLoop};
-use tokio::sync::watch;
+use tokio::sync::broadcast;
 
 #[async_trait]
 pub trait OnMqtt {
     async fn on_mqtt(&mut self, message: &Publish);
 }
 
-pub type Receiver = watch::Receiver<Option<Publish>>;
+pub type Receiver = broadcast::Receiver<Publish>;
-type Sender = watch::Sender<Option<Publish>>;
+type Sender = broadcast::Sender<Publish>;
 
 pub struct Mqtt {
     tx: Sender,
@@ -20,7 +20,7 @@ pub struct Mqtt {
 
 impl Mqtt {
     pub fn new(eventloop: EventLoop) -> Self {
-        let (tx, _rx) = watch::channel(None);
+        let (tx, _rx) = broadcast::channel(100);
         Self { tx, eventloop }
     }
 
@@ -35,7 +35,7 @@ impl Mqtt {
                 let notification = self.eventloop.poll().await;
                 match notification {
                     Ok(Event::Incoming(Incoming::Publish(p))) => {
-                        self.tx.send(Some(p)).ok();
+                        self.tx.send(p).ok();
                     },
                     Ok(..) => continue,
                     Err(err) => {

src/presence.rs

@@ -30,15 +30,12 @@ pub async fn start(mut mqtt_rx: mqtt::Receiver, mqtt: MqttDeviceConfig, client:
     let mut presence = Presence { devices: HashMap::new(), overall_presence: overall_presence.clone(), mqtt, tx };
 
     tokio::spawn(async move {
-        while mqtt_rx.changed().await.is_ok() {
+        loop {
-            // @TODO Not ideal that we have to clone here, but not sure how to work around that
+            // @TODO Handle errors, warn if lagging
-            let message = mqtt_rx.borrow().clone();
+            if let Ok(message) = mqtt_rx.recv().await {
-            if let Some(message) = message {
                 presence.on_mqtt(&message).await;
             }
         }
-
-        unreachable!("Did not expect this");
     });
 
     return overall_presence;