Abort wake task if user pokes the bulb

backend/src/tasks/lights.rs

@@ -29,6 +29,10 @@ pub async fn lights_task(state: &State) {
         .await
         .expect("Failed to open lights config");
 
+    let (cmd, bulb_states) = BulbManager::launch(config.bulbs.clone(), config.mqtt.clone())
+        .await
+        .expect("Failed to launch bulb manager");
+
     let mut wake_tasks: HashMap<(BulbId, Weekday), JoinHandle<()>> = lights_state
         .get()
         .wake_schedule
@@ -36,7 +40,8 @@ pub async fn lights_task(state: &State) {
         .flat_map(|(bulb, schedule)| schedule.iter().map(move |(day, time)| (bulb, day, time)))
         .map(|(bulb, day, time)| {
             let handle = spawn(wake_task(
-                state.client_message.clone(),
+                state.client_message.subscribe(),
+                cmd.clone(),
                 bulb.clone(),
                 *day,
                 *time,
@@ -46,10 +51,6 @@ pub async fn lights_task(state: &State) {
         })
         .collect();
 
-    let (cmd, bulb_states) = BulbManager::launch(config.bulbs.clone(), config.mqtt.clone())
-        .await
-        .expect("Failed to launch bulb manager");
-
     loop {
         let notify = bulb_states.notify_on_change();
         sleep(tokio::time::Duration::from_millis(1000 / 10)).await; // limit to 10 updates/second
@@ -118,7 +119,8 @@ pub async fn lights_task(state: &State) {
 }
 
 async fn wake_task(
-    channel: broadcast::Sender<ClientRequest>,
+    mut client_messages: broadcast::Receiver<ClientRequest>,
+    cmd: mpsc::Sender<BulbCommand>,
     id: BulbId,
     day: Weekday,
     time: NaiveTime,
@@ -145,23 +147,49 @@ async fn wake_task(
         sleep((alarm - Local::now()).to_std().unwrap()).await;
         alarm += chrono::Duration::weeks(1);
 
+        // slowly turn up brightness of bulb
         for brightness in (1..=50).map(|i| (i as f32) * 0.01) {
-            sleep(Duration::from_secs(12)).await;
+            select! {
+                // abort if the client pokes the bulb
+                _ = wait_for_bulb_command(&id, &mut client_messages) => break,
+                _ = sleep(Duration::from_secs(12)) => {}
+            };
 
-            let message = ClientMessage::SetBulbColor {
-                id: id.clone(),
-                color: BulbColor::Kelvin {
+            if cmd
+                .send(BulbCommand::SetColor(
+                    BulbSelector::Id(id.clone()),
+                    BulbColor::Kelvin {
                         t: 0.0,
                         b: brightness,
                     },
-            };
-
-            let (response, _) = mpsc::channel(1);
-            let request = ClientRequest { message, response };
-
-            if channel.send(request).is_err() {
+                ))
+                .await
+                .is_err()
+            {
                 return;
             };
         }
     }
 }
+
+/// Wait until we receive a client request that mutates the given bulb
+async fn wait_for_bulb_command(
+    bulb_id: &BulbId,
+    client_messages: &mut broadcast::Receiver<ClientRequest>,
+) {
+    loop {
+        match client_messages.recv().await {
+            Err(_) => return,
+            Ok(request) => match request.message {
+                ClientMessage::SetBulbColor { id, .. }
+                | ClientMessage::SetBulbPower { id, .. }
+                | ClientMessage::SetBulbWakeTime { id, .. }
+                    if &id == bulb_id =>
+                {
+                    break
+                }
+                _ => continue,
+            },
+        }
+    }
+}