From d83189417ada67677c93d8c3da04120400425d35 Mon Sep 17 00:00:00 2001 From: ju6ge Date: Sun, 10 May 2026 17:30:02 +0200 Subject: [PATCH] feat: store MQTT config update payload --- rust/src/mqtt.rs | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/rust/src/mqtt.rs b/rust/src/mqtt.rs index 2d5718f..d29f68a 100644 --- a/rust/src/mqtt.rs +++ b/rust/src/mqtt.rs @@ -8,6 +8,8 @@ use alloc::{format, string::ToString}; use core::sync::atomic::Ordering; use embassy_executor::Spawner; use embassy_net::Stack; +use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex; +use embassy_sync::mutex::Mutex; use embassy_sync::once_lock::OnceLock; use embassy_time::{Duration, Timer, WithTimeout}; use log::info; @@ -37,6 +39,7 @@ static MQTT_CONNECTED_EVENT_RECEIVED: AtomicBool = AtomicBool::new(false); static MQTT_ROUND_TRIP_RECEIVED: AtomicBool = AtomicBool::new(false); pub static MQTT_STAY_ALIVE: AtomicBool = AtomicBool::new(false); static MQTT_BASE_TOPIC: OnceLock = OnceLock::new(); +static MQTT_CONFIG_UPDATE_PAYLOAD: Mutex> = Mutex::new(None); pub fn is_stay_alive() -> bool { MQTT_STAY_ALIVE.load(Ordering::Relaxed) @@ -141,6 +144,7 @@ pub async fn mqtt_init( let last_will_topic = format!("{base_topic}/state"); let round_trip_topic = format!("{base_topic}/internal/roundtrip"); let stay_alive_topic = format!("{base_topic}/stay_alive"); + let config_update_payload_topic = format!("{base_topic}/config/update_payload"); let mut builder: McutieBuilder<'_, String, PublishDisplay, 0> = McutieBuilder::new(stack, "plant ctrl", mqtt_url); @@ -159,10 +163,11 @@ pub async fn mqtt_init( //TODO make configurable builder = builder.with_device_id("plantctrl"); - let builder: McutieBuilder<'_, String, PublishDisplay, 2> = builder + let builder: McutieBuilder<'_, String, PublishDisplay, 3> = builder .with_subscriptions([ Topic::General(round_trip_topic.clone()), Topic::General(stay_alive_topic.clone()), + Topic::General(config_update_payload_topic.clone()), ]); let keep_alive = Duration::from_secs(60 * 60 * 2).as_secs() as u16; @@ -172,6 +177,7 @@ pub async fn mqtt_init( receiver, round_trip_topic.clone(), stay_alive_topic.clone(), + config_update_payload_topic.clone(), )?); spawner.spawn(mqtt_runner(task)?); @@ -218,7 +224,7 @@ pub async fn mqtt_init( #[embassy_executor::task] async fn mqtt_runner( - task: McutieTask<'static, String, PublishDisplay<'static, String, &'static str>, 2>, + task: McutieTask<'static, String, PublishDisplay<'static, String, &'static str>, 3>, ) { task.run().await; } @@ -228,6 +234,7 @@ async fn mqtt_incoming_task( receiver: McutieReceiver, round_trip_topic: String, stay_alive_topic: String, + config_update_payload_topic: String, ) { loop { let message = receiver.receive().await; @@ -253,6 +260,11 @@ async fn mqtt_incoming_task( }; log(LogMessage::MqttStayAliveRec, a, 0, "", ""); MQTT_STAY_ALIVE.store(value, Ordering::Relaxed); + } else if subtopic.eq(config_update_payload_topic.as_str()) { + let payload_str = String::from_utf8_lossy(&payload[..]).to_string(); + let mut buffer = MQTT_CONFIG_UPDATE_PAYLOAD.lock().await; + *buffer = Some(payload_str); + info!("MQTT config update payload received"); } else { log(LogMessage::UnknownTopic, 0, 0, "", &topic); }