diff --git a/Cargo.lock b/Cargo.lock index a858994..a2ed4ea 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -154,33 +154,6 @@ version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0c4b4d0bd25bd0b74681c0ad21497610ce1b7c91b1022cd21c80c6fbdd9476b0" -[[package]] -name = "aws-lc-rs" -version = "1.8.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4ae74d9bd0a7530e8afd1770739ad34b36838829d6ad61818f9230f683f5ad77" -dependencies = [ - "aws-lc-sys", - "mirai-annotations", - "paste", - "zeroize", -] - -[[package]] -name = "aws-lc-sys" -version = "0.20.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2e89b6941c2d1a7045538884d6e760ccfffdf8e1ffc2613d8efa74305e1f3752" -dependencies = [ - "bindgen", - "cc", - "cmake", - "dunce", - "fs_extra", - "libc", - "paste", -] - [[package]] name = "az" version = "1.2.1" @@ -223,29 +196,6 @@ version = "0.21.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9d297deb1925b89f2ccc13d7635fa0714f12c87adce1c75356b39ca9b7178567" -[[package]] -name = "bindgen" -version = "0.69.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a00dc851838a2120612785d195287475a3ac45514741da670b735818822129a0" -dependencies = [ - "bitflags 2.9.1", - "cexpr", - "clang-sys", - "itertools 0.10.5", - "lazy_static", - "lazycell", - "log", - "prettyplease", - "proc-macro2", - "quote", - "regex", - "rustc-hash 1.1.0", - "shlex", - "syn 2.0.87", - "which", -] - [[package]] name = "bit-set" version = "0.5.3" @@ -341,19 +291,6 @@ name = "cc" version = "1.1.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2aba8f4e9906c7ce3c73463f62a7f0c65183ada1a2d47e397cc8810827f9694f" -dependencies = [ - "jobserver", - "libc", -] - -[[package]] -name = "cexpr" -version = "0.6.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6fac387a98bb7c37292057cffc56d62ecb629900026402633ae9160df93a8766" -dependencies = [ - "nom", -] [[package]] name = "cfg-if" @@ -371,26 +308,6 @@ dependencies = [ "inout", ] -[[package]] -name = "clang-sys" -version = "1.8.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0b023947811758c97c59bf9d1c188fd619ad4718dcaa767947df1cadb14f39f4" -dependencies = [ - "glob", - "libc", - "libloading", -] - -[[package]] -name = "cmake" -version = "0.1.50" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a31c789563b815f77f4250caee12365734369f942439b7defd71e18a48197130" -dependencies = [ - "cc", -] - [[package]] name = "codespan-reporting" version = "0.11.1" @@ -549,7 +466,7 @@ checksum = "59d6ec798758febb089bd969109385b48dd0fb966193fe97a5f0f8b2d622145e" dependencies = [ "cortex-m", "cortex-m-rt", - "defmt", + "defmt 0.3.100", "embassy-futures", "embassy-net-driver-channel", "embassy-sync 0.5.0", @@ -566,7 +483,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f429446fe9420ee0a8743fdffb1d08a90f9332116635bbc819f63649264e480c" dependencies = [ "cyw43", - "defmt", + "defmt 0.3.100", "embassy-rp", "fixed", "pio", @@ -624,9 +541,18 @@ dependencies = [ [[package]] name = "defmt" -version = "0.3.8" +version = "0.3.100" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a99dd22262668b887121d4672af5a64b238f026099f1a2a1b322066c9ecfe9e0" +checksum = "f0963443817029b2024136fc4dd07a5107eb8f977eaf18fcd1fdeb11306b64ad" +dependencies = [ + "defmt 1.0.1", +] + +[[package]] +name = "defmt" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "548d977b6da32fa1d1fda2876453da1e7df63ad0304c8b3dae4dbe7b96f39b78" dependencies = [ "bitflags 1.3.2", "defmt-macros", @@ -634,12 +560,12 @@ dependencies = [ [[package]] name = "defmt-macros" -version = "0.3.9" +version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e3a9f309eff1f79b3ebdf252954d90ae440599c26c2c553fe87a2d17195f2dcb" +checksum = "3d4fc12a85bcf441cfe44344c4b72d58493178ce635338a3f3b78943aceb258e" dependencies = [ "defmt-parser", - "proc-macro-error", + "proc-macro-error2", "proc-macro2", "quote", "syn 2.0.87", @@ -647,11 +573,11 @@ dependencies = [ [[package]] name = "defmt-parser" -version = "0.3.4" +version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ff4a5fefe330e8d7f31b16a318f9ce81000d8e35e69b93eae154d16d2278f70f" +checksum = "10d60334b3b2e7c9d91ef8150abfb6fa4c1c39ebbcf4a81c2e346aad939fee3e" dependencies = [ - "thiserror 1.0.68", + "thiserror 2.0.12", ] [[package]] @@ -661,7 +587,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bab697b3dbbc1750b7c8b821aa6f6e7f2480b47a99bc057a2ed7b170ebef0c51" dependencies = [ "critical-section", - "defmt", + "defmt 0.3.100", ] [[package]] @@ -738,12 +664,6 @@ version = "0.15.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1aaf95b3e5c8f23aa320147307562d361db0ae0d51242340f558153b4eb2439b" -[[package]] -name = "dunce" -version = "1.0.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "56ce8c6da7551ec6c462cbaf3bfbc75131ebbfa1c944aeaa9dab51ca1c5f0c3b" - [[package]] name = "either" version = "1.13.0" @@ -794,7 +714,7 @@ checksum = "ec648daedd2143466eff4b3e8002024f9f6c1de4ab7666bb679688752624c925" dependencies = [ "cortex-m", "critical-section", - "defmt", + "defmt 0.3.100", "document-features", "embassy-executor-macros", "embassy-time-driver", @@ -827,7 +747,7 @@ checksum = "a0ec47cf8bab914018d4bd2b4f0aaeb46e4f52ab1e7985df88aeef2c6eda5aed" dependencies = [ "cortex-m", "critical-section", - "defmt", + "defmt 0.3.100", "num-traits", ] @@ -839,7 +759,7 @@ checksum = "55cf91dd36dfd623de32242af711fd294d41159f02130052fc93c5c5ba93febe" dependencies = [ "as-slice 0.2.1", "atomic-pool", - "defmt", + "defmt 0.3.100", "document-features", "embassy-net-driver", "embassy-sync 0.5.0", @@ -860,7 +780,7 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "524eb3c489760508f71360112bca70f6e53173e6fe48fc5f0efd0f5ab217751d" dependencies = [ - "defmt", + "defmt 0.3.100", ] [[package]] @@ -885,7 +805,7 @@ dependencies = [ "cortex-m", "cortex-m-rt", "critical-section", - "defmt", + "defmt 0.3.100", "document-features", "embassy-embedded-hal", "embassy-futures", @@ -958,7 +878,7 @@ checksum = "274c019608a9004aed3cafc871e2a3c87ce9351d537dcaab4cc5db184d4a04b1" dependencies = [ "cfg-if", "critical-section", - "defmt", + "defmt 0.3.100", "document-features", "embassy-time-driver", "embassy-time-queue-driver", @@ -990,7 +910,7 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4fc247028eae04174b6635104a35b1ed336aabef4654f5e87a8f32327d231970" dependencies = [ - "defmt", + "defmt 0.3.100", ] [[package]] @@ -1034,7 +954,7 @@ version = "0.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "edd0f118536f44f5ccd48bcb8b111bdc3de888b58c74639dfb034a357d0f206d" dependencies = [ - "defmt", + "defmt 0.3.100", ] [[package]] @@ -1043,7 +963,7 @@ version = "0.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3ff09972d4073aa8c299395be75161d582e7629cd663171d62af73c8d50dba3f" dependencies = [ - "defmt", + "defmt 0.3.100", "embedded-io", ] @@ -1091,7 +1011,7 @@ checksum = "a6efb76fdd004a4ef787640177237b83449e6c5847765ea50bf15900061fd601" dependencies = [ "aes-gcm", "atomic-polyfill", - "defmt", + "defmt 0.3.100", "digest", "embedded-io", "embedded-io-async", @@ -1121,16 +1041,6 @@ version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5443807d6dff69373d433ab9ef5378ad8df50ca6298caf15de6e52e24aaf54d5" -[[package]] -name = "errno" -version = "0.3.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "534c5cf6194dfab3db3242765c03bbe257cf92f22b38f6bc0c58d59108a820ba" -dependencies = [ - "libc", - "windows-sys", -] - [[package]] name = "exponential_distribution" version = "0.1.0" @@ -1171,7 +1081,7 @@ dependencies = [ "crc", "cyw43", "cyw43-pio", - "defmt", + "defmt 1.0.1", "defmt-rtt", "dotenvy", "embassy-executor", @@ -1183,6 +1093,7 @@ dependencies = [ "embedded-io-async", "embedded-nal-async", "heapless 0.8.0", + "mqtt", "nb 1.1.0", "panic-probe", "portable-atomic", @@ -1190,6 +1101,8 @@ dependencies = [ "reqwless", "ruff_python_ast 0.0.0 (git+https://github.com/astral-sh/ruff)", "ruff_python_parser", + "serde", + "serde_json", "static_cell", "thiserror 2.0.12", ] @@ -1228,12 +1141,6 @@ version = "1.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" -[[package]] -name = "fs_extra" -version = "1.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" - [[package]] name = "futures" version = "0.3.30" @@ -1395,12 +1302,6 @@ version = "0.29.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40ecd4077b5ae9fd2e9e169b102c6c330d0605168eb0e8bf79952b256dbefffd" -[[package]] -name = "glob" -version = "0.3.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d2fabcfbdc87f4758337ca535fb41a6d701b65693ce38287d856d1674551ec9b" - [[package]] name = "group" version = "0.13.0" @@ -1492,7 +1393,7 @@ version = "0.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0bfb9eb618601c89945a70e254898da93b13be0388091d42117462b265bb3fad" dependencies = [ - "defmt", + "defmt 0.3.100", "hash32 0.3.1", "stable_deref_trait", ] @@ -1533,15 +1434,6 @@ dependencies = [ "digest", ] -[[package]] -name = "home" -version = "0.5.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e3d1354bf6b7235cb4a0576c2619fd4ed18183f689b12b006a0ee7329eeff9a5" -dependencies = [ - "windows-sys", -] - [[package]] name = "httparse" version = "1.9.4" @@ -1636,15 +1528,6 @@ version = "1.0.15" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4a5f13b858c8d314ee3e8f639011f7ccefe71f97f96e50151fb991f267928e2c" -[[package]] -name = "jobserver" -version = "0.1.32" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "48d1dbcbbeb6a7fec7e059840aa538bd62aaccf972c7346c4d9d2059312853d0" -dependencies = [ - "libc", -] - [[package]] name = "lalrpop" version = "0.19.12" @@ -1676,34 +1559,12 @@ dependencies = [ "regex", ] -[[package]] -name = "lazy_static" -version = "1.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" - -[[package]] -name = "lazycell" -version = "1.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "830d08ce1d1d941e6b30645f1a0eb5643013d835ce3779a5fc208261dbe10f55" - [[package]] name = "libc" version = "0.2.155" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "97b3888a4aecf77e811145cadf6eef5901f4782c53886191b2f693f24761847c" -[[package]] -name = "libloading" -version = "0.8.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4979f22fdb869068da03c9f7528f8297c6fd2606bc3a4affe42e6a823fdb8da4" -dependencies = [ - "cfg-if", - "windows-targets", -] - [[package]] name = "libm" version = "0.2.8" @@ -1740,12 +1601,6 @@ dependencies = [ "pkg-config", ] -[[package]] -name = "linux-raw-sys" -version = "0.4.14" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "78b3ae25bc7c8c38cec158d1f2757ee79e9b3740fbc7ccf0e59e4b08d793fa89" - [[package]] name = "litrs" version = "0.4.1" @@ -1812,12 +1667,6 @@ version = "2.7.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "78ca9ab1a0babb1e7d5695e3530886289c18cf2f87ec19a575a0abdce112e3a3" -[[package]] -name = "minimal-lexical" -version = "0.2.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a" - [[package]] name = "miniz_oxide" version = "0.7.4" @@ -1839,22 +1688,12 @@ dependencies = [ "windows-sys", ] -[[package]] -name = "mirai-annotations" -version = "1.12.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c9be0862c1b3f26a88803c4a49de6889c10e608b3ee9344e6ef5b45fb37ad3d1" - [[package]] name = "mqtt" version = "0.1.0" dependencies = [ - "bytes", - "dotenvy", - "thiserror 1.0.68", - "tokio", - "tokio-rustls", - "webpki-roots", + "defmt 1.0.1", + "serde", ] [[package]] @@ -1895,23 +1734,13 @@ version = "0.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "43794a0ace135be66a25d3ae77d41b91615fb68ae937f904090203e81f755b65" -[[package]] -name = "nom" -version = "7.1.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d273983c5a657a70a3e8f2a01329822f3b8c8172b73826411a55751e404a0a4a" -dependencies = [ - "memchr", - "minimal-lexical", -] - [[package]] name = "nourl" version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b3c12edfda65fe16901d81d3bd93fd18ac07078b5007875a1c3b0d35f7725269" dependencies = [ - "defmt", + "defmt 0.3.100", ] [[package]] @@ -2171,16 +2000,6 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "925383efa346730478fb4838dbe9137d2a47675ad789c546d150a6e1dd4ab31c" -[[package]] -name = "prettyplease" -version = "0.2.20" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5f12335488a2f3b0a83b14edad48dca9879ce89b2edd10e80237e4e852dd645e" -dependencies = [ - "proc-macro2", - "syn 2.0.87", -] - [[package]] name = "primeorder" version = "0.13.6" @@ -2214,6 +2033,28 @@ dependencies = [ "version_check", ] +[[package]] +name = "proc-macro-error-attr2" +version = "2.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96de42df36bb9bba5542fe9f1a054b8cc87e172759a1868aa05c1f3acc89dfc5" +dependencies = [ + "proc-macro2", + "quote", +] + +[[package]] +name = "proc-macro-error2" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "11ec05c52be0a07b08061f7dd003e7d7092e0472bc731b4af7bb1ef876109802" +dependencies = [ + "proc-macro-error-attr2", + "proc-macro2", + "quote", + "syn 2.0.87", +] + [[package]] name = "proc-macro-utils" version = "0.10.0" @@ -2368,7 +2209,7 @@ checksum = "66584b58c3b0cd8cfea750408a932fe86df79ee05ccc87ee745a8e6be4a1abe0" dependencies = [ "base64", "buffered-io", - "defmt", + "defmt 0.3.100", "embedded-io", "embedded-io-async", "embedded-nal-async", @@ -2381,21 +2222,6 @@ dependencies = [ "rand_core", ] -[[package]] -name = "ring" -version = "0.17.8" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c17fa4cb658e3583423e915b9f3acc01cceaee1860e33d59ebae66adc3a2dc0d" -dependencies = [ - "cc", - "cfg-if", - "getrandom", - "libc", - "spin", - "untrusted", - "windows-sys", -] - [[package]] name = "rp-pac" version = "6.0.0" @@ -2430,7 +2256,7 @@ dependencies = [ "ruff_python_trivia 0.0.0 (git+https://github.com/astral-sh/ruff?rev=9bee8376a17401f9736b45fdefffb62edc2f1668)", "ruff_source_file 0.0.0 (git+https://github.com/astral-sh/ruff?rev=9bee8376a17401f9736b45fdefffb62edc2f1668)", "ruff_text_size 0.0.0 (git+https://github.com/astral-sh/ruff?rev=9bee8376a17401f9736b45fdefffb62edc2f1668)", - "rustc-hash 2.1.1", + "rustc-hash", "thiserror 2.0.12", ] @@ -2448,7 +2274,7 @@ dependencies = [ "ruff_python_trivia 0.0.0 (git+https://github.com/astral-sh/ruff)", "ruff_source_file 0.0.0 (git+https://github.com/astral-sh/ruff)", "ruff_text_size 0.0.0 (git+https://github.com/astral-sh/ruff)", - "rustc-hash 2.1.1", + "rustc-hash", "thiserror 2.0.12", ] @@ -2465,7 +2291,7 @@ dependencies = [ "ruff_python_ast 0.0.0 (git+https://github.com/astral-sh/ruff?rev=9bee8376a17401f9736b45fdefffb62edc2f1668)", "ruff_python_trivia 0.0.0 (git+https://github.com/astral-sh/ruff?rev=9bee8376a17401f9736b45fdefffb62edc2f1668)", "ruff_text_size 0.0.0 (git+https://github.com/astral-sh/ruff?rev=9bee8376a17401f9736b45fdefffb62edc2f1668)", - "rustc-hash 2.1.1", + "rustc-hash", "static_assertions", "unicode-ident", "unicode-normalization", @@ -2531,12 +2357,6 @@ version = "0.1.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "719b953e2095829ee67db738b3bfa9fa368c94900df327b3f07fe6e794d2fe1f" -[[package]] -name = "rustc-hash" -version = "1.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "08d43f7aa6b08d49f382cde6a7982047c3426db949b1424bc4b7ec9ae12c6ce2" - [[package]] name = "rustc-hash" version = "2.1.1" @@ -2561,52 +2381,6 @@ dependencies = [ "semver 1.0.23", ] -[[package]] -name = "rustix" -version = "0.38.34" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "70dc5ec042f7a43c4a73241207cecc9873a06d45debb38b329f8541d85c2730f" -dependencies = [ - "bitflags 2.9.1", - "errno", - "libc", - "linux-raw-sys", - "windows-sys", -] - -[[package]] -name = "rustls" -version = "0.23.12" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c58f8c84392efc0a126acce10fa59ff7b3d2ac06ab451a33f2741989b806b044" -dependencies = [ - "aws-lc-rs", - "log", - "once_cell", - "rustls-pki-types", - "rustls-webpki", - "subtle", - "zeroize", -] - -[[package]] -name = "rustls-pki-types" -version = "1.7.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "976295e77ce332211c0d24d92c0e83e50f5c5f046d11082cea19f3df13a3562d" - -[[package]] -name = "rustls-webpki" -version = "0.102.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8e6b52d4fda176fd835fdc55a835d4a89b8499cad995885a21149d5ad62f852e" -dependencies = [ - "aws-lc-rs", - "ring", - "rustls-pki-types", - "untrusted", -] - [[package]] name = "rustversion" version = "1.0.17" @@ -2661,24 +2435,36 @@ checksum = "388a1df253eca08550bef6c72392cfe7c30914bf41df5269b68cbd6ff8f570a3" [[package]] name = "serde" -version = "1.0.204" +version = "1.0.219" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bc76f558e0cbb2a839d37354c575f1dc3fdc6546b5be373ba43d95f231bf7c12" +checksum = "5f0e2c6ed6606019b4e29e69dbaba95b11854410e5347d525002456dbbb786b6" dependencies = [ "serde_derive", ] [[package]] name = "serde_derive" -version = "1.0.204" +version = "1.0.219" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e0cd7e117be63d3c3678776753929474f3b04a43a080c744d6b0ae2a8c28e222" +checksum = "5b0276cf7f2c73365f7157c8123c21cd9a50fbbd844757af28ca1f5925fc2a00" dependencies = [ "proc-macro2", "quote", "syn 2.0.87", ] +[[package]] +name = "serde_json" +version = "1.0.140" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "20068b6e96dc6c9bd23e01df8827e6c7e1f2fddd43c21810382803c136b99373" +dependencies = [ + "itoa", + "memchr", + "ryu", + "serde", +] + [[package]] name = "serialport" version = "4.4.0" @@ -2710,12 +2496,6 @@ dependencies = [ "digest", ] -[[package]] -name = "shlex" -version = "1.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" - [[package]] name = "signal-hook-registry" version = "1.4.2" @@ -2752,7 +2532,7 @@ dependencies = [ "bitflags 1.3.2", "byteorder", "cfg-if", - "defmt", + "defmt 0.3.100", "heapless 0.8.0", "managed", ] @@ -2975,17 +2755,6 @@ dependencies = [ "syn 2.0.87", ] -[[package]] -name = "tokio-rustls" -version = "0.26.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0c7bc40d0e5a97695bb96e27995cd3a08538541b0a846f65bba7a359f36700d4" -dependencies = [ - "rustls", - "rustls-pki-types", - "tokio", -] - [[package]] name = "typenum" version = "1.17.0" @@ -3066,12 +2835,6 @@ dependencies = [ "subtle", ] -[[package]] -name = "untrusted" -version = "0.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" - [[package]] name = "vcell" version = "0.1.3" @@ -3105,27 +2868,6 @@ version = "0.11.0+wasi-snapshot-preview1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9c8d87e72b64a3b4db28d11ce29237c246188f4f51057d65a7eab63b7987e423" -[[package]] -name = "webpki-roots" -version = "0.26.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bd7c23921eeb1713a4e851530e9b9756e4fb0e89978582942612524cf09f01cd" -dependencies = [ - "rustls-pki-types", -] - -[[package]] -name = "which" -version = "4.4.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "87ba24419a2078cd2b0f2ede2691b6c66d8e47836da3b6db8265ebad47afbfc7" -dependencies = [ - "either", - "home", - "once_cell", - "rustix", -] - [[package]] name = "winapi" version = "0.3.9" @@ -3235,17 +2977,3 @@ name = "zeroize" version = "1.8.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ced3678a2879b30306d323f4542626697a464a97c0a07c9aebf7ebca65cd4dde" -dependencies = [ - "zeroize_derive", -] - -[[package]] -name = "zeroize_derive" -version = "1.4.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ce36e65b0d2999d2aafac989fb249189a141aee1f53c612c1f37d72631959f69" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.87", -] diff --git a/Cargo.toml b/Cargo.toml index 3101d56..cf5266a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,3 +1,11 @@ [workspace] -members = ["debug-listener", "exponential_distribution", "fan-controller", "mqtt", ] -resolver = "2" +members = [ + "debug-listener", + "exponential_distribution", + "fan-controller", + "mqtt", +] +resolver = "3" + +[workspace.dependencies] +defmt = { version = "1.0.1" } diff --git a/fan-controller/Cargo.toml b/fan-controller/Cargo.toml index ae4d833..52a8f7f 100644 --- a/fan-controller/Cargo.toml +++ b/fan-controller/Cargo.toml @@ -9,7 +9,7 @@ cortex-m-rt = "0.7.3" crc = "3.2.1" cyw43 = { version = "0.1.0", features = ["defmt", "firmware-logs"] } cyw43-pio = { version = "0.1.0", features = ["defmt", "overclock"] } -defmt = "0.3.8" +defmt = { workspace = true } defmt-rtt = "0.4.1" embassy-executor = { version = "0.5.0", features = [ "task-arena-size-98304", @@ -40,6 +40,7 @@ embassy-time = { version = "0.3.1", features = ["defmt"] } embedded-io-async = "0.6.1" embedded-nal-async = "0.7.1" heapless = "0.8.0" +mqtt = { version = "0.1.0", path = "../mqtt", features = ["defmt"] } nb = "1.1.0" panic-probe = "0.3.2" portable-atomic = { version = "1.7", features = ["critical-section"] } @@ -52,6 +53,9 @@ debug = 2 [build-dependencies] dotenvy = "0.15.7" +mqtt = { version = "0.1.0", path = "../mqtt", features = ["serde"] } ruff_python_ast = { git = "https://github.com/astral-sh/ruff", version = "0.0.0" } ruff_python_parser = { git = "https://github.com/astral-sh/ruff", rev = "9bee8376a17401f9736b45fdefffb62edc2f1668", version = "0.0.0" } +serde = { version = "1.0.219", features = ["derive"] } +serde_json = "1.0.140" thiserror = "2.0.12" diff --git a/fan-controller/build.rs b/fan-controller/build.rs index 9a0a666..d5fe60a 100644 --- a/fan-controller/build.rs +++ b/fan-controller/build.rs @@ -15,7 +15,119 @@ use std::io::Write; use std::path::PathBuf; use std::rc::Rc; +use mqtt::QualityOfService; use ruff_python_ast::{DictItem, Expr, Stmt}; +use serde::Serialize; + +#[derive(Serialize, Clone, Debug)] +#[serde(untagged)] +enum ListOrString { + List(Vec), + String(&'static str), +} + +/// Information about the device this fan is a part of to tie it into the device registry. Only works when unique_id is set. At least one of identifiers or connections must be present to identify the device. +#[derive(Serialize, Default)] +struct Device { + /// A list of IDs that uniquely identify the device. For example a serial number. + #[serde(rename = "ids")] + #[serde(skip_serializing_if = "Option::is_none")] + identifiers: Option, + /// The name of the device. + #[serde(skip_serializing_if = "Option::is_none")] + name: Option<&'static str>, + /// The model of the device. + #[serde(rename = "mdl")] + #[serde(skip_serializing_if = "Option::is_none")] + model: Option<&'static str>, + /// The manufacturer of the device. + #[serde(rename = "mf")] + #[serde(skip_serializing_if = "Option::is_none")] + manufacturer: Option<&'static str>, + /// The hardware version of the device. + #[serde(rename = "hw")] + #[serde(skip_serializing_if = "Option::is_none")] + hardware_version: Option<&'static str>, + /// The firmware version of the device. + #[serde(rename = "sw")] + #[serde(skip_serializing_if = "Option::is_none")] + software_version: Option<&'static str>, +} + +#[derive(Serialize, Default)] +struct Origin { + /// The name of the application that is the origin of the discovered MQTT item. (Required) + name: &'static str, + /// Software version of the application that supplies the discovered MQTT item. + #[serde(rename = "sw")] + #[serde(skip_serializing_if = "Option::is_none")] + software_version: Option<&'static str>, + /// Support URL of the application that supplies the discovered MQTT item. + #[serde(rename = "url")] + #[serde(skip_serializing_if = "Option::is_none")] + support_url: Option<&'static str>, +} + +/// Internally tagged by the required `platform` (`p`) field +#[derive(Serialize)] +#[serde(tag = "p")] +enum Component { + Fan { + /// The name of the fan. Can be set to null if only the device name is relevant. + name: Option<&'static str>, + /// An ID that uniquely identifies this fan. If two fans have the same unique ID, Home Assistant will raise an exception. Required when used with device-based discovery. + #[serde(rename = "uniq_id")] + #[serde(skip_serializing_if = "Option::is_none")] + unique_id: Option<&'static str>, + /// The MQTT topic subscribed to receive state updates. A “None” payload resets to an unknown state. An empty payload is ignored. By default, valid state payloads are OFF and ON. The accepted payloads can be overridden with the payload_off and payload_on config options. + #[serde(rename = "stat_t")] + #[serde(skip_serializing_if = "Option::is_none")] + state_topic: Option<&'static str>, + + /// The MQTT topic to publish commands to change the fan state. + #[serde(rename = "cmd_t")] + command_topic: &'static str, + /// The MQTT topic subscribed to receive fan speed based on percentage. + #[serde(rename = "pct_stat_t")] + percentage_state_topic: Option<&'static str>, + /// The MQTT topic to publish commands to change the fan speed state based on a percentage. + #[serde(rename = "pct_cmd_t")] + percentage_command_topic: Option<&'static str>, + /// The maximum of numeric output range (representing 100 %). The percentage_step is defined by 100 / the number of speeds within the speed range. + /// Default: 100 + #[serde(rename = "spd_rng_max")] + speed_range_max: Option, + }, +} + +/// Home Assistant MQTT device-based Discovery Payload +/// This is for the multi [device discovery payload](https://www.home-assistant.io/integrations/mqtt/#device-discovery-payload). +/// It requires +/// - device +/// - origin +#[derive(Serialize, Default)] +struct DiscoveryPayload { + #[serde(rename = "dev")] + device: Device, + #[serde(rename = "o")] + origin: Origin, + #[serde(rename = "cmps")] + components: HashMap, + #[serde(rename = "qos")] + #[serde(skip_serializing_if = "Option::is_none")] + quality_of_service: Option, + #[serde(rename = "stat_t")] + #[serde(skip_serializing_if = "Option::is_none")] + state_topic: Option<&'static str>, + #[serde(rename = "cmd_t")] + #[serde(skip_serializing_if = "Option::is_none")] + command_topic: Option<&'static str>, + /// The encoding of the payloads received and published messages. Set to "" to disable decoding of incoming payload. + /// Default is "utf-8" + #[serde(rename = "e")] + #[serde(skip_serializing_if = "Option::is_none")] + encoding: Option, +} #[derive(Debug, thiserror::Error)] enum BuildError { @@ -154,6 +266,7 @@ fn extract_abreviations() -> HashMap, Rc> { /// Overengineering saving a couple of bytes from a JSON string. /// Minimizes payload with abbreviations and removing whitespace. /// Abbreviations are loaded from the official home assistant repository. +/// TODO: device and origin abbreviations fn setup_discovery_payload() -> Result<(), DiscoveryPayloadError> { const PATH_DISCOVER_JSON: &str = "discovery_payload.json"; //TODO validate discovery payload @@ -217,9 +330,64 @@ fn setup_discovery_payload() -> Result<(), DiscoveryPayloadError> { Ok(()) } +fn set_discovery_payload() { + // No way to const a HashMap + let payload = DiscoveryPayload { + device: Device { + identifiers: Some(ListOrString::String("fancontroller-device")), + name: Some("Fan Controller"), + model: Some("Raspberry Pi Pico W 1"), + manufacturer: Some("claas.dev"), + hardware_version: Some("1.0"), + software_version: Some(env!("CARGO_PKG_VERSION")), + }, + origin: Origin { + name: "fan-controller", + software_version: Some(env!("CARGO_PKG_VERSION")), + support_url: Some("https://github.com/SantaClaas/embedded-fan-control"), + }, + components: HashMap::from([ + // Fan 1 + ( + "fan-1".to_string(), + Component::Fan { + name: Some("Fan 1"), + unique_id: Some("fancontroller/fan-1"), + state_topic: Some("fancontroller/fan-1/on/state"), + command_topic: "fancontroller/fan-1/on/set", + percentage_state_topic: Some("fancontroller/fan-1/speed/percentage_state"), + percentage_command_topic: Some("fancontroller/fan-1/speed/percentage"), + speed_range_max: Some(32_000), + }, + ), + // Fan 2 + ( + "fan-2".to_string(), + Component::Fan { + name: Some("Fan 2"), + unique_id: Some("fancontroller/fan-2"), + state_topic: Some("fancontroller/fan-2/on/state"), + command_topic: "fancontroller/fan-2/on/set", + percentage_state_topic: Some("fancontroller/fan-2/speed/percentage_state"), + percentage_command_topic: Some("fancontroller/fan-2/speed/percentage"), + speed_range_max: Some(32_000), + }, + ), + ]), + quality_of_service: None, + state_topic: Some("fancontroller/on/state"), + command_topic: Some("fancontroller/on/set"), + encoding: None, + }; + + let payload = serde_json::to_string(&payload).unwrap(); + println!("cargo:rustc-env=FAN_CONTROLLER_DISCOVERY_PAYLOAD={payload}",); +} + fn main() -> Result<(), BuildError> { ensure_memory_x_file()?; setup_configuration()?; - setup_discovery_payload()?; + // setup_discovery_payload()?; + set_discovery_payload(); Ok(()) } diff --git a/fan-controller/discovery_payload.json b/fan-controller/discovery_payload.json index 56f74bf..a64d2a8 100644 --- a/fan-controller/discovery_payload.json +++ b/fan-controller/discovery_payload.json @@ -1,4 +1,10 @@ { + "device": { + "ids": "fancontroller-device", + "name": "Fan Controller", + "manufacturer": "claas.dev", + "model": "Raspberry Pi Pico W 1" + }, "name": "Fans", "unique_id": "fancontroller", "stat_t": "fancontroller/on/state", @@ -6,9 +12,5 @@ "pct_stat_t": "fancontroller/speed/percentage_state", "pct_cmd_t": "fancontroller/speed/percentage", "spd_rng_max": 32000, - "dev": { - "ids": "fancontroller-device", - "name": "Fan Controller", - "model": "Raspberry Pi Pico W 1" - } + "origin": {} } diff --git a/fan-controller/src/main.rs b/fan-controller/src/main.rs index 7aab5b4..be51b23 100644 --- a/fan-controller/src/main.rs +++ b/fan-controller/src/main.rs @@ -63,8 +63,8 @@ use crate::mqtt::packet::subscribe_acknowledgement::SubscribeAcknowledgement; use crate::mqtt::packet::{connect, publish, subscribe}; use crate::mqtt::packet::{get_parts, FromPublish, FromSubscribeAcknowledgement}; use crate::mqtt::task::send; -use crate::mqtt::QualityOfService; use crate::mqtt::TryEncode; +use ::mqtt::QualityOfService; mod async_callback; mod configuration; diff --git a/fan-controller/src/mqtt/mod.rs b/fan-controller/src/mqtt/mod.rs index 0ed9172..18978cc 100644 --- a/fan-controller/src/mqtt/mod.rs +++ b/fan-controller/src/mqtt/mod.rs @@ -42,26 +42,6 @@ pub(super) enum ConnectErrorReasonCode { #[derive(Debug, Clone, Format)] pub struct UnknownConnectErrorReasonCode(u8); -#[derive(Format)] -pub(super) enum QualityOfService { - /// At most once delivery or 0 - AtMostOnceDelivery = 0x00, - /// At least once delivery or 1 - AtLeastOnceDelivery = 0x01, - /// Exactly once delivery or 2 - ExactlyOnceDelivery = 0x02, -} - -impl QualityOfService { - const fn to_byte(&self) -> u8 { - match self { - QualityOfService::AtMostOnceDelivery => 0, - QualityOfService::AtLeastOnceDelivery => 1, - QualityOfService::ExactlyOnceDelivery => 2, - } - } -} - pub(crate) trait Encode { fn encode(&self, buffer: &mut [u8], offset: &mut usize); } diff --git a/fan-controller/src/mqtt/packet/connect_acknowledgement.rs b/fan-controller/src/mqtt/packet/connect_acknowledgement.rs index 553e74e..719098a 100644 --- a/fan-controller/src/mqtt/packet/connect_acknowledgement.rs +++ b/fan-controller/src/mqtt/packet/connect_acknowledgement.rs @@ -1,7 +1,6 @@ -use crate::mqtt::{ - variable_byte_integer, ConnectErrorReasonCode, QualityOfService, UnknownConnectErrorReasonCode, -}; +use crate::mqtt::{variable_byte_integer, ConnectErrorReasonCode, UnknownConnectErrorReasonCode}; use crate::TryDecode; +use ::mqtt::QualityOfService; use defmt::{info, Format}; // pub(crate) struct Properties { diff --git a/fan-controller/src/mqtt/packet/subscribe.rs b/fan-controller/src/mqtt/packet/subscribe.rs index 2088f72..a96c446 100644 --- a/fan-controller/src/mqtt/packet/subscribe.rs +++ b/fan-controller/src/mqtt/packet/subscribe.rs @@ -1,10 +1,9 @@ +use crate::mqtt::variable_byte_integer; +use crate::mqtt::TryEncode; +use ::mqtt::QualityOfService; use core::num::NonZeroU16; - use defmt::{info, Format}; -use crate::mqtt::TryEncode; -use crate::mqtt::{variable_byte_integer, QualityOfService}; - #[derive(Debug)] pub(crate) struct Options(u8); diff --git a/fan-controller/src/mqtt/packet/subscribe_acknowledgement.rs b/fan-controller/src/mqtt/packet/subscribe_acknowledgement.rs index 1418d5a..38aa66c 100644 --- a/fan-controller/src/mqtt/packet/subscribe_acknowledgement.rs +++ b/fan-controller/src/mqtt/packet/subscribe_acknowledgement.rs @@ -1,4 +1,5 @@ -use crate::mqtt::{variable_byte_integer, QualityOfService}; +use crate::mqtt::variable_byte_integer; +use ::mqtt::QualityOfService; use defmt::Format; #[derive(Debug, Clone, Format)] diff --git a/mqtt/Cargo.toml b/mqtt/Cargo.toml index 17798d5..76ca8cc 100644 --- a/mqtt/Cargo.toml +++ b/mqtt/Cargo.toml @@ -1,12 +1,11 @@ [package] name = "mqtt" version = "0.1.0" -edition = "2021" +edition = "2024" [dependencies] -thiserror = "1.0.63" -tokio = { version = "1.39.1", features = ["full"] } -tokio-rustls = "0.26.0" -webpki-roots = "0.26.3" -bytes = "1.6.1" -dotenvy = "0.15.7" \ No newline at end of file +defmt = { workspace = true, optional = true } +serde = { version = "1.0.219", features = ["derive"], optional = true } +[features] +defmt = ["dep:defmt"] +serde = ["dep:serde"] diff --git a/mqtt/src/lib.rs b/mqtt/src/lib.rs new file mode 100644 index 0000000..d48305a --- /dev/null +++ b/mqtt/src/lib.rs @@ -0,0 +1,22 @@ +#![no_std] +#[cfg_attr(feature = "defmt", derive(defmt::Format))] +#[cfg_attr(feature = "serde", derive(serde::Serialize))] +#[repr(u8)] +pub enum QualityOfService { + /// At most once delivery or 0 + AtMostOnceDelivery = 0x00, + /// At least once delivery or 1 + AtLeastOnceDelivery = 0x01, + /// Exactly once delivery or 2 + ExactlyOnceDelivery = 0x02, +} + +impl QualityOfService { + pub const fn to_byte(&self) -> u8 { + match self { + QualityOfService::AtMostOnceDelivery => 0, + QualityOfService::AtLeastOnceDelivery => 1, + QualityOfService::ExactlyOnceDelivery => 2, + } + } +} diff --git a/mqtt/src/main.rs b/mqtt/src/main.rs deleted file mode 100644 index 8d23daa..0000000 --- a/mqtt/src/main.rs +++ /dev/null @@ -1,850 +0,0 @@ -use std::env; -use std::net::{AddrParseError, ToSocketAddrs}; -use std::str::Utf8Error; -use std::sync::Arc; -use std::time::Duration; - -use tokio::io::{AsyncReadExt, AsyncWriteExt}; -use tokio::net::TcpStream; -use tokio::sync::{mpsc, oneshot}; -use tokio::time::error::Elapsed; -use tokio::time::Interval; -use tokio_rustls::rustls::{ClientConfig, RootCertStore}; -use tokio_rustls::rustls::pki_types::{InvalidDnsNameError, ServerName}; -use tokio_rustls::TlsConnector; - -use packet::{ConnectErrorReasonCode, ConnectReasonCode, UnknownConnectReasonCode}; -use variable_byte_integer::{DecodeVariableByteIntegerError, VariableByteInteger}; - -use crate::packet::{QualityOfService, RetainHandling, Subscription, SubscriptionOptions}; - -mod packet; -mod variable_byte_integer; - -/// We send a ping request every 60 seconds to keep the connection alive -/// The server will wait 1.5 times the keep alive time before disconnecting -const KEEP_ALIVE_SECONDS: u16 = 60; - -#[derive(Debug, thiserror::Error)] -enum AppError { - #[error("Failed to create address")] - CreateAddressError(#[from] AddrParseError), - #[error("Could not create server name")] - CreateServerNameError(#[from] InvalidDnsNameError), - #[error("Failed to connect to broker")] - ConnectError(#[from] std::io::Error), - #[error("Failed to load .env file and environment variables")] - DotenvError(#[from] dotenvy::Error), - #[error("Error reading environment variable")] - EnvVarError(#[from] env::VarError), - - #[error("Port environment variable is not a valid number")] - PortEnvVarError(#[from] std::num::ParseIntError), -} - -async fn set_up_tls_connection( - broker_address: String, - port: u16, -) -> Result { - let mut root_certificate_store = RootCertStore::empty(); - root_certificate_store.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned()); - // root_certificate_store.add(CertificateDer::from(certificate)).unwrap(); - let configuration = ClientConfig::builder() - .with_root_certificates(root_certificate_store) - .with_no_client_auth(); - let connector = TlsConnector::from(Arc::new(configuration)); - - let address = (broker_address.as_ref(), port) - .to_socket_addrs()? - .next() - .unwrap(); - - let server_name = ServerName::try_from(broker_address)?; - - let stream = TcpStream::connect(&address).await?; - - let stream = connector.connect(server_name, stream).await?; - - Ok(stream) -} - -/// Errors while reading the Connect Acknowledgement packet (CONNACK) -#[derive(Debug, thiserror::Error)] -enum ConnectAcknowledgementError { - #[error("Error while reading from the stream: {0}")] - ReadError(#[from] std::io::Error), - /// The server sent a different packet type than connect acknowledgement (2) - #[error("Unexpected packet type: {0:#04x}")] - UnexpectedPacketType(u8), - #[error("Invalid remaining length: {0}")] - InvalidRemainingLength(DecodeVariableByteIntegerError), - #[error("Invalid reason code: {0}")] - InvalidReasonCode(UnknownConnectReasonCode), - #[error("Error reason code: {0:?}")] - ErrorReasonCode(ConnectErrorReasonCode), -} - -#[derive(Debug)] -enum ConnectError { - SendError(std::io::Error), - FlushError(std::io::Error), - AcknowledgementError(ConnectAcknowledgementError), -} - -/// Errors while reading the Subscribe Acknowledgement packet (SUBACK) -#[derive(Debug, thiserror::Error)] -enum SubscribeAcknowledgementError { - #[error("Error while reading from the stream: {0}")] - ReadError(#[from] std::io::Error), - /// The server sent a different packet type than connect acknowledgement (2) - #[error("Unexpected packet type: {0:#04x}")] - UnexpectedPacketType(u8), - #[error("Invalid remaining length: {0}")] - InvalidRemainingLength(DecodeVariableByteIntegerError), - #[error("Invalid properties length: {0}")] - InvalidPropertiesLength(DecodeVariableByteIntegerError), - #[error("Invalid reason code length: expected {expected} topic(s), got {actual}")] - InvalidReasonCodeLength { expected: usize, actual: usize }, -} - -#[derive(Debug)] -enum SubscribeError { - NotConnected, - SendError(std::io::Error), - FlushError(std::io::Error), - AcknowledgementError(SubscribeAcknowledgementError), -} - -enum MqttActorMessage { - Connect { - client_identifier: Arc, - username: Arc, - password: Arc<[u8]>, - responder: oneshot::Sender>, - }, - Publish { - topic: Arc, - payload: Arc<[u8]>, - responder: oneshot::Sender>, - }, - Subscribe { - subscriptions: Arc<[Subscription]>, - responder: oneshot::Sender>, - }, -} - -struct MqttActor { - receiver: mpsc::Receiver, - stream: TcpStream, - send_publish: mpsc::Sender<()>, - // Use ticker in embassy-time - //TODO put this in a "Connected" variant as this has no effect if the connection is not established - /// The last time we send a packet to the broker. Used with [crate::KEEP_ALIVE_SECONDS] - /// to determine when to send the next ping request to keep the connection alive - ping_interval: Interval, - is_connected: bool, -} - -async fn read_connect_acknowledgement( - stream: &mut TcpStream, -) -> Result<(), ConnectAcknowledgementError> { - // Fixed header - let packet_type_and_flags = stream.read_u8().await?; - if (packet_type_and_flags >> 4) != 2 { - return Err(ConnectAcknowledgementError::UnexpectedPacketType( - packet_type_and_flags >> 4, - )); - } - - let (remaining_length, _remaining_length_length) = VariableByteInteger::decode(stream) - .await - .map_err(ConnectAcknowledgementError::InvalidRemainingLength)?; - - let mut buffer = Vec::with_capacity(remaining_length); - // Could warn if we read less or more than remaining_length - let _bytes_read = stream.read_buf(&mut buffer).await?; - - // CONNACK Variable header - // Acknowledge flags - // Bit 7-1 are reserved and must be set to 0, although we don't validate this - // Bit 0 is the Session Present Flag - // let flags = stream.read_u8().await?; - // let is_session_present = (flags & 0b0000_0001) != 0; - - let connect_reason_code = buffer[1]; - let code = ConnectReasonCode::try_from(connect_reason_code) - .map_err(ConnectAcknowledgementError::InvalidReasonCode)?; - - if let ConnectReasonCode::Error(error_code) = code { - return Err(ConnectAcknowledgementError::ErrorReasonCode(error_code)); - } - - // Ignore properties for now - Ok(()) -} - -async fn send_connect( - actor: &mut MqttActor, - client_identifier: Arc, - username: Arc, - password: Arc<[u8]>, -) -> Result<(), ConnectError> { - let packet = packet::create_connect::( - client_identifier.as_ref(), - username.as_ref(), - password.as_ref(), - ); - - actor - .stream - .write_all(&packet) - .await - .map_err(ConnectError::SendError)?; - - actor - .stream - .flush() - .await - .map_err(ConnectError::FlushError)?; - - actor.ping_interval.reset(); - - // We read the acknowledgement "synchronous" as we can't receive or send other packets before the connection is established - // Response for connection has to come in next - read_connect_acknowledgement(&mut actor.stream) - .await - .map_err(ConnectError::AcknowledgementError) -} - -async fn read_subscribe_acknowledgement( - stream: &mut TcpStream, - subscriptions: Arc<[Subscription]>, -) -> Result<(), SubscribeAcknowledgementError> { - // Fixed header - let packet_type_and_flags = stream.read_u8().await?; - if (packet_type_and_flags >> 4) != 9 { - return Err(SubscribeAcknowledgementError::UnexpectedPacketType( - packet_type_and_flags >> 4, - )); - } - - let (mut remaining_length, _remaining_length_length) = VariableByteInteger::decode(stream) - .await - .map_err(SubscribeAcknowledgementError::InvalidRemainingLength)?; - - let mut buffer = Vec::with_capacity(remaining_length); - // Could warn if we read less or more than remaining_length - let _bytes_read = stream.read_buf(&mut buffer).await?; - - // Variable header - let packet_identifier: u16 = ((buffer[0] as u16) << 8) | buffer[1] as u16; - println!("Packet identifier {packet_identifier}"); - remaining_length -= 2; - println!("Remaining length {remaining_length}"); - - // Properties - let (properties_length, properties_length_length) = - VariableByteInteger::decode_2(buffer.as_ref()) - .map_err(SubscribeAcknowledgementError::InvalidPropertiesLength)?; - println!("Properties length {properties_length}"); - remaining_length -= properties_length_length; - println!("Remaining length {remaining_length}"); - //TODO read properties - - // Payload - // Reason code for each subscribed topic in the same order - let mut topic_index = 0; - if remaining_length != subscriptions.len() { - return Err(SubscribeAcknowledgementError::InvalidReasonCodeLength { - expected: subscriptions.len(), - actual: remaining_length, - }); - } - - // Index should be aligned as we checked length before - while remaining_length > 0 { - let reason_code = buffer[2 + properties_length_length + topic_index]; - remaining_length -= 1; - - let topic = &subscriptions[topic_index].topic_filter; - topic_index += 1; - println!("Reason code for topic {topic}: {reason_code:#04x}"); - println!("Remaining length {remaining_length}"); - } - - Ok(()) -} - -/// Errors while reading the Subscribe Acknowledgement packet (SUBACK) -#[derive(Debug, thiserror::Error)] -enum ReadPublishError { - #[error("Error while reading from the stream: {0}")] - ReadError(#[from] std::io::Error), - /// The server sent a different packet type than connect acknowledgement (2) - #[error("Unexpected packet type: {0:#04x}")] - UnexpectedPacketType(u8), - #[error("Invalid remaining length: {0}")] - InvalidRemainingLength(DecodeVariableByteIntegerError), - #[error("Topic name is not valid UTF-8: {0}")] - InvalidTopicName(#[from] Utf8Error), - #[error("Zero length topic name is a protocol error")] - ZeroLengthTopicName, - #[error("Invalid properties length: {0}")] - InvalidPropertiesLength(DecodeVariableByteIntegerError), - #[error("Invalid reason code length: expected {expected} topic(s), got {actual}")] - InvalidReasonCodeLength { expected: usize, actual: usize }, -} - -struct Publish { - topic_name: Arc, - payload: Arc<[u8]>, -} - -async fn receive_publish(stream: &mut TcpStream) -> Result { - let type_and_flags = stream.read_u8().await?; - - if (type_and_flags >> 4) != 3 { - return Err(ReadPublishError::UnexpectedPacketType(type_and_flags >> 4)); - } - - // Flags matter here - // let is_re_delivery = (type_and_flags & 0b0000_1000) != 0; - let quality_of_service_level = (type_and_flags & 0b0000_0110) >> 1; - if quality_of_service_level > 0 { - // This changes the layout of the packet and adds a packet identifier which we don't expect right now - unimplemented!("Quality of service level other than 0 is not implemented yet") - } - - // Ignore retain as it should only matter for packets send to the server - - let (remaining_length, _remaining_length_length) = VariableByteInteger::decode(stream) - .await - .map_err(ReadPublishError::InvalidRemainingLength)?; - - let mut buffer = Vec::with_capacity(remaining_length); - let _bytes_read = stream.read_buf(&mut buffer).await?; - //TODO check if remaining length matches bytes read - - // Variable header - let topic_length = ((buffer[0] as u16) << 8) | buffer[1] as u16; - if topic_length == 0 { - return Err(ReadPublishError::ZeroLengthTopicName); - } - - let topic_name = std::str::from_utf8(&buffer[2..2 + topic_length as usize])?; - //TODO validate topic name does not contain MQTT wildcard characters - - //TODO read packet identifier if QoS > 0 - let (property_length, property_length_length) = - VariableByteInteger::decode_2(&buffer[2 + topic_length as usize..]) - .map_err(ReadPublishError::InvalidPropertiesLength)?; - - // Ignore properties for now - - // Payload - let variable_header_length = - 2 + topic_length as usize + property_length_length + property_length; - let payload_length = remaining_length - variable_header_length; - //TODO validate there is enough space left in the buffer - let payload = &buffer[variable_header_length..variable_header_length + payload_length]; - - Ok(Publish { - topic_name: topic_name.into(), - payload: payload.into(), - }) -} - -#[derive(Debug)] -enum PublishError { - NotConnected, - WriteError(std::io::Error), - FlushError(std::io::Error), -} - -async fn send_publish( - actor: &mut MqttActor, - topic: Arc, - payload: Arc<[u8]>, -) -> Result<(), PublishError> { - if !actor.is_connected { - return Err(PublishError::NotConnected); - } - - let packet = packet::create_publish(topic.as_ref(), payload.as_ref()); - - // Fire and forget with quality of service 0, but we can at least confirm locally if the packet was sent - actor - .stream - .write_all(&packet) - .await - .map_err(PublishError::WriteError)?; - actor - .stream - .flush() - .await - .map_err(PublishError::FlushError)?; - - actor.ping_interval.reset(); - - // We don't wait for the PUBACK packet if there is no Quality of Service (QoS) configured - // If there was QoS, then we would need to assign a packet identifier and store the channel for as long as we wait for the PUBACK - Ok(()) -} - -async fn send_subscribe( - actor: &mut MqttActor, - subscriptions: Arc<[Subscription]>, -) -> Result<(), SubscribeError> { - if !actor.is_connected { - return Err(SubscribeError::NotConnected); - } - - //TODO manage available identifiers - let packet = packet::create_subscribe(9, subscriptions.as_ref()); - - actor - .stream - .write_all(&packet) - .await - .map_err(SubscribeError::SendError)?; - actor - .stream - .flush() - .await - .map_err(SubscribeError::FlushError)?; - - actor.ping_interval.reset(); - // We wait for the subscribe acknowledgement, but we should probably do that "asynchronously" - read_subscribe_acknowledgement(&mut actor.stream, subscriptions) - .await - .map_err(SubscribeError::AcknowledgementError) -} - -async fn process_message(actor: &mut MqttActor, message: MqttActorMessage) { - //TODO handle case when connection is closed or lost and might want to reconnect - match message { - MqttActorMessage::Connect { - client_identifier, - username, - password, - responder, - } => { - let result = send_connect(actor, client_identifier, username, password).await; - if result.is_ok() { - actor.is_connected = true; - } - // Ignore error if they cancelled waiting for the response - let _ = responder.send(result); - } - MqttActorMessage::Publish { - topic, - payload, - responder, - } => { - let result = send_publish(actor, topic, payload).await; - - // Ignore error if they cancelled waiting for the response - let _ = responder.send(result); - } - MqttActorMessage::Subscribe { - subscriptions, - responder, - } => { - let result = send_subscribe(actor, subscriptions).await; - - // Ignore error if they cancelled waiting for the response - let _ = responder.send(result); - } - } -} - -async fn process_publish(actor: &mut MqttActor, publish: Publish) { - println!("Received publish"); - println!("Topic: {}", publish.topic_name); - println!("Payload: {:?}", publish.payload); - - match publish.topic_name.as_ref() { - "testfan/speed/percentage" => { - let payload = match std::str::from_utf8(&publish.payload) { - Ok(payload) => payload, - Err(error) => { - eprintln!( - "Expected percentage_command_topic payload (speed percentage) to be a valid UTF-8 string with a number: {}", - error - ); - return; - } - }; - - let set_point = payload.parse::(); - - match set_point { - Ok(set_point) if set_point > 64000 => { - eprintln!("Received higher speed set point than configured by this device: {set_point}"); - } - Ok(set_point) => { - println!("Set point: {}", set_point); - } - Err(error) => { - eprintln!("Expected speed percentage to be a number: {}", error); - } - } - - //TODO implement debounce. Gather all update requests for a while and only send the latest if there hasn't been any in specific time window. This should avoid overburdening the modbus channel and avoid weird race conditions - // Mock sending update to the device - tokio::time::sleep(Duration::from_secs(1)).await; - - // Send update through publish to percentage_state_topic - let result = crate::send_publish( - actor, - "testfan/speed/percentage_state".into(), - publish.payload, - ) - .await; - //TODO error handling: sink the result in some log or something - println!("Published result: {:?}", result); - } - "testfan/on/set" => { - let payload = match std::str::from_utf8(&publish.payload) { - Ok(payload) => payload, - Err(error) => { - eprintln!( - "Expected comman_topic payload to be a valid UTF-8 string with either \"ON\" or \"OFF\": {}", - error - ); - return; - } - }; - - match payload { - "ON" => { - println!("Turning on"); - } - "OFF" => { - println!("Turning off"); - } - unknown => { - eprintln!( - "Expected either \"ON\" or \"OFF\" but received: {}", - unknown - ); - } - } - - //TODO implement debounce. Gather all update requests for a while and only send the latest if there hasn't been any in specific time window. This should avoid overburdening the modbus channel and avoid weird race conditions - // Mock sending update to the device - tokio::time::sleep(Duration::from_secs(1)).await; - // We just echo the state send to us - let result = send_publish(actor, "testfan/on/state".into(), publish.payload).await; - //TODO error handling: sink the result in some log or something - println!("Published result: {:?}", result); - } - unexpected => { - eprintln!( - "Unexpected topic: {:?} with payload: {:?}", - unexpected, publish.payload - ); - } - } -} - -#[derive(Debug)] -enum PingRequestError { - WriteError(std::io::Error), - FlushError(std::io::Error), - TimeoutError(Elapsed), - ReadResponseError(std::io::Error), - UnexpectedPacketType(u8), -} - -async fn send_ping_request(actor: &mut MqttActor) -> Result<(), PingRequestError> { - println!("Sending ping request"); - // Probably the simplest of them all - actor - .stream - .write_all(&[12 << 4, 0]) - .await - .map_err(PingRequestError::WriteError)?; - actor - .stream - .flush() - .await - .map_err(PingRequestError::FlushError)?; - - // Wait for the ping response - let read = actor.stream.read_u16(); - - // Specification only says "reasonable amount of time". How long is reasonable? - let fixed_header = tokio::time::timeout(Duration::from_secs(30), read) - .await - .map_err(PingRequestError::TimeoutError)? - .map_err(PingRequestError::ReadResponseError)?; - - let packet_type = fixed_header >> 12; - if packet_type != 13 { - return Err(PingRequestError::UnexpectedPacketType(packet_type as u8)); - } - - Ok(()) -} - -async fn process_ping(actor: &mut MqttActor) { - let result = send_ping_request(actor).await; - - println!("Ping result: {:?}", result); - - let Err(error) = result else { - return; - }; - - match error { - PingRequestError::WriteError(_) - | PingRequestError::FlushError(_) - | PingRequestError::TimeoutError(_) => {} - PingRequestError::ReadResponseError(_) => { - actor.is_connected = false; - println!("Disconnected due to ping response read error"); - } - PingRequestError::UnexpectedPacketType(packet_type) => { - println!("Unexpected packet type: {packet_type:#04x}"); - } - } -} - -async fn run_actor(mut actor: MqttActor) { - loop { - tokio::select! { - Some(message) = actor.receiver.recv() => process_message(&mut actor, message).await, - Ok(publish) = receive_publish(&mut actor.stream) => process_publish(&mut actor, publish).await, - _ = actor.ping_interval.tick(), if actor.is_connected => process_ping(&mut actor).await, - else => break, - } - } - - println!("Actor done"); -} - -struct MqttActorHandle { - sender: mpsc::Sender, -} - -#[derive(Debug)] -enum HandleError { - SendError(SE), - ReceiveError(RE), - ActorError(AE), -} - -impl MqttActorHandle { - fn new(stream: TcpStream, send_publish: mpsc::Sender<()>) -> Self { - let (sender, receiver) = mpsc::channel(8); - let actor = MqttActor { - stream, - receiver, - send_publish, - ping_interval: tokio::time::interval(Duration::from_secs(KEEP_ALIVE_SECONDS as u64)), - is_connected: false, - }; - - tokio::spawn(run_actor(actor)); - - Self { sender } - } - - async fn connect( - &self, - client_identifier: Arc, - username: Arc, - password: Arc<[u8]>, - ) -> Result<(), HandleError> { - let (sender, receiver) = oneshot::channel(); - - type Error = HandleError< - mpsc::error::SendError, - oneshot::error::RecvError, - ConnectError, - >; - - self.sender - .send(MqttActorMessage::Connect { - client_identifier, - username, - password, - responder: sender, - }) - .await - .map_err(Error::SendError)?; - - receiver - .await - .map_err(Error::ReceiveError)? - .map_err(Error::ActorError) - } - - async fn publish( - &self, - topic: Arc, - payload: Arc<[u8]>, - ) -> Result<(), HandleError> { - type Error = HandleError< - mpsc::error::SendError, - oneshot::error::RecvError, - PublishError, - >; - let (sender, receiver) = oneshot::channel(); - self.sender - .send(MqttActorMessage::Publish { - topic, - payload, - responder: sender, - }) - .await - .map_err(Error::SendError)?; - - receiver - .await - .map_err(Error::ReceiveError)? - .map_err(Error::ActorError) - } - - async fn subscribe( - &self, - subscriptions: Arc<[Subscription]>, - ) -> Result<(), HandleError> - { - let (sender, receiver) = oneshot::channel(); - - type Error = HandleError< - mpsc::error::SendError, - oneshot::error::RecvError, - SubscribeError, - >; - - self.sender - .send(MqttActorMessage::Subscribe { - subscriptions, - responder: sender, - }) - .await - .map_err(Error::SendError)?; - - receiver - .await - .map_err(Error::ReceiveError)? - .map_err(Error::ActorError) - } -} - -async fn set_up_tcp_connection( - broker_address: String, - broker_port: u16, -) -> Result { - let address = (broker_address.as_ref(), broker_port) - .to_socket_addrs()? - .next() - .unwrap(); - - let stream = TcpStream::connect(&address).await?; - - Ok(stream) -} - -#[tokio::main] -async fn main() -> Result<(), AppError> { - // Set up .env environment variables - dotenvy::dotenv()?; - let broker_address = env::var("MQTT_BROKER_ADDRESS")?; - let broker_port: u16 = env::var("MQTT_BROKER_PORT")?.parse()?; - let username = env::var("MQTT_BROKER_USERNAME")?; - // The password might need to be surrounded by single quotes (e.g. 'password') to be read correctly - let password = env::var("MQTT_BROKER_PASSWORD")?; - - // Channel to receive publish messages - let (sender, mut receiver) = mpsc::channel(8); - - // Could directly set it up in the handle new but eh - let stream = set_up_tcp_connection(broker_address, broker_port).await?; - let actor = MqttActorHandle::new(stream, sender); - - actor - .connect( - "testclient".into(), - username.into(), - password.as_bytes().into(), - ) - .await - .unwrap(); - // homeassistant/{domain}/{object_id}/config - - // Prefix is "homeassistant", but it can be changed in home assistant configuration - const DISCOVERY_TOPIC: &str = "homeassistant/fan/testfan/config"; - - // Configuration is like the YAML configuration that would be added in Home Assistant but as JSON - // Command topic: The MQTT topic to publish commands to change the state of the fan - //TODO set firmware version from Cargo.toml package version - //TODO think about setting hardware version, support url, and manufacturer - //TODO create single home assistant device with multiple entities for sensors in fan and the bypass - //TODO add diagnostic entity like IP address - //TODO availability topic - // Using abbreviations to save space of binary and on the wire (haven't measured effect though...) - // name -> name - // uniq_id -> unique_id - // stat_t -> state_topic - // cmd_t -> command_topic - // pct_stat_t -> percentage_state_topic - // pct_cmd_t -> percentage_command_topic - // spd_rng_max -> speed_range_max - // Don't need to set speed_range_min because it is 1 by default - //TODO remove whitespace at compile time through macro, build script or const fn - const DISCOVERY_PAYLOAD: &[u8] = br#"{ - "name": "Fan", - "uniq_id": "testfan", - "stat_t": "testfan/on/state", - "cmd_t": "testfan/on/set", - "pct_stat_t": "testfan/speed/percentage_state", - "pct_cmd_t": "testfan/speed/percentage", - "spd_rng_max": 64000 - }"#; - - println!("Connected"); - - // Subscribing to the state topics before telling home assistant to use them to control the fan - - // Send discover publish packet - actor - .publish(DISCOVERY_TOPIC.into(), DISCOVERY_PAYLOAD.into()) - .await - .unwrap(); - - println!("Published discovery"); - let subscriptions = Arc::new([ - // Listen to when the fan should be turned on or off - // Payload will be "ON" or "OFF" - Subscription::new( - "testfan/on/set".into(), - SubscriptionOptions::new( - QualityOfService::AtMostOnceDelivery, - false, - false, - RetainHandling::DoNotSend, - ), - ), - // Listen to speed changes from home assistant - Subscription::new( - "testfan/speed/percentage".into(), - SubscriptionOptions::new( - QualityOfService::AtMostOnceDelivery, - false, - false, - RetainHandling::DoNotSend, - ), - ), - ]); - actor.subscribe(subscriptions).await.unwrap(); - - while let Some(_) = receiver.recv().await { - println!("Received message") - } - - Ok(()) -} diff --git a/mqtt/src/packet.rs b/mqtt/src/packet.rs deleted file mode 100644 index e84f879..0000000 --- a/mqtt/src/packet.rs +++ /dev/null @@ -1,636 +0,0 @@ -use std::{convert::TryFrom, sync::Arc}; - -/// KA: Keep alive in seconds -pub(super) fn create_connect( - client_identifier: &str, - username: &str, - password: &[u8], -) -> Vec { - let identifier_length = client_identifier.len() as u16; - - // Length is length of bytes not characters - //TODO validate user name is less than u16::MAX - let username_length = username.len() as u16; - //TODO validate password is less than u16::MAX - let password_length = password.len() as u16; - - // Remaining length can only be a byte - //TODO use variable byte integer - //TODO check length is not exceeding variable byte integer max - - // Create variable header - let variable_header: [u8; 11] = [ - // Protocol name length - 0x00, - 0x04, - // Protocol name - b'M', - b'Q', - b'T', - b'T', - // Protocol version - 5, - // Connect Flags - // USER_NAME_FLAG | PASSWORD_FLAG | CLEAN_START - 0b1100_0010, - // Keep alive - (KA >> 8) as u8, - KA as u8, - // Property length 0 (no properties). Has to be set to 0 if there are no properties - 0, - ]; - - // 2 = length of length fields - let remaining_length: u8 = (variable_header.len() as u16 - + 2 - + identifier_length - + 2 - + username_length - + 2 - + password_length) as u8; - - // Fixed header - let fixed_header = [ - // CONNECT package type (1) in the first four bits - 1 << 4, - // Remaining length - remaining_length, - ]; - - let mut packet = Vec::with_capacity(fixed_header.len() + remaining_length as usize); - // packet.append(fixed_header) - packet.extend_from_slice(&fixed_header); - - // Variable header - packet.extend_from_slice(&variable_header); - - // Payload - - // Client identifier - let length_bytes = identifier_length.to_be_bytes(); - // Client identifier length - packet.push(length_bytes[0]); - packet.push(length_bytes[1]); - // Client identifier - packet.extend_from_slice(client_identifier.as_bytes()); - - // User name - let length_bytes = username_length.to_be_bytes(); - packet.push(length_bytes[0]); - packet.push(length_bytes[1]); - - // User name - packet.extend_from_slice(username.as_bytes()); - - // Password is any binary data - let length_bytes = password_length.to_be_bytes(); - packet.push(length_bytes[0]); - packet.push(length_bytes[1]); - // Password - packet.extend_from_slice(password); - - packet -} - -//TODO use NonZero -pub(super) fn create_publish(topic_name: &str, payload: &[u8]) -> Vec { - // Length is length of bytes not characters - let topic_name_length = topic_name.len(); - let payload_length = payload.len(); - - // Remaining length is variable byte integer - // length of topic name length + topic name length + properties length - let variable_header_length = size_of::() + topic_name_length + size_of::(); - //TODO ensure u32 is in less than variable byte integer max - let remaining_length = variable_header_length + payload_length; - let remaining_length_bytes = crate::variable_byte_integer::encode(remaining_length as u32); - let fixed_header_length = 1 + remaining_length_bytes.len(); - let packet_length = fixed_header_length + remaining_length; - let mut packet = Vec::with_capacity(packet_length); - - // Fixed header - // Packet type PUBLISH (3) and flags - packet.push(3 << 4); - packet.extend_from_slice(&remaining_length_bytes); - - // Variable header - // Topic name length - let length_bytes: [u8; 2] = (topic_name_length as u16).to_be_bytes(); - packet.push(length_bytes[0]); - packet.push(length_bytes[1]); - // Topic name - packet.extend_from_slice(topic_name.as_bytes()); - // Property length - // No properties supported for now so set to 0 - packet.push(0); - - // Payload - // No need to set length as it can be calculated - packet.extend_from_slice(payload); - - packet -} - -pub(super) enum QualityOfService { - /// At most once delivery or 0 - AtMostOnceDelivery = 0x00, - /// At least once delivery or 1 - AtLeastOnceDelivery = 0x01, - /// Exactly once delivery or 2 - ExactlyOnceDelivery = 0x02, -} - -impl QualityOfService { - const fn to_byte(&self) -> u8 { - match self { - QualityOfService::AtMostOnceDelivery => 0, - QualityOfService::AtLeastOnceDelivery => 1, - QualityOfService::ExactlyOnceDelivery => 2, - } - } -} - -/// Retain handling option of the subscription options -/// Specifies whether retained messages are sent when the subscription is established. -/// Does not affect sending of retained messages after subscribe. If there are no retained messages -/// matching a topic filter, then all options act the same. -pub(super) enum RetainHandling { - // Send retained messages at the time of subscribe (0) - SendAtSubscribe, - // Send retained messages only if the subscription does not currently exist (1) - OnlyIfNotExists, - // Do not send retained messages (2) - DoNotSend, -} - -impl RetainHandling { - const fn to_byte(&self) -> u8 { - match self { - RetainHandling::SendAtSubscribe => 0, - RetainHandling::OnlyIfNotExists => 1, - RetainHandling::DoNotSend => 2, - } - } -} - -pub(super) struct SubscriptionOptions(u8); - -impl SubscriptionOptions { - pub(super) const fn new( - maximum_quality_of_service: QualityOfService, - is_no_local: bool, - is_retain_as_published: bool, - retain_handling: RetainHandling, - ) -> Self { - Self( - // Bit 0 and 1 (no shift necessary) - maximum_quality_of_service.to_byte() - // Boolean values are stored in a byte as either 0x00 or 0x01 - // Meaning the first bit is on for true and off for false - // << 2 shifts that bit to the 3rd position (0 indexed) - // Bit 2 - | (is_no_local as u8) << 2 - // Bit 3 - | (is_retain_as_published as u8) << 3 - // Bit 4 and 5 - | (retain_handling.to_byte() << 4), - // Bit 6 and 7 are reserved and must be set to 0 - ) - } -} - -pub(super) struct Subscription { - pub(super) topic_filter: Arc, - options: SubscriptionOptions, -} - -impl Subscription { - pub(super) fn new(topic_filter: Arc, options: SubscriptionOptions) -> Self { - Self { - options, - topic_filter, - } - } - - fn length(&self) -> usize { - self.topic_filter.len() + size_of::() - } -} - -pub(super) fn create_subscribe(packet_identifier: u16, subscriptions: &[Subscription]) -> Vec { - // Packet identifier + property length - let variable_header_length = size_of::() + size_of::(); - // Payload contains a list of pairs of subscription options and topic name - // Maybe size_of_val would be accurate too? - let payload_length = - // Length bytes for each subscription - subscriptions.len() * size_of::() + - subscriptions - .iter() - .map(|subscription| subscription.length()) - .sum::(); - let remaining_length = variable_header_length + payload_length; - let remaining_length_bytes = crate::variable_byte_integer::encode(remaining_length as u32); - let fixed_header_length = 1 + remaining_length_bytes.len(); - let packet_length = fixed_header_length + remaining_length; - let mut packet = Vec::with_capacity(packet_length); - - // Subscribe packet type (8) and bits must be set to 0010 - packet.push(0b1000_0010); - // Remaining length - packet.extend_from_slice(&remaining_length_bytes); - - // Variable header - // Packet Identifier - let packet_identifier_bytes = packet_identifier.to_be_bytes(); - packet.extend_from_slice(&packet_identifier_bytes); - - // Property length - // No properties supported for now so set to 0 - packet.push(0); - - for subscription in subscriptions { - let topic_name_length_bytes = (subscription.topic_filter.len() as u16).to_be_bytes(); - packet.extend_from_slice(&topic_name_length_bytes); - packet.extend_from_slice(&subscription.topic_filter.as_bytes()); - let options: u8 = subscription.options.0; - packet.push(options); - } - - packet -} - -#[derive(Debug)] -pub(super) enum ConnectErrorReasonCode { - UnspecifiedError = 0x80, - MalformedPacket = 0x81, - ProtocolError = 0x82, - ImplementationSpecificError = 0x83, - UnsupportedProtocolVersion = 0x84, - ClientIdentifierNotValid = 0x85, - BadUserNameOrPassword = 0x86, - NotAuthorized = 0x87, - ServerUnavailable = 0x88, - ServerBusy = 0x89, - Banned = 0x8A, - BadAuthenticationMethod = 0x8C, - TopicNameInvalid = 0x90, - PacketTooLarge = 0x95, - QuotaExceeded = 0x97, - PayloadFormatInvalid = 0x99, - RetainNotSupported = 0x9A, - QosNotSupported = 0x9B, - UseAnotherServer = 0x9C, - ServerMoved = 0x9D, - ConnectionRateExceeded = 0x9F, -} - -pub(super) enum ConnectReasonCode { - Success, - Error(ConnectErrorReasonCode), -} - -#[derive(Debug, thiserror::Error)] -#[error("Unknown connect reason code: {0}")] -pub(super) struct UnknownConnectReasonCode(u8); -impl TryFrom for ConnectReasonCode { - type Error = UnknownConnectReasonCode; - - fn try_from(value: u8) -> Result>::Error> { - match value { - 0x00 => Ok(ConnectReasonCode::Success), - 0x80 => Ok(ConnectReasonCode::Error( - ConnectErrorReasonCode::UnspecifiedError, - )), - 0x81 => Ok(ConnectReasonCode::Error( - ConnectErrorReasonCode::MalformedPacket, - )), - 0x82 => Ok(ConnectReasonCode::Error( - ConnectErrorReasonCode::ProtocolError, - )), - 0x83 => Ok(ConnectReasonCode::Error( - ConnectErrorReasonCode::ImplementationSpecificError, - )), - 0x84 => Ok(ConnectReasonCode::Error( - ConnectErrorReasonCode::UnsupportedProtocolVersion, - )), - 0x85 => Ok(ConnectReasonCode::Error( - ConnectErrorReasonCode::ClientIdentifierNotValid, - )), - 0x86 => Ok(ConnectReasonCode::Error( - ConnectErrorReasonCode::BadUserNameOrPassword, - )), - 0x87 => Ok(ConnectReasonCode::Error( - ConnectErrorReasonCode::NotAuthorized, - )), - 0x88 => Ok(ConnectReasonCode::Error( - ConnectErrorReasonCode::ServerUnavailable, - )), - 0x89 => Ok(ConnectReasonCode::Error(ConnectErrorReasonCode::ServerBusy)), - 0x8A => Ok(ConnectReasonCode::Error(ConnectErrorReasonCode::Banned)), - 0x8C => Ok(ConnectReasonCode::Error( - ConnectErrorReasonCode::BadAuthenticationMethod, - )), - 0x90 => Ok(ConnectReasonCode::Error( - ConnectErrorReasonCode::TopicNameInvalid, - )), - 0x95 => Ok(ConnectReasonCode::Error( - ConnectErrorReasonCode::PacketTooLarge, - )), - 0x97 => Ok(ConnectReasonCode::Error( - ConnectErrorReasonCode::QuotaExceeded, - )), - 0x99 => Ok(ConnectReasonCode::Error( - ConnectErrorReasonCode::PayloadFormatInvalid, - )), - 0x9A => Ok(ConnectReasonCode::Error( - ConnectErrorReasonCode::RetainNotSupported, - )), - 0x9B => Ok(ConnectReasonCode::Error( - ConnectErrorReasonCode::QosNotSupported, - )), - 0x9C => Ok(ConnectReasonCode::Error( - ConnectErrorReasonCode::UseAnotherServer, - )), - 0x9D => Ok(ConnectReasonCode::Error( - ConnectErrorReasonCode::ServerMoved, - )), - 0x9F => Ok(ConnectReasonCode::Error( - ConnectErrorReasonCode::ConnectionRateExceeded, - )), - - unknown => Err(UnknownConnectReasonCode(unknown)), - } - } -} - -enum ConnectAcknowledgementProperty {} - -#[derive(Debug)] -pub(super) enum PacketType { - Connect = 1, - ConnectAcknowledgement = 2, - Publish = 3, - PublishAcknowledgement = 4, - PublishReceived = 5, - PublishRelease = 6, - PublishComplete = 7, - Subscribe = 8, - SubscribeAcknowledgement = 9, - Unsubscribe = 10, - UnsubscribeAcknowledgement = 11, - PingRequest = 12, - PingResponse = 13, - Disconnect = 14, - Authentication = 15, -} - -#[derive(Debug)] -pub(super) struct UnknownPacketTypeError(u8); -impl TryFrom for PacketType { - type Error = UnknownPacketTypeError; - - fn try_from(value: u8) -> Result { - match value >> 4 { - 1 => Ok(PacketType::Connect), - 2 => Ok(PacketType::ConnectAcknowledgement), - 3 => Ok(PacketType::Publish), - 4 => Ok(PacketType::PublishAcknowledgement), - 5 => Ok(PacketType::PublishReceived), - 6 => Ok(PacketType::PublishRelease), - 7 => Ok(PacketType::PublishComplete), - 8 => Ok(PacketType::Subscribe), - 9 => Ok(PacketType::SubscribeAcknowledgement), - 10 => Ok(PacketType::Unsubscribe), - 11 => Ok(PacketType::UnsubscribeAcknowledgement), - 12 => Ok(PacketType::PingRequest), - 13 => Ok(PacketType::PingResponse), - 14 => Ok(PacketType::Disconnect), - 15 => Ok(PacketType::Authentication), - unknown => Err(UnknownPacketTypeError(unknown)), - } - } -} - -#[cfg(test)] -mod tests { - use super::{ - create_connect, create_publish, create_subscribe, QualityOfService, RetainHandling, - Subscription, SubscriptionOptions, - }; - - #[test] - fn can_create_publish_packet() { - const PAYLOAD: [u8; 11] = *b"testpayload"; - // Act - let publish_packet = create_publish("test", &PAYLOAD); - - // Assert - // Fixed header - assert_eq!(publish_packet[..2], [0b0011_0000, 18]); - - // Variable header - assert_eq!( - publish_packet[2..9], - [ - // Topic name length - 0x00, 0x04, // Topic name - b't', b'e', b's', b't', // Property length - 0x00, - ] - ); - - // Payload - assert_eq!(publish_packet[9..], PAYLOAD); - } - - #[test] - fn can_create_connect_packet() { - // Act - let connect_packet = create_connect::<60>("mqttx_0x668d0d", "admin", b"public"); - // Assert - // Fixed header - assert_eq!(connect_packet[..2], [0b0001_0000, 42]); - // Variable header - assert_eq!( - connect_packet[2..13], - [ - // Protocol name length - 0x00, - 0x04, - // Protocol name - b'M', - b'Q', - b'T', - b'T', - // 0x4d, 0x51, 0x54, 0x54, - // Protocol version - 5, - // Connect Flags - // USER_NAME_FLAG | PASSWORD_FLAG | CLEAN_START - 0b1100_0010, - // 0xc2, - // Keep alive 60 seconds - 0x00, - 60, - // Property length 0 (no properties). Has to be set to 0 if there are no properties - 0, - ] - ); - - // Payload - - #[rustfmt::skip] - assert_eq!( - connect_packet[13..], - // Rust formatting is doing some wild stuff here for some reason - [ - // Client identifier length - 0x00, 0x0e, - // Client identifier - b'm', b'q', b't', b't', b'x', b'_', b'0', b'x', b'6', b'6', b'8', b'd', b'0', b'd', - // User name length - 0x00, 0x05, - // User name - 0x61, 0x64, 0x6d, 0x69, 0x6e, - // Password length - 0x00, 0x06, - // Password - b'p', b'u', b'b', b'l', b'i', b'c', - ] - ); - } - - #[test] - fn can_create_options() { - let subscription_options = SubscriptionOptions::new( - QualityOfService::ExactlyOnceDelivery, - true, - true, - RetainHandling::DoNotSend, - ); - - // Assert bit 0 and 1 are set for QualityOfService - let expected: u8 = 0b0000_0010; - let actual = subscription_options.0 & 0b0000_0010; - assert_eq!( - actual, expected, - "Expected bit 0 and 1 to be set for Quality of service {expected:#010b} but was {actual:#010b}" - ); - - // Assert bit 2 is set - let expected: u8 = 0b0000_0100; - let actual = subscription_options.0 & 0b0000_0100; - assert_eq!( - actual, expected, - "Expected bit 2 to be set for No Local but was {actual:#010b}" - ); - - // Assert bit 3 is set - let expected: u8 = 0b0000_1000; - let actual = subscription_options.0 & 0b0000_1000; - assert_eq!( - actual, expected, - "Expected bit 3 to be set for Retain as Published but was {actual:#010b}" - ); - - // Assert bit 4 and 5 is set - let expected: u8 = 0b0001_000; - let actual = subscription_options.0 & 0b0000_1000; - assert_eq!( - actual, expected, - "Expected bit 4 and 5 to be set for Retain Handling but was {actual:#010b}" - ); - - // Assert bit 6 and 7 are 0 - let expected: u8 = 0b0000_0000; - let actual = subscription_options.0 & 0b1100_0000; - assert_eq!(actual, expected); - } - - #[test] - fn can_correctly_set_retain_handling() { - let subscription_options = SubscriptionOptions::new( - QualityOfService::AtMostOnceDelivery, - false, - false, - RetainHandling::SendAtSubscribe, - ); - - assert_eq!(subscription_options.0, 0); - - let subscription_options = SubscriptionOptions::new( - QualityOfService::AtMostOnceDelivery, - false, - false, - RetainHandling::DoNotSend, - ); - - // Assert bit 4 is set - let expected: u8 = 0b0010_0000; - // ------------------------------------------7654_3210 - let actual = subscription_options.0 & 0b0010_0000; - assert_eq!( - actual, expected, - "Expected bit 5 to be set for Retain Handling but was {:#010b}", - subscription_options.0 - ); - - let subscription_options = SubscriptionOptions::new( - QualityOfService::AtMostOnceDelivery, - false, - false, - RetainHandling::OnlyIfNotExists, - ); - - let expected: u8 = 0b0001_0000; - let actual = subscription_options.0 & 0b0001_0000; - assert_eq!( - actual, expected, - "Expected bit 4 to be set for Retain Handling but was {:#010b}", - subscription_options.0 - ); - } - - #[test] - fn can_create_subscribe_packet() { - // Arrange - let subscription = Subscription::new( - "demo", - SubscriptionOptions::new( - QualityOfService::ExactlyOnceDelivery, - false, - false, - RetainHandling::SendAtSubscribe, - ), - ); - - // Act - let subscribe_packet = create_subscribe(1470, &[subscription]); - - // Assert - // Check fixed header - assert_eq!(subscribe_packet[..2], [(8 << 4 | 2), 0x0a]); - - // Check variable header - assert_eq!( - subscribe_packet[2..5], - [0x05, 0xbe, 0x00], - "Actual: {:#04x?}", - &subscribe_packet[2..5] - ); - - // Check payload - // Topic length - assert_eq!(subscribe_packet[5..7], [0x00, 0x04]); - // Topic filter - assert_eq!(&subscribe_packet[7..11], b"demo"); - // Options - let expected = 0b0000_0010; - let actual = subscribe_packet[11]; - assert_eq!( - expected, actual, - "Expected {expected:#010b} actual {actual:#010b}", - ); - } -} diff --git a/mqtt/src/variable_byte_integer.rs b/mqtt/src/variable_byte_integer.rs deleted file mode 100644 index 894e93b..0000000 --- a/mqtt/src/variable_byte_integer.rs +++ /dev/null @@ -1,354 +0,0 @@ -use std::num::{NonZero, NonZeroUsize}; - -use tokio::{io::AsyncReadExt, net::TcpStream}; - -/// Encodes an integer as variable byte integer according to the MQTT specification -pub(super) fn encode(mut value: u32) -> Vec { - //TODO Try to accept any integer type - //TODO Error if value is too large - - // Each byte can hold 7 bits, so how many 7 bit "bytes" do we need for 32 bits? - // 32 / 7 - let length: usize = match value { - // Short circuit when it fits in one byte - 0..=127 => return vec![value as u8], - 128..=16_383 => 2, - 16_384..=2_097_151 => 3, - 2_097_152..=268_435_455 => 4, - _ => 5, - }; - let mut output = Vec::with_capacity(length); - loop { - // 128 = 0b1000_0000 - - // x = 128 = 0b1000_0000 - // encoded_byte = 128 % 0b1000_0000 = 0 - let mut encoded_byte = value % 0b1000_0000; - - // x = 128 / 128 = 1 - value = value / 0b1000_0000; - // If there is more data to encode, set the top bit of this byte - if value > 0 { - encoded_byte |= 0b1000_0000; - } - - output.push(encoded_byte as u8); - - if value == 0 { - break; - } - } - - output -} - -const fn length() -> usize { - if N < 128 { - return 1; - } - - if N < 16_384 { - return 2; - } - - if N < 2_097_152 { - return 3; - } - - if N < 268_435_456 { - return 4; - } - - panic!("Variable byte integer can only hold up to 268_435_455 which is 4 bytes"); -} - -const fn encode_const(mut value: u32) -> [u8; N] { - let mut output = [0; N]; - let mut index = 0; - - loop { - let mut encoded_byte = value % 0b1000_0000; - value = value / 0b1000_0000; - if value > 0 { - encoded_byte |= 0b1000_0000; - } - output[index] = encoded_byte as u8; - index += 1; - if value == 0 { - break; - } - } - output -} - -enum VariableByteIntegerError { - /// The integer is larger than 268,435,455 ([VariableByteInteger::MAX]). - TooLarge, -} - -/// The variable byte integer is an integer that can be encoded in 1-4 bytes. -/// The first bit of every byte is a continuation bit, if it is set, the next byte is also part of the integer. -/// Internally it is represented as a 4 byte array because variable size byte arrays with known size at compile time is hard. -/// And we can't use vectors because they are not const and no std. -/// So this really just saves on data send on the wire. -pub(super) struct VariableByteInteger([u8; 4]); - -/// Helps to mark an integer as a non zero integer when you know at compile time that it is not zero -macro_rules! non_zero_usize { - (0) => { - compile_error!("0 is not a valid NonZeroUsize") - }; - ($value:expr) => {{ - const VALUE: NonZeroUsize = { - let Some(value) = NonZeroUsize::new($value) else { - // Using compile_error!() would always cause a compile error. - // So panic at const time it is - // Formatting goes strange here - panic!(stringify!($value is not a valid NonZeroUsize)); - }; - - value - }; - - VALUE - }}; -} - -impl VariableByteInteger { - const MAX: u32 = 268_435_455; - - pub(super) const fn encode(mut value: u32) -> Result { - if value > Self::MAX { - return Err(VariableByteIntegerError::TooLarge); - } - - // Short circuit if it fits in one byte (0-127) - if value < 128 { - return Ok(Self([value as u8, 0, 0, 0])); - } - - let mut output = [0; 4]; - let mut index = 0; - - loop { - let mut encoded_byte = value % 0b1000_0000; - value = value / 0b1000_0000; - if value > 0 { - encoded_byte |= 0b1000_0000; - } - output[index] = encoded_byte as u8; - index += 1; - if value == 0 { - break; - } - } - Ok(Self(output)) - } - - //TODO clean up having all these decode and encode versions - pub(super) async fn decode( - stream: &mut TcpStream, - ) -> Result<(usize, usize), DecodeVariableByteIntegerError> { - let mut multiplier = 1; - let mut value = 0; - let mut index = 0; - - loop { - let encoded_byte = stream.read_u8().await?; - index += 1; - - value += (encoded_byte & 127) as usize * multiplier; - - if multiplier > 128 * 128 * 128 { - return Err(DecodeVariableByteIntegerError::MalformedVariableByteIntegerError); - } - - multiplier *= 128; - - // The last byte has the most significant bit set to 0 indicating that there are no more bytes to follow - if (encoded_byte & 128) == 0 { - break; - } - - // The current byte indicates the next byte is part of the integer but we would be past 4 bytes - if index > 4 { - return Err(DecodeVariableByteIntegerError::InvalidLength); - } - } - - // As index is immediately increased it reflects the length at this point - // index is usize but can't pass the value 4 - Ok((value, index)) - } - - pub(super) fn decode_2(bytes: &[u8]) -> Result<(usize, usize), DecodeVariableByteIntegerError> { - let mut multiplier = 1; - let mut value = 0; - let mut index = 0; - - loop { - let encoded_byte = bytes[index]; - index += 1; - - value += (encoded_byte & 127) as usize * multiplier; - - if multiplier > 128 * 128 * 128 { - return Err(DecodeVariableByteIntegerError::MalformedVariableByteIntegerError); - } - - multiplier *= 128; - - // The last byte has the most significant bit set to 0 indicating that there are no more bytes to follow - if (encoded_byte & 128) == 0 { - break; - } - - // The current byte indicates the next byte is part of the integer but we would be past 4 bytes - if index > 4 { - return Err(DecodeVariableByteIntegerError::InvalidLength); - } - } - - // As index is immediately increased it reflects the length at this point - // index is usize but can't pass the value 4 - Ok((value, index)) - } - - /// Returns the length of the encoded integer in bytes which is either 1, 2, 3 or 4. - /// This helps to determine how many bytes are needed on the wire to represent this integer. - const fn length(&self) -> NonZero { - if (self.0[2] & 0b1000_0000) != 0 { - return non_zero_usize!(4); - } - - if (self.0[1] & 0b1000_0000) != 0 { - return non_zero_usize!(3); - } - - if (self.0[0] & 0b1000_0000) != 0 { - return non_zero_usize!(2); - } - - return non_zero_usize!(1); - } -} - -#[test] -fn can_encode_variable_byte_integer() { - // Random examples from https://www.emqx.com/en/blog/mqtt-5-0-control-packets-01-connect-connack - assert_eq!(encode_const(47), [0x2f]); - assert_eq!(encode_const(19), [0x13]); - - // Range one byte - assert_eq!(encode_const(0), [0b0000_0000]); - assert_eq!(encode_const(1), [0b0000_0001]); - assert_eq!(encode_const(127), [0b0111_1111]); - // Range two bytes - assert_eq!(encode_const(128), [0b1000_0000, 0b0000_0001]); - assert_eq!(encode_const(16_383), [0b1111_1111, 0b0111_1111]); - - // Range three bytes - assert_eq!( - encode_const(16_384), - [0b1000_0000, 0b1000_0000, 0b0000_0001] - ); - assert_eq!( - encode_const(2_097_151), - [0b1111_1111, 0b1111_1111, 0b0111_1111] - ); - // Range four bytes - assert_eq!( - encode_const(2_097_152), - [0b1000_0000, 0b1000_0000, 0b1000_0000, 0b0000_0001] - ); - assert_eq!( - encode_const(268_435_455), - [0b1111_1111, 0b1111_1111, 0b1111_1111, 0b0111_1111] - ); -} - -#[derive(Debug, thiserror::Error)] -pub(super) enum DecodeVariableByteIntegerError { - #[error("Error reading from stream: {0}")] - ReadError(#[from] std::io::Error), - #[error("The variable byte integer is malformed")] - MalformedVariableByteIntegerError, - /// The last byte indicates that the next byte is part of the variable byte integer but there was no next byte - #[error("The variable byte ended unexpectedly")] - UnexpectedEndOfInput, - #[error("The variable byte integer is longer than 4 bytes")] - InvalidLength, -} - -fn decode_variable_byte_integer( - bytes: impl IntoIterator + std::fmt::Debug, -) -> Result { - let mut multiplier = 1; - let mut value: u32 = 0; - - let mut bytes = bytes.into_iter(); - loop { - let encoded_byte = bytes - .next() - .ok_or(DecodeVariableByteIntegerError::UnexpectedEndOfInput)?; - - value += (encoded_byte & 127) as u32 * multiplier; - - if multiplier > 128 * 128 * 128 { - return Err(DecodeVariableByteIntegerError::MalformedVariableByteIntegerError); - } - - multiplier *= 128; - - // The last byte has the most significant bit set to 0 indicating that there are no more bytes to follow - if (encoded_byte & 128) != 0 { - continue; - } else { - break; - } - } - - Ok(value) -} - -#[test] -fn can_decode_variable_byte_integer() { - // Random examples from https://www.emqx.com/en/blog/mqtt-5-0-control-packets-01-connect-connack - assert_eq!(decode_variable_byte_integer([0x2f]).unwrap(), 47); - assert_eq!(decode_variable_byte_integer([0x13]).unwrap(), 19); - - // Range one byte - assert_eq!(decode_variable_byte_integer([0b0000_0000]).unwrap(), 0); - assert_eq!(decode_variable_byte_integer([0b0000_0001]).unwrap(), 1); - assert_eq!(decode_variable_byte_integer([0b0111_1111]).unwrap(), 127); - - // Range two bytes - assert_eq!( - decode_variable_byte_integer([0b1000_0000, 0b0000_0001]).unwrap(), - 128 - ); - assert_eq!( - decode_variable_byte_integer([0b1111_1111, 0b0111_1111]).unwrap(), - 16_383 - ); - - // Range three bytes - assert_eq!( - decode_variable_byte_integer([0b1000_0000, 0b1000_0000, 0b0000_0001]).unwrap(), - 16_384 - ); - assert_eq!( - decode_variable_byte_integer([0b1111_1111, 0b1111_1111, 0b0111_1111]).unwrap(), - 2_097_151 - ); - - // Range four bytes - assert_eq!( - decode_variable_byte_integer([0b1000_0000, 0b1000_0000, 0b1000_0000, 0b0000_0001]).unwrap(), - 2_097_152 - ); - assert_eq!( - decode_variable_byte_integer([0b1111_1111, 0b1111_1111, 0b1111_1111, 0b0111_1111]).unwrap(), - 268_435_455 - ); -}