commit

src/coordinator/mqtt.rs

@@ -1,7 +1,8 @@
 use std::str::FromStr;
 use std::time::Duration;
 
-use rumqttc::{Event, Incoming, LastWill, MqttOptions, QoS};
+use log::{error, info};
+use rumqttc::{ConnectionError, Event, Incoming, LastWill, MqttOptions, QoS};
 use tokio::sync::broadcast;
 
 use deckster_shared::handler_communication::{HandlerCommand, HandlerEvent};
@@ -19,6 +20,8 @@ pub async fn start_mqtt_client(
     let topic_prefix = config.topic_prefix.to_owned();
     let config_topic = format!("{topic_prefix}/config");
 
+    let handler_hosts_config_json = serde_json::to_string(handler_hosts_config).unwrap().into_bytes();
+
     let mut options = MqttOptions::new(&config.client_id, &config.host, config.port);
     options.set_keep_alive(Duration::from_secs(3));
     options.set_clean_session(true);
@@ -36,10 +39,6 @@ pub async fn start_mqtt_client(
 
     let (client, mut event_loop) = rumqttc::AsyncClient::new(options, 32);
 
-    client.subscribe(format!("{topic_prefix}/keys/+/+/style"), QoS::ExactlyOnce).await.unwrap();
-    client.subscribe(format!("{topic_prefix}/knobs/+/+/style"), QoS::ExactlyOnce).await.unwrap();
-    client.subscribe(format!("{topic_prefix}/knobs/+/+/value"), QoS::ExactlyOnce).await.unwrap();
-
     tokio::spawn({
         let topic_prefix = topic_prefix.clone();
         let client = client.clone();
@@ -79,10 +78,41 @@ pub async fn start_mqtt_client(
 
     tokio::spawn({
         let topic_prefix = topic_prefix.clone();
+        let client = client.clone();
 
         async move {
-            while let Ok(event) = event_loop.poll().await {
+            let mut is_connected = false;
-                if let Event::Incoming(Incoming::Publish(event)) = event {
+            let mut is_first_try = true;
+
+            loop {
+                match event_loop.poll().await {
+                    Err(error) => {
+                        if is_first_try {
+                            error!("MQTT connection failed (retrying in the background): {error}");
+                            is_first_try = false;
+                        } else if is_connected {
+                            error!("MQTT connection lost: {error}");
+                        }
+
+                        is_connected = false;
+                        tokio::time::sleep(Duration::from_secs(2)).await;
+                    }
+                    Ok(Event::Incoming(event)) => match event {
+                        Incoming::ConnAck(_) => {
+                            info!("MQTT connection established.");
+                            is_connected = true;
+                            is_first_try = false;
+
+                            client.subscribe(format!("{topic_prefix}/keys/+/+/style"), QoS::ExactlyOnce).await.unwrap();
+                            client.subscribe(format!("{topic_prefix}/knobs/+/+/style"), QoS::ExactlyOnce).await.unwrap();
+                            client.subscribe(format!("{topic_prefix}/knobs/+/+/value"), QoS::ExactlyOnce).await.unwrap();
+
+                            client
+                                .publish(config_topic.clone(), QoS::ExactlyOnce, true, handler_hosts_config_json.clone())
+                                .await
+                                .unwrap()
+                        }
+                        Incoming::Publish(event) => {
                             let segments = event.topic.strip_prefix(&topic_prefix).unwrap().split('/').skip(1).collect::<Vec<&str>>();
                             let page_id = segments[1];
                             let position = segments[2];
@@ -143,12 +173,11 @@ pub async fn start_mqtt_client(
                                 _ => {}
                             };
                         }
+                        _ => {}
+                    },
+                    _ => {}
+                }
             }
         }
     });
-
-    client
-        .publish(config_topic, QoS::ExactlyOnce, true, serde_json::to_string(handler_hosts_config).unwrap())
-        .await
-        .unwrap()
 }

src/handler_host/mod.rs

@@ -1,7 +1,7 @@
 use std::path::Path;
 
 use color_eyre::Result;
-use log::{debug, info, warn};
+use log::{debug, info, trace, warn};
 use tokio::sync::{broadcast, mpsc};
 
 use deckster_shared::handler_communication::{HandlerCommand, HandlerEvent};
@@ -26,8 +26,11 @@ pub async fn start(config_directory: &Path, config: Config) -> Result<()> {
     while let Some(handler_hosts_config) = handler_hosts_config_receiver.recv().await {
         match handler_hosts_config {
             None => {
-                info!("Coordinator was stopped. Sending stop event to handlers.");
+                if is_running {
+                    info!("Stopping handlers…");
                     events_sender.send(HandlerEvent::Stop).unwrap();
+                }
+
                 is_running = false;
             }
             Some(handler_hosts_config) => {
@@ -39,7 +42,7 @@ pub async fn start(config_directory: &Path, config: Config) -> Result<()> {
                 is_running = true;
 
                 info!("Received new configuration. Starting handlers…");
-                debug!("New configuration: {handler_hosts_config:#?}");
+                trace!("New configuration: {handler_hosts_config:#?}");
 
                 handler_runner::start(
                     config.host_id.clone(),
@@ -52,7 +55,6 @@ pub async fn start(config_directory: &Path, config: Config) -> Result<()> {
             }
         }
     }
-    dbg!("hey");
 
     Ok(())
 }

src/handler_host/mqtt.rs

@@ -1,7 +1,7 @@
 use std::str::FromStr;
 use std::time::Duration;
 
-use log::error;
+use log::{error, info};
 use rumqttc::{Event, Incoming, MqttOptions, QoS};
 use tokio::sync::{broadcast, mpsc};
 
@@ -29,23 +29,40 @@ pub async fn start_mqtt_client(
 
     let (client, mut event_loop) = rumqttc::AsyncClient::new(options, 32);
 
+    tokio::spawn({
+        let topic_prefix = topic_prefix.clone();
+        let client = client.clone();
+
+        async move {
+            let mut is_connected = false;
+            let mut is_first_try = true;
+
+            loop {
+                match event_loop.poll().await {
+                    Err(error) => {
+                        if is_first_try {
+                            error!("MQTT connection failed (retrying in the background): {error}");
+                            is_first_try = false;
+                        } else if is_connected {
+                            error!("MQTT connection lost: {error}");
+                            handler_hosts_config_sender.send(None).await.unwrap();
+                        }
+
+                        is_connected = false;
+                        tokio::time::sleep(Duration::from_secs(2)).await;
+                    }
+                    Ok(Event::Incoming(event)) => {
+                        match event {
+                            Incoming::ConnAck(_) => {
+                                info!("MQTT connection established.");
+                                is_connected = true;
+                                is_first_try = false;
+
                                 client.subscribe(format!("{topic_prefix}/config"), QoS::ExactlyOnce).await.unwrap();
                                 client.subscribe(format!("{topic_prefix}/keys/+/+/events"), QoS::ExactlyOnce).await.unwrap();
                                 client.subscribe(format!("{topic_prefix}/knobs/+/+/events"), QoS::ExactlyOnce).await.unwrap();
-
-    tokio::spawn({
-        let topic_prefix = topic_prefix.clone();
-
-        async move {
-            loop {
-                let poll_result = event_loop.poll().await;
-
-                match poll_result {
-                    Err(error) => {
-                        error!("{error}")
                             }
-                    Ok(event) => {
+                            Incoming::Publish(event) => {
-                        if let Event::Incoming(Incoming::Publish(event)) = event {
                                 let topic_segments = event.topic.strip_prefix(&topic_prefix).unwrap().split('/').skip(1).collect::<Vec<&str>>();
 
                                 if topic_segments[0] == "config" {
@@ -71,7 +88,7 @@ pub async fn start_mqtt_client(
                                                         },
                                                         event,
                                                     })
-                                                .unwrap();
+                                                    .ok(); // This can be Err when events are received before the configuration
                                             } else {
                                                 log::error!("Could not deserialize the latest event from {}", event.topic);
                                             };
@@ -86,7 +103,7 @@ pub async fn start_mqtt_client(
                                                         },
                                                         event,
                                                     })
-                                                .unwrap();
+                                                    .ok(); // This can be Err when events are received before the configuration
                                             } else {
                                                 log::error!("Could not deserialize the latest event from {}", event.topic);
                                             };
@@ -95,8 +112,11 @@ pub async fn start_mqtt_client(
                                     }
                                 }
                             }
+                            _ => {}
                         }
                     }
+                    _ => {}
+                }
             }
         }
     });