diff --git a/config.yml.example b/config.yml.example index 0f799ed..407eddd 100644 --- a/config.yml.example +++ b/config.yml.example @@ -9,6 +9,8 @@ mqtt: rooms: - "!iFCnqQcvrVIeehrjQO:chat.labolyon.fr" +debounce_delay: 10 # seconds + # username / password login # /!\ will create a new device every time the bot starts username: "login" diff --git a/src/config.rs b/src/config.rs index bf9ce3e..e242757 100644 --- a/src/config.rs +++ b/src/config.rs @@ -46,6 +46,7 @@ pub enum AuthConfig { pub struct Config { pub mqtt: MQTTConfig, pub rooms: Vec, + pub debounce_delay: u64, #[serde(flatten)] pub auth: AuthConfig, pub homeserver_uri: String diff --git a/src/main.rs b/src/main.rs index bfe98b8..7162419 100644 --- a/src/main.rs +++ b/src/main.rs @@ -24,7 +24,8 @@ mod matrix; mod mqtt; use tokio::{ - sync::{Notify, broadcast}, + sync::{Notify, broadcast, mpsc}, + time::{sleep, Duration}, task::JoinHandle, }; @@ -48,11 +49,16 @@ struct Args { config: String, } -#[derive(PartialEq)] -enum LolOpenerStatus { +#[derive(PartialEq, Clone, Copy, Debug)] +enum DoorSensorStatus { Open, Closed, } +#[derive(Debug)] +struct LolOpenerState { + sensor_status: Option, + update_pending: bool, +} #[tokio::main] async fn main() -> anyhow::Result<()> { @@ -80,25 +86,54 @@ async fn main() -> anyhow::Result<()> { matrix_ready.notified().await; info!("Matrix is ready"); - let current_status = Arc::new(Mutex::new(None)); - let current_status = Arc::clone(¤t_status); + let current_state = Arc::new(Mutex::new(LolOpenerState { + sensor_status: None, + update_pending: false + })); + let current_state = Arc::clone(¤t_state); + + let (debounce_tx, mut debounce_rx) = mpsc::unbounded_channel::<(DoorSensorStatus, String)>(); + let debounce_delay = config.debounce_delay; + let rooms = config.rooms.clone(); + let handle = tokio::spawn(async move { + + while let Some((new_status, msg)) = debounce_rx.recv().await { + let should_wait = { + //XXX: Panic if the lock is poisoned + let state = current_state.lock().unwrap(); + !state.update_pending + }; + + if should_wait { + sleep(Duration::from_secs(debounce_delay)).await; + //XXX: Panic if the lock is poisoned + let mut state = current_state.lock().unwrap(); + state.update_pending = false; + + if Some(new_status) != state.sensor_status { + debug!("Sending msg {}", msg); + //XXX: Panic if the matrix module can't get the message + bcast_tx.send((msg, rooms.clone())).unwrap(); + state.sensor_status = Some(new_status); + } + } + } + }); let mut mqtt = MQTT::new(config.mqtt.clone()).await; mqtt.subscribe(move |status: StatusMessage| { let (new_status, msg) = match status { - StatusMessage { contact: true, .. } => (Some(LolOpenerStatus::Closed), "🔴 Le LOL est fermé :("), - StatusMessage { contact: false, .. } => (Some(LolOpenerStatus::Open), "🟢 Le LOL est ouvert :)"), + StatusMessage { contact: true, .. } => (DoorSensorStatus::Closed, "🔴 Le LOL est fermé :("), + StatusMessage { contact: false, .. } => (DoorSensorStatus::Open, "🟢 Le LOL est ouvert :)"), }; - //XXX: Panic if the lock is poisoned (why would that happen?) - let mut current_status = current_status.lock().unwrap(); - if new_status != *current_status { - debug!("Sending msg {}", msg); - //XXX: Panic if the matrix module can't get the message - bcast_tx.send((msg.into(), config.rooms.clone())).unwrap(); - *current_status = new_status; - } + //XXX: Panic if the channel is closed + debounce_tx.send((new_status, String::from(msg))).unwrap(); + }).await.unwrap(); //XXX: Panic if the mqtt broker fails + + handle.abort(); + handle.await.unwrap(); }) };