From 64f3052b76e1e605329c34ced21aafada3a81122 Mon Sep 17 00:00:00 2001 From: Dreaded_X Date: Wed, 26 Aug 2026 05:44:33 +0200 Subject: [PATCH] feat: Migrate to rumqttc-next --- Cargo.lock | 247 +++++++++++++++++------ Cargo.toml | 4 +- automation_devices/Cargo.toml | 2 +- automation_devices/src/contact_sensor.rs | 5 +- automation_devices/src/hue_switch.rs | 5 +- automation_devices/src/ikea_remote.rs | 5 +- automation_devices/src/light_sensor.rs | 5 +- automation_devices/src/presence.rs | 9 +- automation_devices/src/wake_on_lan.rs | 5 +- automation_devices/src/washer.rs | 5 +- automation_devices/src/zigbee/light.rs | 27 ++- automation_devices/src/zigbee/outlet.rs | 15 +- automation_lib/Cargo.toml | 2 +- automation_lib/src/mqtt.rs | 11 +- 14 files changed, 250 insertions(+), 97 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c271c1e..0f7359c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -62,7 +62,27 @@ checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667" dependencies = [ "proc-macro2", "quote", - "syn 3.0.3", + "syn 3.0.4", +] + +[[package]] +name = "async-tungstenite" +version = "0.34.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8447f02eaa65412035e2d3eeaa3fc82bbb8d7137c84c5976b4af685136012ee9" +dependencies = [ + "atomic-waker", + "futures-core", + "futures-io", + "futures-task", + "futures-util", + "log", + "pin-project-lite", + "rustls-native-certs", + "rustls-pki-types", + "tokio", + "tokio-rustls", + "tungstenite", ] [[package]] @@ -94,7 +114,7 @@ dependencies = [ "lua_typed", "mlua", "reqwest", - "rumqttc", + "rumqttc-next", "serde", "serde_json", "thiserror", @@ -126,7 +146,7 @@ dependencies = [ "lua_typed", "mlua", "reqwest", - "rumqttc", + "rumqttc-next", "serde", "serde_json", "serde_repr", @@ -151,7 +171,7 @@ dependencies = [ "inventory", "lua_typed", "mlua", - "rumqttc", + "rumqttc-next", "serde", "serde_json", "thiserror", @@ -167,7 +187,7 @@ dependencies = [ "mlua", "proc-macro2", "quote", - "syn 3.0.3", + "syn 3.0.4", ] [[package]] @@ -277,6 +297,15 @@ version = "2.13.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b588b76d00fde79687d7646a9b5bdf3cc0f655e0bbd080335a95d7e96f3587da" +[[package]] +name = "block-buffer" +version = "0.10.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" +dependencies = [ + "generic-array", +] + [[package]] name = "bme280" version = "0.5.1" @@ -342,7 +371,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" dependencies = [ "cfg-if", - "cpufeatures", + "cpufeatures 0.3.0", "rand_core 0.10.1", ] @@ -380,9 +409,9 @@ dependencies = [ [[package]] name = "combine" -version = "4.6.7" +version = "4.6.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ba5a308b75df32fe02788e748662718f03fde005016435c444eea572398219fd" +checksum = "cfc320937d09e6de266b31b9afb480f197d7a861be86be7cb2ea7e5d1bfffc5e" dependencies = [ "bytes", "memchr", @@ -435,6 +464,15 @@ version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" +[[package]] +name = "cpufeatures" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" +dependencies = [ + "libc", +] + [[package]] name = "cpufeatures" version = "0.3.0" @@ -470,6 +508,16 @@ version = "0.8.22" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "61803da095bee82a81bb1a452ecc25d3b2f1416d1897eb86430c6159ef717c17" +[[package]] +name = "crypto-common" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" +dependencies = [ + "generic-array", + "typenum", +] + [[package]] name = "darling" version = "0.20.11" @@ -505,6 +553,12 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "data-encoding" +version = "2.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06" + [[package]] name = "defmt" version = "0.3.100" @@ -576,6 +630,16 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "digest" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" +dependencies = [ + "block-buffer", + "crypto-common", +] + [[package]] name = "displaydoc" version = "0.2.7" @@ -584,7 +648,7 @@ checksum = "c6232dd377dcc64799954cbd3a9bb882e9cdc1308ccd87b1c098f1fb2eaf82a8" dependencies = [ "proc-macro2", "quote", - "syn 3.0.3", + "syn 3.0.4", ] [[package]] @@ -675,9 +739,9 @@ checksum = "1d674e81391d1e1ab681a28d99df07927c6d4aa5b027d7da16ba32d1d21ecd99" [[package]] name = "flume" -version = "0.11.1" +version = "0.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095" +checksum = "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be" dependencies = [ "futures-core", "futures-sink", @@ -761,7 +825,7 @@ checksum = "9fb9654ba8355388abeb8dcb4fc62f511300867002afc858860463bdd9fe0c44" dependencies = [ "proc-macro2", "quote", - "syn 3.0.3", + "syn 3.0.4", ] [[package]] @@ -799,6 +863,16 @@ dependencies = [ "slab", ] +[[package]] +name = "generic-array" +version = "0.14.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" +dependencies = [ + "typenum", + "version_check", +] + [[package]] name = "getrandom" version = "0.2.17" @@ -879,7 +953,7 @@ version = "0.1.0" dependencies = [ "proc-macro2", "quote", - "syn 3.0.3", + "syn 3.0.4", ] [[package]] @@ -1290,9 +1364,9 @@ dependencies = [ [[package]] name = "log" -version = "0.4.33" +version = "0.4.34" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" +checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" [[package]] name = "lru-slab" @@ -1452,6 +1526,16 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "mqttbytes-core-next" +version = "0.34.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "366b6ba2b4209ca4bc5ac731ccddf570d09831981eed07e5fbd63564cf0cf1aa" +dependencies = [ + "bytes", + "thiserror", +] + [[package]] name = "nanoid" version = "0.5.0" @@ -1892,23 +1976,49 @@ dependencies = [ ] [[package]] -name = "rumqttc" -version = "0.25.1" +name = "rumqttc-core-next" +version = "0.34.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0feff8d882bff0b2fddaf99355a10336d43dd3ed44204f85ece28cf9626ab519" +checksum = "249896ab27ed630590971738264baa8f722f18965d2e387c706c40a3c2a572cc" dependencies = [ + "async-tungstenite", + "rustls-native-certs", + "rustls-pki-types", + "rustls-webpki", + "thiserror", + "tokio", + "tokio-rustls", +] + +[[package]] +name = "rumqttc-next" +version = "0.34.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "477c9bbfba8f3aecc7aad31c6de2eacb75822efaa18e7aeecb8d3d8e534fbf07" +dependencies = [ + "rumqttc-v5-next", +] + +[[package]] +name = "rumqttc-v5-next" +version = "0.34.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3dfa6ddcc7a7dd5688f9bf78d8f81cb94f367bce56c055d8d94cf81ecb0518bf" +dependencies = [ + "async-tungstenite", "bytes", "fixedbitset", "flume", "futures-util", "log", + "mqttbytes-core-next", + "rumqttc-core-next", "rustls-native-certs", - "rustls-pemfile", - "rustls-webpki 0.102.8", + "rustls-pki-types", + "rustls-webpki", "thiserror", "tokio", "tokio-rustls", - "tokio-stream", "tokio-util", ] @@ -1934,10 +2044,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06" dependencies = [ "aws-lc-rs", - "log", "once_cell", "rustls-pki-types", - "rustls-webpki 0.103.15", + "rustls-webpki", "subtle", "zeroize", ] @@ -1954,15 +2063,6 @@ dependencies = [ "security-framework", ] -[[package]] -name = "rustls-pemfile" -version = "2.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dce314e5fee3f39953d46bb63bb8a46d40c2f8fb7cc5a3b6cab2bde9721d6e50" -dependencies = [ - "rustls-pki-types", -] - [[package]] name = "rustls-pki-types" version = "1.15.1" @@ -1987,7 +2087,7 @@ dependencies = [ "rustls", "rustls-native-certs", "rustls-platform-verifier-android", - "rustls-webpki 0.103.15", + "rustls-webpki", "security-framework", "security-framework-sys", "webpki-root-certs", @@ -2000,17 +2100,6 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f" -[[package]] -name = "rustls-webpki" -version = "0.102.8" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "64ca1bc8749bd4cf37b5ce386cc146580777b4e8572c7b97baf22c83f444bee9" -dependencies = [ - "ring", - "rustls-pki-types", - "untrusted", -] - [[package]] name = "rustls-webpki" version = "0.103.15" @@ -2125,7 +2214,7 @@ checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" dependencies = [ "proc-macro2", "quote", - "syn 3.0.3", + "syn 3.0.4", ] [[package]] @@ -2160,7 +2249,7 @@ checksum = "8d3b1629de253c70a0508c3899572da79ca359fdab27c7920ff00406df418906" dependencies = [ "proc-macro2", "quote", - "syn 3.0.3", + "syn 3.0.4", ] [[package]] @@ -2184,6 +2273,17 @@ dependencies = [ "serde", ] +[[package]] +name = "sha1" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a978451301f4db1d02937a4ab3ccce137717b81826e79b7d49ffe3244a13c3b8" +dependencies = [ + "cfg-if", + "cpufeatures 0.2.17", + "digest", +] + [[package]] name = "sharded-slab" version = "0.1.7" @@ -2315,9 +2415,9 @@ dependencies = [ [[package]] name = "syn" -version = "3.0.3" +version = "3.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3" +checksum = "e6275cddf4610d1775e6d1fe9469b2e77d0f39fd98fb7450901b821e0c53649f" dependencies = [ "proc-macro2", "quote", @@ -2361,7 +2461,7 @@ checksum = "bc04cd3e1236dd4a98afca4569f2deb3f120e5422a4023be2cb683f8486292af" dependencies = [ "proc-macro2", "quote", - "syn 3.0.3", + "syn 3.0.4", ] [[package]] @@ -2437,7 +2537,7 @@ checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" dependencies = [ "proc-macro2", "quote", - "syn 3.0.3", + "syn 3.0.4", ] [[package]] @@ -2450,17 +2550,6 @@ dependencies = [ "tokio", ] -[[package]] -name = "tokio-stream" -version = "0.1.19" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a3d06f0b082ba57c26b79407372e57cf2a1e28124f78e9479fe80322cf53420b" -dependencies = [ - "futures-core", - "pin-project-lite", - "tokio", -] - [[package]] name = "tokio-util" version = "0.7.19" @@ -2616,12 +2705,36 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "tungstenite" +version = "0.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c01152af293afb9c7c2a57e4b559c5620b421f6d133261c60dd2d0cdb38e6b8" +dependencies = [ + "bytes", + "data-encoding", + "http", + "httparse", + "log", + "rand 0.9.5", + "rustls", + "rustls-pki-types", + "sha1", + "thiserror", +] + [[package]] name = "typeid" version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bc7d623258602320d5c55d1bc22793b57daff0ec7efc270ea7d55ce1d5f5471c" +[[package]] +name = "typenum" +version = "1.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" + [[package]] name = "unicase" version = "2.9.0" @@ -2666,9 +2779,9 @@ checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" [[package]] name = "uuid" -version = "1.24.1" +version = "1.25.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2cefc03fd367c0c6d4305de1b312cf00248c4114f4a0418ce6a6af769e3b0bd9" +checksum = "f053576934f05a761a402421fbbe3d425d9366f75f978806a037b3ca481abecc" dependencies = [ "getrandom 0.4.3", "js-sys", @@ -2687,6 +2800,12 @@ version = "0.2.15" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426" +[[package]] +name = "version_check" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" + [[package]] name = "wakey" version = "0.4.1" @@ -3095,7 +3214,7 @@ checksum = "34df6fc39dbd26ddc9c10e6a2984476e13acce22e64e4487636ef494369225da" dependencies = [ "proc-macro2", "quote", - "syn 3.0.3", + "syn 3.0.4", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 6a1581b..eaa4f37 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -51,7 +51,7 @@ reqwest = { version = "0.13.3", features = [ "json", "rustls", ], default-features = false } # Use rustls, since the other packages also use rustls -rumqttc = "0.25.1" +rumqttc-next = "0.34.0" serde = { version = "1.0.228", features = ["derive"] } serde_json = "1.0.149" serde_repr = "0.1.20" @@ -80,7 +80,6 @@ lua_typed = { workspace = true } inventory = { workspace = true } mlua = { workspace = true } reqwest = { workspace = true } -rumqttc = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } thiserror = { workspace = true } @@ -88,6 +87,7 @@ tokio = { workspace = true } tokio-cron-scheduler = { workspace = true } tracing = { workspace = true } tracing-subscriber = { workspace = true } +rumqttc-next = { workspace = true } [patch.crates-io] wakey = { git = "https://git.huizinga.dev/Dreaded_X/wakey" } diff --git a/automation_devices/Cargo.toml b/automation_devices/Cargo.toml index 3491da8..e936c01 100644 --- a/automation_devices/Cargo.toml +++ b/automation_devices/Cargo.toml @@ -17,7 +17,7 @@ inventory = { workspace = true } lua_typed = { workspace = true } mlua = { workspace = true } reqwest = { workspace = true } -rumqttc = { workspace = true } +rumqttc-next = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } serde_repr = { workspace = true } diff --git a/automation_devices/src/contact_sensor.rs b/automation_devices/src/contact_sensor.rs index 45cba7a..acc95df 100644 --- a/automation_devices/src/contact_sensor.rs +++ b/automation_devices/src/contact_sensor.rs @@ -159,7 +159,10 @@ impl OpenClose for ContactSensor { #[async_trait] impl OnMqtt for ContactSensor { async fn on_mqtt(&self, message: rumqttc::Publish) { - if !rumqttc::matches(&message.topic, &self.config.mqtt.topic) { + if !rumqttc::matches( + str::from_utf8(&message.topic).expect("Topic should be valid"), + &self.config.mqtt.topic, + ) { return; } diff --git a/automation_devices/src/hue_switch.rs b/automation_devices/src/hue_switch.rs index 3487ac2..39fb522 100644 --- a/automation_devices/src/hue_switch.rs +++ b/automation_devices/src/hue_switch.rs @@ -98,7 +98,10 @@ impl LuaDeviceCreate for HueSwitch { impl OnMqtt for HueSwitch { async fn on_mqtt(&self, message: Publish) { // Check if the message is from the device itself or from a remote - if matches(&message.topic, &self.config.mqtt.topic) { + if matches( + str::from_utf8(&message.topic).expect("Topic should be valid"), + &self.config.mqtt.topic, + ) { let message = match serde_json::from_slice::(&message.payload) { Ok(message) => message, Err(err) => { diff --git a/automation_devices/src/ikea_remote.rs b/automation_devices/src/ikea_remote.rs index 6d522fb..9b80765 100644 --- a/automation_devices/src/ikea_remote.rs +++ b/automation_devices/src/ikea_remote.rs @@ -70,7 +70,10 @@ impl LuaDeviceCreate for IkeaRemote { impl OnMqtt for IkeaRemote { async fn on_mqtt(&self, message: Publish) { // Check if the message is from the deviec itself or from a remote - if matches(&message.topic, &self.config.mqtt.topic) { + if matches( + str::from_utf8(&message.topic).expect("Topic should be valid"), + &self.config.mqtt.topic, + ) { let message = match RemoteMessage::try_from(message) { Ok(message) => message, Err(err) => { diff --git a/automation_devices/src/light_sensor.rs b/automation_devices/src/light_sensor.rs index 4ad04ae..0e05dbf 100644 --- a/automation_devices/src/light_sensor.rs +++ b/automation_devices/src/light_sensor.rs @@ -85,7 +85,10 @@ impl Device for LightSensor { #[async_trait] impl OnMqtt for LightSensor { async fn on_mqtt(&self, message: Publish) { - if !rumqttc::matches(&message.topic, &self.config.mqtt.topic) { + if !rumqttc::matches( + str::from_utf8(&message.topic).expect("Topic should be valid"), + &self.config.mqtt.topic, + ) { return; } diff --git a/automation_devices/src/presence.rs b/automation_devices/src/presence.rs index a77327c..cc5137b 100644 --- a/automation_devices/src/presence.rs +++ b/automation_devices/src/presence.rs @@ -105,7 +105,10 @@ impl Device for Presence { #[async_trait] impl OnMqtt for Presence { async fn on_mqtt(&self, message: Publish) { - if !rumqttc::matches(&message.topic, &self.config.mqtt.topic) { + if !rumqttc::matches( + str::from_utf8(&message.topic).expect("Topic should be valid"), + &self.config.mqtt.topic, + ) { return; } @@ -116,7 +119,9 @@ impl OnMqtt for Presence { .find('+') .or(self.config.mqtt.topic.find('#')) .expect("Presence::create fails if it does not contain wildcards"); - let device_name = message.topic[offset..].into(); + let device_name: String = str::from_utf8(&message.topic[offset..]) + .expect("Topic should be valid") + .into(); if message.payload.is_empty() { // Remove the device from the map diff --git a/automation_devices/src/wake_on_lan.rs b/automation_devices/src/wake_on_lan.rs index 72dfecf..3ba5b0f 100644 --- a/automation_devices/src/wake_on_lan.rs +++ b/automation_devices/src/wake_on_lan.rs @@ -66,7 +66,10 @@ impl Device for WakeOnLAN { #[async_trait] impl OnMqtt for WakeOnLAN { async fn on_mqtt(&self, message: Publish) { - if !rumqttc::matches(&message.topic, &self.config.mqtt.topic) { + if !rumqttc::matches( + str::from_utf8(&message.topic).expect("Topic should be valid"), + &self.config.mqtt.topic, + ) { return; } diff --git a/automation_devices/src/washer.rs b/automation_devices/src/washer.rs index a590f73..4e84513 100644 --- a/automation_devices/src/washer.rs +++ b/automation_devices/src/washer.rs @@ -89,7 +89,10 @@ const HYSTERESIS: isize = 10; #[async_trait] impl OnMqtt for Washer { async fn on_mqtt(&self, message: Publish) { - if !rumqttc::matches(&message.topic, &self.config.mqtt.topic) { + if !rumqttc::matches( + str::from_utf8(&message.topic).expect("Topic should be valid"), + &self.config.mqtt.topic, + ) { return; } diff --git a/automation_devices/src/zigbee/light.rs b/automation_devices/src/zigbee/light.rs index cd5b01a..450026a 100644 --- a/automation_devices/src/zigbee/light.rs +++ b/automation_devices/src/zigbee/light.rs @@ -1,3 +1,4 @@ +use core::str; use std::fmt::Debug; use std::ops::Deref; use std::sync::Arc; @@ -16,7 +17,7 @@ use google_home::errors::ErrorCode; use google_home::traits::{Brightness, Color, ColorSetting, ColorTemperatureRange, OnOff}; use google_home::types::Type; use lua_typed::Typed; -use rumqttc::{Publish, matches}; +use rumqttc::{Publish, PublishOptions, matches}; use serde::{Deserialize, Serialize}; use serde_json::json; use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard}; @@ -177,7 +178,10 @@ where impl OnMqtt for LightOnOff { async fn on_mqtt(&self, message: Publish) { // Check if the message is from the device itself or from a remote - if matches(&message.topic, &self.config.mqtt.topic) { + if matches( + str::from_utf8(&message.topic).expect("Topic should be valid"), + &self.config.mqtt.topic, + ) { let state = match serde_json::from_slice::(&message.payload) { Ok(state) => state, Err(err) => { @@ -210,7 +214,10 @@ impl OnMqtt for LightOnOff { impl OnMqtt for LightBrightness { async fn on_mqtt(&self, message: Publish) { // Check if the message is from the deviec itself or from a remote - if matches(&message.topic, &self.config.mqtt.topic) { + if matches( + str::from_utf8(&message.topic).expect("Topic should be valid"), + &self.config.mqtt.topic, + ) { let state = match serde_json::from_slice::(&message.payload) { Ok(state) => state, Err(err) => { @@ -249,7 +256,10 @@ impl OnMqtt for LightBrightness { impl OnMqtt for LightColorTemperature { async fn on_mqtt(&self, message: Publish) { // Check if the message is from the deviec itself or from a remote - if matches(&message.topic, &self.config.mqtt.topic) { + if matches( + str::from_utf8(&message.topic).expect("Topic should be valid"), + &self.config.mqtt.topic, + ) { let state = match serde_json::from_slice::(&message.payload) { Ok(state) => state, Err(err) => { @@ -342,9 +352,8 @@ where .client .publish( &topic, - rumqttc::QoS::AtLeastOnce, - false, serde_json::to_string(&message).unwrap(), + PublishOptions::at_least_once(), ) .await .map_err(|err| warn!("Failed to update state on {topic}: {err}")) @@ -386,9 +395,8 @@ where .client .publish( &topic, - rumqttc::QoS::AtLeastOnce, - false, serde_json::to_string(&message).unwrap(), + PublishOptions::at_least_once(), ) .await .map_err(|err| warn!("Failed to update state on {topic}: {err}")) @@ -434,9 +442,8 @@ where .client .publish( &topic, - rumqttc::QoS::AtLeastOnce, - false, serde_json::to_string(&message).unwrap(), + PublishOptions::at_least_once(), ) .await .map_err(|err| warn!("Failed to update state on {topic}: {err}")) diff --git a/automation_devices/src/zigbee/outlet.rs b/automation_devices/src/zigbee/outlet.rs index 406621a..8c7b21d 100644 --- a/automation_devices/src/zigbee/outlet.rs +++ b/automation_devices/src/zigbee/outlet.rs @@ -16,7 +16,7 @@ use google_home::errors::ErrorCode; use google_home::traits::OnOff; use google_home::types::Type; use lua_typed::Typed; -use rumqttc::{Publish, matches}; +use rumqttc::{Publish, PublishOptions, matches}; use serde::{Deserialize, Serialize}; use serde_json::json; use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard}; @@ -162,7 +162,10 @@ where impl OnMqtt for OutletOnOff { async fn on_mqtt(&self, message: Publish) { // Check if the message is from the device itself or from a remote - if matches(&message.topic, &self.config.mqtt.topic) { + if matches( + str::from_utf8(&message.topic).expect("Topic should be valid"), + &self.config.mqtt.topic, + ) { let state = match serde_json::from_slice::(&message.payload) { Ok(state) => state, Err(err) => { @@ -195,7 +198,10 @@ impl OnMqtt for OutletOnOff { impl OnMqtt for OutletPower { async fn on_mqtt(&self, message: Publish) { // Check if the message is from the deviec itself or from a remote - if matches(&message.topic, &self.config.mqtt.topic) { + if matches( + str::from_utf8(&message.topic).expect("Topic should be valid"), + &self.config.mqtt.topic, + ) { let state = match serde_json::from_slice::(&message.payload) { Ok(state) => state, Err(err) => { @@ -284,9 +290,8 @@ where .client .publish( &topic, - rumqttc::QoS::AtLeastOnce, - false, serde_json::to_string(&message).unwrap(), + PublishOptions::at_least_once(), ) .await .map_err(|err| warn!("Failed to update state on {topic}: {err}")) diff --git a/automation_lib/Cargo.toml b/automation_lib/Cargo.toml index b05bee5..d18d57b 100644 --- a/automation_lib/Cargo.toml +++ b/automation_lib/Cargo.toml @@ -15,7 +15,7 @@ hostname = { workspace = true } inventory = { workspace = true } lua_typed = { workspace = true } mlua = { workspace = true } -rumqttc = { workspace = true } +rumqttc-next = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } thiserror = { workspace = true } diff --git a/automation_lib/src/mqtt.rs b/automation_lib/src/mqtt.rs index bd1b48f..b17436a 100644 --- a/automation_lib/src/mqtt.rs +++ b/automation_lib/src/mqtt.rs @@ -1,10 +1,9 @@ use std::ops::{Deref, DerefMut}; -use std::time::Duration; use automation_macro::LuaDeviceConfig; use lua_typed::Typed; use mlua::FromLua; -use rumqttc::{AsyncClient, Event, Incoming, MqttOptions, Transport}; +use rumqttc::{AsyncClient, Event, Incoming, MqttOptions, PublishOptions, Transport}; use serde::Deserialize; use tracing::{debug, warn}; @@ -24,9 +23,9 @@ pub struct MqttConfig { impl From for MqttOptions { fn from(value: MqttConfig) -> Self { - let mut mqtt_options = MqttOptions::new(value.client_name, value.host, value.port); + let mut mqtt_options = MqttOptions::new(value.client_name, (value.host, value.port)); mqtt_options.set_credentials(value.username, value.password); - mqtt_options.set_keep_alive(Duration::from_secs(5)); + mqtt_options.set_keep_alive(5); if value.tls { mqtt_options.set_transport(Transport::tls_with_default_config()); @@ -92,7 +91,7 @@ impl mlua::UserData for WrappedAsyncClient { debug!("message = {message}"); this.0 - .publish(topic, rumqttc::QoS::AtLeastOnce, true, message) + .publish(topic, message, PublishOptions::at_least_once().retained()) .await .unwrap(); @@ -104,7 +103,7 @@ impl mlua::UserData for WrappedAsyncClient { pub fn start(config: MqttConfig, event_channel: &EventChannel) -> WrappedAsyncClient { let tx = event_channel.get_tx(); - let (client, mut eventloop) = AsyncClient::new(config.into(), 100); + let (client, mut eventloop) = AsyncClient::builder(config.into()).capacity(100).build(); tokio::spawn(async move { debug!("Listening for MQTT events");