feat: Migrate to rumqttc-next

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"
+name = "rumqttc-core-next"
-version = "0.25.1"
+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-pki-types",
- "rustls-webpki 0.102.8",
+ "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]]

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" }

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 }

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;
         }
 

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::<State>(&message.payload) {
                 Ok(message) => message,
                 Err(err) => {

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) => {

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;
         }
 

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

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;
         }
 

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;
         }
 

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::<StateOnOff>(&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::<StateBrightness>(&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::<StateColorTemperature>(&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}"))

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::<StateOnOff>(&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::<StatePower>(&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}"))

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 }

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<MqttConfig> 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");