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..52f9a01 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, + pending: Option<(DoorSensorStatus, String)>, +} #[tokio::main] async fn main() -> anyhow::Result<()> { @@ -80,25 +86,75 @@ 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, + pending: None + })); + 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 mut state = current_state.lock().unwrap(); + let should_wait = state.pending.is_none(); + + state.pending = Some((new_status, msg)); + should_wait + }; + + if should_wait { + let tx = bcast_tx.clone(); + let rooms = rooms.clone(); + let state = current_state.clone(); + + tokio::spawn(async move { + sleep(Duration::from_secs(debounce_delay)).await; + + let msg = { + //XXX: Panic if the lock is poisoned + let mut state = state.lock().unwrap(); + let pending = state.pending.take(); + + match pending { + Some((status, msg)) + if state.sensor_status.as_ref() != Some(&status) => { + state.sensor_status = Some(status); + Some(msg) + } + + _ => None, + } + }; + + if let Some(msg) = msg { + debug!("Sending msg {}", msg); + //XXX: Panic if the matrix module can't get the message + tx.send((msg, rooms)).unwrap(); + } + }); + } + } + }); 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(); }) };