commit 2a0543d0b1aa1cfc18b651c6258bb33a65f17bab Author: Stephen Waits Date: Fri Apr 3 17:01:27 2026 -0600 feat: add Rust client library and mux daemon Add donglora-client (clients/rust/) and donglora-mux (mux/rust/) as Rust implementations of the existing Python client library and mux daemon. Both are standalone Cargo projects with no-panic guarantees enforced by clippy deny lints. donglora-client provides protocol types, COBS framing, USB device discovery, serial/mux transports, and a high-level Client with send/recv. donglora-mux is a fully async (tokio) multiplexer daemon with dongle hot-plug reconnect, per-client bounded send queues, SetConfig locking, and StartRx/StopRx reference counting. 52 unit tests across both crates. Updates README roadmap checkboxes and CONTRIBUTING docs to reflect the new implementations. diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ea8c4bf --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +/target diff --git a/Cargo.lock b/Cargo.lock new file mode 100644 index 0000000..8ec2cee --- /dev/null +++ b/Cargo.lock @@ -0,0 +1,903 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "aho-corasick" +version = "1.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ddd31a130427c27518df266943a5308ed92d4b226cc639f5a8f1002816174301" +dependencies = [ + "memchr", +] + +[[package]] +name = "anstream" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "824a212faf96e9acacdbd09febd34438f8f711fb84e09a8916013cd7815ca28d" +dependencies = [ + "anstyle", + "anstyle-parse", + "anstyle-query", + "anstyle-wincon", + "colorchoice", + "is_terminal_polyfill", + "utf8parse", +] + +[[package]] +name = "anstyle" +version = "1.0.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000" + +[[package]] +name = "anstyle-parse" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52ce7f38b242319f7cabaa6813055467063ecdc9d355bbb4ce0c68908cd8130e" +dependencies = [ + "utf8parse", +] + +[[package]] +name = "anstyle-query" +version = "1.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" +dependencies = [ + "windows-sys 0.61.2", +] + +[[package]] +name = "anstyle-wincon" +version = "3.0.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" +dependencies = [ + "anstyle", + "once_cell_polyfill", + "windows-sys 0.61.2", +] + +[[package]] +name = "anyhow" +version = "1.0.102" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" + +[[package]] +name = "bitflags" +version = "1.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a" + +[[package]] +name = "bitflags" +version = "2.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "843867be96c8daad0d758b57df9392b6d8d271134fce549de6ce169ff98a92af" + +[[package]] +name = "bytes" +version = "1.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" + +[[package]] +name = "cfg-if" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" + +[[package]] +name = "cfg_aliases" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" + +[[package]] +name = "clap" +version = "4.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b193af5b67834b676abd72466a96c1024e6a6ad978a1f484bd90b85c94041351" +dependencies = [ + "clap_builder", + "clap_derive", +] + +[[package]] +name = "clap_builder" +version = "4.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "714a53001bf66416adb0e2ef5ac857140e7dc3a0c48fb28b2f10762fc4b5069f" +dependencies = [ + "anstream", + "anstyle", + "clap_lex", + "strsim", +] + +[[package]] +name = "clap_derive" +version = "4.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1110bd8a634a1ab8cb04345d8d878267d57c3cf1b38d91b71af6686408bbca6a" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "clap_lex" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9" + +[[package]] +name = "cobs" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0fa961b519f0b462e3a3b4a34b64d119eeaca1d59af726fe450bbba07a9fc0a1" +dependencies = [ + "thiserror", +] + +[[package]] +name = "colorchoice" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" + +[[package]] +name = "core-foundation" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b2a6cd9ae233e7f62ba4e9353e81a88df7fc8a5987b8d445b4d90c879bd156f6" +dependencies = [ + "core-foundation-sys", + "libc", +] + +[[package]] +name = "core-foundation-sys" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" + +[[package]] +name = "donglora-client" +version = "0.1.0" +dependencies = [ + "anyhow", + "cobs", + "serialport", + "tracing", +] + +[[package]] +name = "donglora-mux" +version = "0.1.0" +dependencies = [ + "anyhow", + "clap", + "donglora-client", + "tokio", + "tokio-serial", + "tokio-util", + "tracing", + "tracing-subscriber", +] + +[[package]] +name = "errno" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + +[[package]] +name = "futures" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b147ee9d1f6d097cef9ce628cd2ee62288d963e16fb287bd9286455b241382d" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-channel" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d" +dependencies = [ + "futures-core", + "futures-sink", +] + +[[package]] +name = "futures-core" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" + +[[package]] +name = "futures-executor" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "baf29c38818342a3b26b5b923639e7b1f4a61fc5e76102d4b1981c6dc7a7579d" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-io" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718" + +[[package]] +name = "futures-macro" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "futures-sink" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c39754e157331b013978ec91992bde1ac089843443c49cbc7f46150b0fad0893" + +[[package]] +name = "futures-task" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "037711b3d59c33004d3856fbdc83b99d4ff37a24768fa1be9ce3538a1cde4393" + +[[package]] +name = "futures-util" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" +dependencies = [ + "futures-channel", + "futures-core", + "futures-io", + "futures-macro", + "futures-sink", + "futures-task", + "memchr", + "pin-project-lite", + "slab", +] + +[[package]] +name = "heck" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" + +[[package]] +name = "io-kit-sys" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "617ee6cf8e3f66f3b4ea67a4058564628cde41901316e19f559e14c7c72c5e7b" +dependencies = [ + "core-foundation-sys", + "mach2", +] + +[[package]] +name = "is_terminal_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" + +[[package]] +name = "lazy_static" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" + +[[package]] +name = "libc" +version = "0.2.184" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "48f5d2a454e16a5ea0f4ced81bd44e4cfc7bd3a507b61887c99fd3538b28e4af" + +[[package]] +name = "libudev" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78b324152da65df7bb95acfcaab55e3097ceaab02fb19b228a9eb74d55f135e0" +dependencies = [ + "libc", + "libudev-sys", +] + +[[package]] +name = "libudev-sys" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3c8469b4a23b962c1396b9b451dda50ef5b283e8dd309d69033475fa9b334324" +dependencies = [ + "libc", + "pkg-config", +] + +[[package]] +name = "lock_api" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965" +dependencies = [ + "scopeguard", +] + +[[package]] +name = "log" +version = "0.4.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" + +[[package]] +name = "mach2" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d640282b302c0bb0a2a8e0233ead9035e3bed871f0b7e81fe4a1ec829765db44" +dependencies = [ + "libc", +] + +[[package]] +name = "matchers" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9" +dependencies = [ + "regex-automata", +] + +[[package]] +name = "memchr" +version = "2.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79" + +[[package]] +name = "mio" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "50b7e5b27aa02a74bac8c3f23f448f8d87ff11f92d3aac1a6ed369ee08cc56c1" +dependencies = [ + "libc", + "log", + "wasi", + "windows-sys 0.61.2", +] + +[[package]] +name = "mio-serial" +version = "5.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "029e1f407e261176a983a6599c084efd322d9301028055c87174beac71397ba3" +dependencies = [ + "log", + "mio", + "nix 0.29.0", + "serialport", + "winapi", +] + +[[package]] +name = "nix" +version = "0.26.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "598beaf3cc6fdd9a5dfb1630c2800c7acd31df7aaf0f565796fba2b53ca1af1b" +dependencies = [ + "bitflags 1.3.2", + "cfg-if", + "libc", +] + +[[package]] +name = "nix" +version = "0.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "71e2746dc3a24dd78b3cfcb7be93368c6de9963d30f43a6a73998a9cf4b17b46" +dependencies = [ + "bitflags 2.11.0", + "cfg-if", + "cfg_aliases", + "libc", +] + +[[package]] +name = "nu-ansi-term" +version = "0.50.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" +dependencies = [ + "windows-sys 0.61.2", +] + +[[package]] +name = "once_cell" +version = "1.21.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" + +[[package]] +name = "once_cell_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" + +[[package]] +name = "parking_lot" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a" +dependencies = [ + "lock_api", + "parking_lot_core", +] + +[[package]] +name = "parking_lot_core" +version = "0.9.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1" +dependencies = [ + "cfg-if", + "libc", + "redox_syscall", + "smallvec", + "windows-link", +] + +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + +[[package]] +name = "pkg-config" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c" + +[[package]] +name = "proc-macro2" +version = "1.0.106" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fd00f0bb2e90d81d1044c2b32617f68fcb9fa3bb7640c23e9c748e53fb30934" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.45" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41f2619966050689382d2b44f664f4bc593e129785a36d6ee376ddf37259b924" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "redox_syscall" +version = "0.5.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" +dependencies = [ + "bitflags 2.11.0", +] + +[[package]] +name = "regex-automata" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e1dd4122fc1595e8162618945476892eefca7b88c52820e74af6262213cae8f" +dependencies = [ + "aho-corasick", + "memchr", + "regex-syntax", +] + +[[package]] +name = "regex-syntax" +version = "0.8.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc897dd8d9e8bd1ed8cdad82b5966c3e0ecae09fb1907d58efaa013543185d0a" + +[[package]] +name = "scopeguard" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" + +[[package]] +name = "serialport" +version = "4.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4d91116f97173694f1642263b2ff837f80d933aa837e2314969f6728f661df3" +dependencies = [ + "bitflags 2.11.0", + "cfg-if", + "core-foundation", + "core-foundation-sys", + "io-kit-sys", + "libudev", + "mach2", + "nix 0.26.4", + "scopeguard", + "unescaper", + "windows-sys 0.52.0", +] + +[[package]] +name = "sharded-slab" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6" +dependencies = [ + "lazy_static", +] + +[[package]] +name = "signal-hook-registry" +version = "1.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" +dependencies = [ + "errno", + "libc", +] + +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + +[[package]] +name = "smallvec" +version = "1.15.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" + +[[package]] +name = "socket2" +version = "0.6.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + +[[package]] +name = "strsim" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f" + +[[package]] +name = "syn" +version = "2.0.117" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e665b8803e7b1d2a727f4023456bbbbe74da67099c585258af0ad9c5013b9b99" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "thiserror" +version = "2.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4288b5bcbc7920c07a1149a35cf9590a2aa808e0bc1eafaade0b80947865fbc4" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebc4ee7f67670e9b64d05fa4253e753e016c6c95ff35b89b7941d6b856dec1d5" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "thread_local" +version = "1.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f60246a4944f24f6e018aa17cdeffb7818b76356965d03b07d6a9886e8962185" +dependencies = [ + "cfg-if", +] + +[[package]] +name = "tokio" +version = "1.51.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2bd1c4c0fc4a7ab90fc15ef6daaa3ec3b893f004f915f2392557ed23237820cd" +dependencies = [ + "bytes", + "libc", + "mio", + "parking_lot", + "pin-project-lite", + "signal-hook-registry", + "socket2", + "tokio-macros", + "windows-sys 0.61.2", +] + +[[package]] +name = "tokio-macros" +version = "2.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "385a6cb71ab9ab790c5fe8d67f1645e6c450a7ce006a33de03daa956cf70a496" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "tokio-serial" +version = "5.4.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aa1d5427f11ba7c5e6384521cfd76f2d64572ff29f3f4f7aa0f496282923fdc8" +dependencies = [ + "cfg-if", + "futures", + "log", + "mio-serial", + "serialport", + "tokio", +] + +[[package]] +name = "tokio-util" +version = "0.7.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ae9cec805b01e8fc3fd2fe289f89149a9b66dd16786abd8b19cfa7b48cb0098" +dependencies = [ + "bytes", + "futures-core", + "futures-sink", + "pin-project-lite", + "tokio", +] + +[[package]] +name = "tracing" +version = "0.1.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" +dependencies = [ + "pin-project-lite", + "tracing-attributes", + "tracing-core", +] + +[[package]] +name = "tracing-attributes" +version = "0.1.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "tracing-core" +version = "0.1.36" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" +dependencies = [ + "once_cell", + "valuable", +] + +[[package]] +name = "tracing-log" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3" +dependencies = [ + "log", + "once_cell", + "tracing-core", +] + +[[package]] +name = "tracing-subscriber" +version = "0.3.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb7f578e5945fb242538965c2d0b04418d38ec25c79d160cd279bf0731c8d319" +dependencies = [ + "matchers", + "nu-ansi-term", + "once_cell", + "regex-automata", + "sharded-slab", + "smallvec", + "thread_local", + "tracing", + "tracing-core", + "tracing-log", +] + +[[package]] +name = "unescaper" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4064ed685c487dbc25bd3f0e9548f2e34bab9d18cefc700f9ec2dba74ba1138e" +dependencies = [ + "thiserror", +] + +[[package]] +name = "unicode-ident" +version = "1.0.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" + +[[package]] +name = "utf8parse" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" + +[[package]] +name = "valuable" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" + +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + +[[package]] +name = "winapi" +version = "0.3.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419" +dependencies = [ + "winapi-i686-pc-windows-gnu", + "winapi-x86_64-pc-windows-gnu", +] + +[[package]] +name = "winapi-i686-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6" + +[[package]] +name = "winapi-x86_64-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" + +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-sys" +version = "0.52.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" +dependencies = [ + "windows-targets", +] + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + +[[package]] +name = "windows-targets" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" +dependencies = [ + "windows_aarch64_gnullvm", + "windows_aarch64_msvc", + "windows_i686_gnu", + "windows_i686_gnullvm", + "windows_i686_msvc", + "windows_x86_64_gnu", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc", +] + +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" + +[[package]] +name = "windows_aarch64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" + +[[package]] +name = "windows_i686_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" + +[[package]] +name = "windows_i686_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" + +[[package]] +name = "windows_i686_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" + +[[package]] +name = "windows_x86_64_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" + +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" + +[[package]] +name = "windows_x86_64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" diff --git a/Cargo.toml b/Cargo.toml new file mode 100644 index 0000000..2f356a3 --- /dev/null +++ b/Cargo.toml @@ -0,0 +1,25 @@ +[package] +name = "donglora-mux" +version = "0.1.0" +edition = "2024" +authors = ["Stephen Waits "] +license = "MIT" +description = "DongLoRa USB Multiplexer — share one dongle with multiple applications" +repository = "https://github.com/swaits/donglora" +keywords = ["lora", "radio", "usb", "multiplexer"] +categories = ["hardware-support"] + +[dependencies] +donglora-client = { path = "../../clients/rust" } +anyhow = "1" +clap = { version = "4", features = ["derive"] } +tokio = { version = "1", features = ["full"] } +tokio-serial = "5" +tokio-util = "0.7" +tracing = "0.1" +tracing-subscriber = { version = "0.3", features = ["env-filter"] } + +[lints.clippy] +unwrap_used = "deny" +expect_used = "deny" +panic = "deny" diff --git a/justfile b/justfile new file mode 100644 index 0000000..bfd0025 --- /dev/null +++ b/justfile @@ -0,0 +1,19 @@ +set shell := ["bash", "-c"] + +default: run + +# Run the mux daemon +run *args: + cargo run -- {{args}} + +# Run with verbose logging +verbose *args: + cargo run -- --verbose {{args}} + +# Run clippy and tests +check: + cargo clippy -- -D warnings && cargo test + +# Build release +build: + cargo build --release diff --git a/src/daemon.rs b/src/daemon.rs new file mode 100644 index 0000000..97762a6 --- /dev/null +++ b/src/daemon.rs @@ -0,0 +1,628 @@ +//! Core mux daemon: dongle I/O, client management, reconnect loop. +//! +//! The daemon owns the USB serial connection and exposes Unix domain socket +//! and optional TCP listeners. It routes commands from clients to the dongle +//! and broadcasts/routes responses back. + +use std::collections::HashMap; +use std::path::Path; +use std::sync::Arc; +use std::time::Duration; + +use donglora_client::{ + encode_frame, FrameReader, CMD_TAG_SET_CONFIG, CMD_TAG_START_RX, CMD_TAG_STOP_RX, + RADIO_CONFIG_SIZE, RESP_TAG_OK, RESP_TAG_RX_PACKET, +}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, UnixListener}; +use tokio::sync::{mpsc, oneshot, Mutex}; + +/// A queued command: (client_id, raw_cmd_bytes). +type QueuedCmd = (u64, Vec); + +/// Shared command receiver, wrapped for use across reconnect cycles. +type SharedCmdRx = Arc>>; +use tokio_util::sync::CancellationToken; +use tracing::{debug, error, info, warn}; + +use crate::intercept::{self, MuxState}; +use crate::session::ClientSession; + +/// Command timeout waiting for dongle response. +const COMMAND_TIMEOUT: Duration = Duration::from_secs(10); + +/// Delay between dongle reconnect attempts. +const RECONNECT_DELAY: Duration = Duration::from_secs(2); + +/// Max consecutive dongle read errors before declaring disconnect. +const MAX_CONSECUTIVE_ERRORS: u32 = 3; + +/// Sentinel client ID for synthetic commands (e.g. auto StopRx). +const SYNTHETIC_CLIENT_ID: u64 = u64::MAX; + +/// Shared mux state, protected by async mutex. +type SharedState = Arc>; + +struct MuxInner { + sessions: HashMap, + mux_state: MuxState, +} + +/// The multiplexer daemon. +pub struct MuxDaemon { + serial_port: String, + socket_path: String, + tcp_addr: Option<(String, u16)>, + shutdown: CancellationToken, +} + +impl MuxDaemon { + pub fn new( + serial_port: String, + socket_path: String, + tcp_addr: Option<(String, u16)>, + shutdown: CancellationToken, + ) -> Self { + Self { + serial_port, + socket_path, + tcp_addr, + shutdown, + } + } + + /// Run the mux daemon. Returns when shutdown is signalled. + pub async fn run(&self) -> anyhow::Result<()> { + let state: SharedState = Arc::new(Mutex::new(MuxInner { + sessions: HashMap::new(), + mux_state: MuxState::new(), + })); + + // Command channel: client handlers → dongle writer + let (cmd_tx, cmd_rx) = mpsc::channel::<(u64, Vec)>(64); + let cmd_rx = Arc::new(Mutex::new(cmd_rx)); + + // Pending response slot: dongle writer installs, dongle reader resolves + let pending_response: Arc>>>> = + Arc::new(Mutex::new(None)); + + // Prepare socket directory + let sock_path = Path::new(&self.socket_path); + if let Some(parent) = sock_path.parent() { + tokio::fs::create_dir_all(parent).await.ok(); + } + if sock_path.exists() { + tokio::fs::remove_file(sock_path).await.ok(); + } + + // Start Unix socket server + let unix_listener = UnixListener::bind(&self.socket_path) + .map_err(|e| anyhow::anyhow!("failed to bind Unix socket {}: {e}", self.socket_path))?; + info!("Unix socket listening on {}", self.socket_path); + + // Start TCP server (optional) + let tcp_listener = if let Some((ref host, port)) = self.tcp_addr { + let addr = format!("{host}:{port}"); + let listener = TcpListener::bind(&addr) + .await + .map_err(|e| anyhow::anyhow!("failed to bind TCP {addr}: {e}"))?; + info!("TCP listening on {addr}"); + Some(listener) + } else { + None + }; + + // Spawn client accept tasks + tokio::spawn(accept_unix_clients( + unix_listener, + state.clone(), + cmd_tx.clone(), + self.shutdown.clone(), + )); + if let Some(listener) = tcp_listener { + tokio::spawn(accept_tcp_clients( + listener, + state.clone(), + cmd_tx.clone(), + self.shutdown.clone(), + )); + } + + // Connect/reconnect loop + self.reconnect_loop(state, cmd_tx, cmd_rx, pending_response) + .await; + + // Cleanup + info!("shutting down..."); + if Path::new(&self.socket_path).exists() { + tokio::fs::remove_file(&self.socket_path).await.ok(); + } + info!("stopped."); + Ok(()) + } + + async fn reconnect_loop( + &self, + state: SharedState, + _cmd_tx: mpsc::Sender<(u64, Vec)>, // kept alive to prevent channel close + cmd_rx: SharedCmdRx, + pending_response: Arc>>>>, + ) { + loop { + if self.shutdown.is_cancelled() { + break; + } + + info!("opening dongle on {}", self.serial_port); + let serial = loop { + if self.shutdown.is_cancelled() { + return; + } + match open_and_ping(&self.serial_port).await { + Ok(s) => break s, + Err(e) => { + debug!("could not open dongle: {e}"); + tokio::select! { + () = self.shutdown.cancelled() => return, + () = tokio::time::sleep(RECONNECT_DELAY) => continue, + } + } + } + }; + + info!("dongle connected"); + let (serial_read, serial_write) = tokio::io::split(serial); + + let dongle_lost = CancellationToken::new(); + + let reader_handle = tokio::spawn(dongle_reader( + serial_read, + state.clone(), + pending_response.clone(), + dongle_lost.clone(), + self.shutdown.clone(), + )); + + let writer_handle = tokio::spawn(dongle_writer( + serial_write, + cmd_rx.clone(), + state.clone(), + pending_response.clone(), + dongle_lost.clone(), + self.shutdown.clone(), + )); + + // Wait for dongle loss or shutdown + tokio::select! { + () = self.shutdown.cancelled() => {}, + () = dongle_lost.cancelled() => {}, + } + + reader_handle.abort(); + writer_handle.abort(); + let _ = reader_handle.await; + let _ = writer_handle.await; + + // Drain pending commands + { + let mut rx = cmd_rx.lock().await; + while rx.try_recv().is_ok() {} + } + + if !self.shutdown.is_cancelled() { + info!("reconnecting in 2 seconds..."); + tokio::select! { + () = self.shutdown.cancelled() => return, + () = tokio::time::sleep(RECONNECT_DELAY) => {}, + } + } + } + } +} + +// ── Dongle I/O ───────────────────────────────────────────────────── + +async fn open_and_ping(port: &str) -> anyhow::Result { + let builder = tokio_serial::new(port, 115_200).timeout(Duration::from_secs(2)); + let mut serial = tokio_serial::SerialStream::open(&builder) + .map_err(|e| anyhow::anyhow!("failed to open {port}: {e}"))?; + + let ping_frame = encode_frame(&[0]); + AsyncWriteExt::write_all(&mut serial, &ping_frame) + .await + .map_err(|e| anyhow::anyhow!("failed to write ping: {e}"))?; + + let mut reader = FrameReader::new(); + let mut buf = [0u8; 64]; + let deadline = tokio::time::Instant::now() + Duration::from_secs(2); + + loop { + let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); + if remaining.is_zero() { + anyhow::bail!("ping timeout"); + } + + let n = tokio::select! { + result = serial.read(&mut buf) => { + result.map_err(|e| anyhow::anyhow!("serial read error: {e}"))? + } + () = tokio::time::sleep(remaining) => { + anyhow::bail!("ping timeout"); + } + }; + + if n == 0 { + anyhow::bail!("serial port closed during ping"); + } + + for frame in reader.feed(&buf[..n]) { + if frame.first() == Some(&0) { + info!("dongle responded to Ping"); + return Ok(serial); + } + } + } +} + +async fn dongle_reader( + mut serial: tokio::io::ReadHalf, + state: SharedState, + pending_response: Arc>>>>, + dongle_lost: CancellationToken, + shutdown: CancellationToken, +) { + let mut reader = FrameReader::new(); + let mut buf = [0u8; 512]; + let mut consecutive_errors: u32 = 0; + + loop { + if shutdown.is_cancelled() || dongle_lost.is_cancelled() { + return; + } + + let read_result = tokio::select! { + result = serial.read(&mut buf) => result, + () = shutdown.cancelled() => return, + () = dongle_lost.cancelled() => return, + }; + + match read_result { + Ok(0) => { + error!("serial port closed"); + resolve_pending_with_error(&pending_response).await; + dongle_lost.cancel(); + return; + } + Ok(n) => { + consecutive_errors = 0; + for frame in reader.feed(&buf[..n]) { + dispatch_frame(&frame, &state, &pending_response).await; + } + } + Err(e) => { + consecutive_errors += 1; + if consecutive_errors >= MAX_CONSECUTIVE_ERRORS { + error!("dongle read error (giving up after {consecutive_errors}): {e}"); + resolve_pending_with_error(&pending_response).await; + dongle_lost.cancel(); + return; + } + warn!("dongle read glitch ({consecutive_errors}/{MAX_CONSECUTIVE_ERRORS}): {e}"); + tokio::time::sleep(Duration::from_millis(100)).await; + } + } + } +} + +async fn dispatch_frame( + frame: &[u8], + state: &SharedState, + pending_response: &Arc>>>>, +) { + let Some(&tag) = frame.first() else { + return; + }; + + if tag == RESP_TAG_RX_PACKET { + let cobs_frame = encode_frame(frame); + let inner = state.lock().await; + for session in inner.sessions.values() { + session.enqueue(cobs_frame.clone()); + } + } else { + let mut slot = pending_response.lock().await; + if let Some(sender) = slot.take() { + let _ = sender.send(frame.to_vec()); + } else { + warn!("unsolicited non-RxPacket response (tag {tag}) — dropped"); + } + } +} + +async fn resolve_pending_with_error( + pending_response: &Arc>>>>, +) { + let mut slot = pending_response.lock().await; + if let Some(sender) = slot.take() { + let _ = sender.send(vec![5, 1]); // Error(RadioBusy) + } +} + +async fn dongle_writer( + mut serial: tokio::io::WriteHalf, + cmd_rx: SharedCmdRx, + state: SharedState, + pending_response: Arc>>>>, + dongle_lost: CancellationToken, + shutdown: CancellationToken, +) { + loop { + // Pull next command from queue + let (client_id, raw_cmd) = { + let mut rx = cmd_rx.lock().await; + tokio::select! { + cmd = rx.recv() => { + match cmd { + Some(c) => c, + None => return, + } + } + () = shutdown.cancelled() => return, + () = dongle_lost.cancelled() => return, + } + }; + + // Check for interception + let intercepted = { + let mut inner = state.lock().await; + let locked = inner.mux_state.locked_config; + intercept::maybe_intercept( + &raw_cmd, + client_id, + &mut inner.sessions, + &locked, + ) + }; + if let Some(synthetic) = intercepted { + let inner = state.lock().await; + if let Some(session) = inner.sessions.get(&client_id) { + session.enqueue(encode_frame(&synthetic)); + } + continue; + } + + // Install oneshot for the response + let (resp_tx, resp_rx) = oneshot::channel(); + { + let mut slot = pending_response.lock().await; + *slot = Some(resp_tx); + } + + // Send command to dongle + let cobs_frame = encode_frame(&raw_cmd); + if let Err(e) = AsyncWriteExt::write_all(&mut serial, &cobs_frame).await { + error!("dongle write error: {e}"); + resolve_pending_with_error(&pending_response).await; + dongle_lost.cancel(); + return; + } + + // Wait for response + let response = match tokio::time::timeout(COMMAND_TIMEOUT, resp_rx).await { + Ok(Ok(resp)) => resp, + Ok(Err(_)) => { + warn!("response channel dropped"); + vec![5, 1] + } + Err(_) => { + error!("dongle response timeout"); + let mut slot = pending_response.lock().await; + *slot = None; + vec![5, 1] + } + }; + + // Track state on successful forward + let cmd_tag = raw_cmd.first().copied().unwrap_or(0xFF); + let resp_tag = response.first().copied().unwrap_or(0xFF); + { + let mut inner = state.lock().await; + if cmd_tag == CMD_TAG_SET_CONFIG && resp_tag == RESP_TAG_OK { + if let Some(config_bytes) = raw_cmd.get(1..1 + RADIO_CONFIG_SIZE) { + let mut locked = [0u8; RADIO_CONFIG_SIZE]; + locked.copy_from_slice(config_bytes); + inner.mux_state.locked_config = Some(locked); + info!("radio config locked"); + } + } else if cmd_tag == CMD_TAG_START_RX && resp_tag == RESP_TAG_OK { + if let Some(session) = inner.sessions.get_mut(&client_id) { + session.rx_interested = true; + } + } else if cmd_tag == CMD_TAG_STOP_RX + && resp_tag == RESP_TAG_OK + && let Some(session) = inner.sessions.get_mut(&client_id) + { + session.rx_interested = false; + } + } + + // Route response to commanding client + if client_id != SYNTHETIC_CLIENT_ID { + let inner = state.lock().await; + if let Some(session) = inner.sessions.get(&client_id) { + session.enqueue(encode_frame(&response)); + } + } + } +} + +// ── Client management ────────────────────────────────────────────── + +async fn accept_unix_clients( + listener: UnixListener, + state: SharedState, + cmd_tx: mpsc::Sender<(u64, Vec)>, + shutdown: CancellationToken, +) { + loop { + let accept_result = tokio::select! { + result = listener.accept() => result, + () = shutdown.cancelled() => return, + }; + + match accept_result { + Ok((stream, _addr)) => { + let (read_half, write_half) = stream.into_split(); + spawn_client(read_half, write_half, state.clone(), cmd_tx.clone(), shutdown.clone()); + } + Err(e) => warn!("Unix accept error: {e}"), + } + } +} + +async fn accept_tcp_clients( + listener: TcpListener, + state: SharedState, + cmd_tx: mpsc::Sender<(u64, Vec)>, + shutdown: CancellationToken, +) { + loop { + let accept_result = tokio::select! { + result = listener.accept() => result, + () = shutdown.cancelled() => return, + }; + + match accept_result { + Ok((stream, addr)) => { + debug!("TCP connection from {addr}"); + let (read_half, write_half) = stream.into_split(); + spawn_client(read_half, write_half, state.clone(), cmd_tx.clone(), shutdown.clone()); + } + Err(e) => warn!("TCP accept error: {e}"), + } + } +} + +fn spawn_client( + read_half: R, + write_half: W, + state: SharedState, + cmd_tx: mpsc::Sender<(u64, Vec)>, + shutdown: CancellationToken, +) where + R: AsyncReadExt + Unpin + Send + 'static, + W: AsyncWriteExt + Unpin + Send + 'static, +{ + let (session, send_rx) = ClientSession::new(); + let client_id = session.id; + let label = session.label(); + + let state_clone = state.clone(); + let cmd_tx_clone = cmd_tx.clone(); + + tokio::spawn(async move { + { + let mut inner = state_clone.lock().await; + inner.sessions.insert(client_id, session); + info!("{label} connected ({} total)", inner.sessions.len()); + } + + let writer_shutdown = shutdown.clone(); + let writer_handle = tokio::spawn(client_writer(write_half, send_rx, writer_shutdown)); + + client_reader(read_half, client_id, cmd_tx_clone.clone(), shutdown).await; + + writer_handle.abort(); + let _ = writer_handle.await; + + remove_client(client_id, &state_clone, &label, &cmd_tx_clone).await; + }); +} + +async fn client_reader( + mut read_half: R, + client_id: u64, + cmd_tx: mpsc::Sender<(u64, Vec)>, + shutdown: CancellationToken, +) { + let mut reader = FrameReader::new(); + let mut buf = [0u8; 4096]; + + loop { + let read_result = tokio::select! { + result = read_half.read(&mut buf) => result, + () = shutdown.cancelled() => return, + }; + + match read_result { + Ok(0) => return, + Ok(n) => { + for frame in reader.feed(&buf[..n]) { + if cmd_tx.send((client_id, frame)).await.is_err() { + return; + } + } + } + Err(_) => return, + } + } +} + +async fn client_writer( + mut write_half: W, + mut send_rx: mpsc::Receiver>, + shutdown: CancellationToken, +) { + loop { + let frame = tokio::select! { + f = send_rx.recv() => { + match f { + Some(f) => f, + None => return, + } + } + () = shutdown.cancelled() => return, + }; + + if AsyncWriteExt::write_all(&mut write_half, &frame) + .await + .is_err() + { + return; + } + } +} + +async fn remove_client( + client_id: u64, + state: &SharedState, + label: &str, + cmd_tx: &mpsc::Sender<(u64, Vec)>, +) { + let mut inner = state.lock().await; + + let was_interested = inner + .sessions + .get(&client_id) + .is_some_and(|s| s.rx_interested); + + inner.sessions.remove(&client_id); + info!("{label} disconnected ({} remain)", inner.sessions.len()); + + // If last RX-interested client gone, send synthetic StopRx + if was_interested && intercept::rx_interest_count(&inner.sessions) == 0 { + info!("last RX-interested client gone — sending StopRx"); + // Drop the lock before sending to avoid deadlock + drop(inner); + let _ = cmd_tx + .send((SYNTHETIC_CLIENT_ID, vec![CMD_TAG_STOP_RX])) + .await; + return; + } + + // Reset locked config when all clients are gone + if inner.sessions.is_empty() && inner.mux_state.locked_config.is_some() { + info!("all clients gone — config lock released"); + inner.mux_state.locked_config = None; + } +} diff --git a/src/intercept.rs b/src/intercept.rs new file mode 100644 index 0000000..8e62643 --- /dev/null +++ b/src/intercept.rs @@ -0,0 +1,291 @@ +//! Smart multiplexing: SetConfig locking, StartRx/StopRx reference counting. +//! +//! The mux intercepts certain commands to avoid redundant dongle operations +//! when multiple clients are connected. Intercepted commands get synthetic +//! responses without hitting the physical dongle. + +use std::collections::HashMap; + +use donglora_client::{ + CMD_TAG_SET_CONFIG, CMD_TAG_START_RX, CMD_TAG_STOP_RX, ERROR_INVALID_CONFIG, RADIO_CONFIG_SIZE, + RESP_TAG_ERROR, RESP_TAG_OK, +}; +use tracing::{debug, warn}; + +use crate::session::ClientSession; + +/// Mux-level state for interception logic. +pub struct MuxState { + /// Raw 13-byte RadioConfig, locked once set by any client with others connected. + pub locked_config: Option<[u8; RADIO_CONFIG_SIZE]>, +} + +impl MuxState { + pub fn new() -> Self { + Self { + locked_config: None, + } + } +} + +/// Check if a command should be intercepted (synthetic response) vs forwarded to the dongle. +/// +/// Returns `Some(response_bytes)` if intercepted, `None` to forward normally. +pub fn maybe_intercept( + raw_cmd: &[u8], + client_id: u64, + sessions: &mut HashMap, + locked_config: &Option<[u8; RADIO_CONFIG_SIZE]>, +) -> Option> { + let cmd_tag = *raw_cmd.first()?; + + match cmd_tag { + CMD_TAG_SET_CONFIG => intercept_set_config(raw_cmd, client_id, sessions, locked_config), + CMD_TAG_START_RX => intercept_start_rx(client_id, sessions), + CMD_TAG_STOP_RX => intercept_stop_rx(client_id, sessions), + _ => None, + } +} + +fn intercept_set_config( + raw_cmd: &[u8], + client_id: u64, + sessions: &HashMap, + locked_config: &Option<[u8; RADIO_CONFIG_SIZE]>, +) -> Option> { + let config_bytes = raw_cmd.get(1..1 + RADIO_CONFIG_SIZE)?; + + // Single client — allow free config changes (scanner mode) + if sessions.len() <= 1 { + return None; + } + + // First SetConfig with multiple clients — forward and lock on success + let Some(locked) = locked_config else { + return None; + }; + + // Same config — reply Ok without hitting the dongle + if config_bytes == locked.as_slice() { + let label = sessions + .get(&client_id) + .map(ClientSession::label) + .unwrap_or_else(|| format!("id-{client_id}")); + debug!("{label}: SetConfig matches locked config — Ok"); + return Some(vec![RESP_TAG_OK]); + } + + // Different config — reject + let label = sessions + .get(&client_id) + .map(ClientSession::label) + .unwrap_or_else(|| format!("id-{client_id}")); + warn!("{label}: SetConfig rejected (conflicts with locked config)"); + Some(vec![RESP_TAG_ERROR, ERROR_INVALID_CONFIG]) +} + +fn intercept_start_rx( + client_id: u64, + sessions: &mut HashMap, +) -> Option> { + let client = sessions.get(&client_id)?; + + // Already interested — no-op + if client.rx_interested { + return Some(vec![RESP_TAG_OK]); + } + + // Others already receiving — just mark this client + if rx_interest_count(sessions) > 0 { + if let Some(c) = sessions.get_mut(&client_id) { + c.rx_interested = true; + } + return Some(vec![RESP_TAG_OK]); + } + + // First interested client — forward to dongle + None +} + +fn intercept_stop_rx( + client_id: u64, + sessions: &mut HashMap, +) -> Option> { + let client = sessions.get(&client_id)?; + + // Wasn't interested — no-op + if !client.rx_interested { + return Some(vec![RESP_TAG_OK]); + } + + // Mark as not interested + if let Some(c) = sessions.get_mut(&client_id) { + c.rx_interested = false; + } + + // Others still interested — don't stop the dongle + if rx_interest_count(sessions) > 0 { + return Some(vec![RESP_TAG_OK]); + } + + // Last interested client — forward StopRx to dongle + None +} + +/// Count how many sessions have rx_interested set. +pub fn rx_interest_count(sessions: &HashMap) -> usize { + sessions.values().filter(|s| s.rx_interested).count() +} + +// ── Tests ────────────────────────────────────────────────────────── + +#[cfg(test)] +mod tests { + use super::*; + + fn make_sessions(n: usize) -> HashMap { + let mut map = HashMap::new(); + for _ in 0..n { + let (session, _rx) = ClientSession::new(); + let id = session.id; + map.insert(id, session); + } + map + } + + fn first_id(sessions: &HashMap) -> u64 { + *sessions.keys().min().unwrap() + } + + fn second_id(sessions: &HashMap) -> u64 { + let mut ids: Vec = sessions.keys().copied().collect(); + ids.sort(); + ids[1] + } + + // ── SetConfig ──────────────────────────────���─────────────────── + + #[test] + fn set_config_single_client_forwards() { + let mut sessions = make_sessions(1); + let locked = None; + let id = first_id(&sessions); + let cmd = [CMD_TAG_SET_CONFIG, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]; + assert!(maybe_intercept(&cmd, id, &mut sessions, &locked).is_none()); + } + + #[test] + fn set_config_first_with_multiple_forwards() { + let mut sessions = make_sessions(2); + let locked = None; + let id = first_id(&sessions); + let cmd = [CMD_TAG_SET_CONFIG, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13]; + assert!(maybe_intercept(&cmd, id, &mut sessions, &locked).is_none()); + } + + #[test] + fn set_config_matching_returns_ok() { + let mut sessions = make_sessions(2); + let config = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13]; + let locked = Some(config); + let id = first_id(&sessions); + let mut cmd = vec![CMD_TAG_SET_CONFIG]; + cmd.extend_from_slice(&config); + assert_eq!(maybe_intercept(&cmd, id, &mut sessions, &locked), Some(vec![RESP_TAG_OK])); + } + + #[test] + fn set_config_conflicting_returns_error() { + let mut sessions = make_sessions(2); + let locked = Some([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13]); + let id = first_id(&sessions); + let mut cmd = vec![CMD_TAG_SET_CONFIG]; + cmd.extend_from_slice(&[99, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13]); + assert_eq!( + maybe_intercept(&cmd, id, &mut sessions, &locked), + Some(vec![RESP_TAG_ERROR, ERROR_INVALID_CONFIG]) + ); + } + + // ── StartRx ──────────────────────────────────────────────────── + + #[test] + fn start_rx_first_client_forwards() { + let mut sessions = make_sessions(1); + let locked = None; + let id = first_id(&sessions); + assert!(maybe_intercept(&[CMD_TAG_START_RX], id, &mut sessions, &locked).is_none()); + } + + #[test] + fn start_rx_already_interested_returns_ok() { + let mut sessions = make_sessions(1); + let locked = None; + let id = first_id(&sessions); + sessions.get_mut(&id).unwrap().rx_interested = true; + assert_eq!( + maybe_intercept(&[CMD_TAG_START_RX], id, &mut sessions, &locked), + Some(vec![RESP_TAG_OK]) + ); + } + + #[test] + fn start_rx_others_interested_marks_and_returns_ok() { + let mut sessions = make_sessions(2); + let locked = None; + let id1 = first_id(&sessions); + let id2 = second_id(&sessions); + sessions.get_mut(&id1).unwrap().rx_interested = true; + let result = maybe_intercept(&[CMD_TAG_START_RX], id2, &mut sessions, &locked); + assert_eq!(result, Some(vec![RESP_TAG_OK])); + assert!(sessions.get(&id2).unwrap().rx_interested); + } + + // ── StopRx ───────────────────────────────────────────────────── + + #[test] + fn stop_rx_not_interested_returns_ok() { + let mut sessions = make_sessions(1); + let locked = None; + let id = first_id(&sessions); + assert_eq!( + maybe_intercept(&[CMD_TAG_STOP_RX], id, &mut sessions, &locked), + Some(vec![RESP_TAG_OK]) + ); + } + + #[test] + fn stop_rx_others_remain_returns_ok() { + let mut sessions = make_sessions(2); + let locked = None; + let id1 = first_id(&sessions); + let id2 = second_id(&sessions); + sessions.get_mut(&id1).unwrap().rx_interested = true; + sessions.get_mut(&id2).unwrap().rx_interested = true; + let result = maybe_intercept(&[CMD_TAG_STOP_RX], id1, &mut sessions, &locked); + assert_eq!(result, Some(vec![RESP_TAG_OK])); + assert!(!sessions.get(&id1).unwrap().rx_interested); + } + + #[test] + fn stop_rx_last_interested_forwards() { + let mut sessions = make_sessions(1); + let locked = None; + let id = first_id(&sessions); + sessions.get_mut(&id).unwrap().rx_interested = true; + assert!(maybe_intercept(&[CMD_TAG_STOP_RX], id, &mut sessions, &locked).is_none()); + assert!(!sessions.get(&id).unwrap().rx_interested); + } + + // ── Other commands ───────────────────────────────────────────── + + #[test] + fn other_commands_not_intercepted() { + let mut sessions = make_sessions(1); + let locked = None; + let id = first_id(&sessions); + for tag in [0u8, 1, 5, 6, 7, 8] { + assert!(maybe_intercept(&[tag], id, &mut sessions, &locked).is_none()); + } + } +} diff --git a/src/main.rs b/src/main.rs new file mode 100644 index 0000000..5f7562f --- /dev/null +++ b/src/main.rs @@ -0,0 +1,119 @@ +//! DongLoRa USB Multiplexer — share one dongle with multiple applications. +//! +//! Owns the USB serial connection and exposes a Unix domain socket (and +//! optional TCP) that speaks the same COBS-framed protocol. Clients connect +//! and get solicited responses routed, RxPackets broadcast, and StartRx/StopRx +//! reference-counted. + +mod daemon; +mod intercept; +mod session; + +use clap::Parser; +use tokio_util::sync::CancellationToken; +use tracing::info; + +#[derive(Parser)] +#[command(name = "donglora-mux", about = "DongLoRa USB Multiplexer")] +struct Args { + /// Serial port (auto-detect by USB VID:PID if omitted) + #[arg(short, long)] + port: Option, + + /// Unix socket path (default: XDG or /tmp/donglora-mux.sock) + #[arg(short, long)] + socket: Option, + + /// TCP listen address [HOST:]PORT (default: 0.0.0.0:5741, "none" to disable) + #[arg(short, long, default_value = "0.0.0.0:5741")] + tcp: String, + + /// Enable debug logging + #[arg(short, long)] + verbose: bool, +} + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + let args = Args::parse(); + + // Init tracing — daemon-style: wall-clock time, level, message + let filter = if args.verbose { "debug" } else { "info" }; + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new(filter)), + ) + .with_target(false) + .init(); + + // Resolve serial port + let port = match args.port { + Some(p) => p, + None => { + tokio::task::spawn_blocking(|| { + donglora_client::find_port().unwrap_or_else(|| { + info!("waiting for DongLoRa device..."); + donglora_client::wait_for_device() + }) + }) + .await + .map_err(|e| anyhow::anyhow!("device discovery task failed: {e}"))? + } + }; + + // Resolve socket path + let socket_path = args + .socket + .unwrap_or_else(donglora_client::default_socket_path); + + // Parse TCP address + let tcp_addr = parse_tcp_addr(&args.tcp); + + // Shutdown token + let shutdown = CancellationToken::new(); + + // Install signal handlers + let sig_shutdown = shutdown.clone(); + tokio::spawn(async move { + if let Err(e) = signal_handler(sig_shutdown).await { + tracing::error!("signal handler error: {e}"); + } + }); + + // Run the daemon + let daemon = daemon::MuxDaemon::new(port, socket_path, tcp_addr, shutdown); + daemon.run().await +} + +fn parse_tcp_addr(tcp: &str) -> Option<(String, u16)> { + if tcp.eq_ignore_ascii_case("none") { + return None; + } + if let Some((host, port_str)) = tcp.rsplit_once(':') + && let Ok(port) = port_str.parse() + { + return Some((host.to_string(), port)); + } + if let Ok(port) = tcp.parse() { + return Some(("0.0.0.0".to_string(), port)); + } + None +} + +async fn signal_handler(shutdown: CancellationToken) -> anyhow::Result<()> { + let mut sigint = + tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt()) + .map_err(|e| anyhow::anyhow!("failed to register SIGINT handler: {e}"))?; + let mut sigterm = + tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) + .map_err(|e| anyhow::anyhow!("failed to register SIGTERM handler: {e}"))?; + + tokio::select! { + _ = sigint.recv() => info!("received SIGINT"), + _ = sigterm.recv() => info!("received SIGTERM"), + } + + shutdown.cancel(); + Ok(()) +} diff --git a/src/session.rs b/src/session.rs new file mode 100644 index 0000000..55a1ce5 --- /dev/null +++ b/src/session.rs @@ -0,0 +1,57 @@ +//! Per-client session state and bounded send queue. +//! +//! Each connected client gets a [`ClientSession`] that tracks its ID, +//! RX interest flag, and a bounded channel for outgoing COBS frames. + +use std::sync::atomic::{AtomicU64, Ordering}; + +use tokio::sync::mpsc; +use tracing::warn; + +/// Bounded send queue capacity per client. +const SEND_QUEUE_CAP: usize = 256; + +/// Global client ID counter. +static NEXT_CLIENT_ID: AtomicU64 = AtomicU64::new(0); + +/// Per-client state tracked by the mux daemon. +pub struct ClientSession { + /// Unique client identifier. + pub id: u64, + /// Whether this client has called StartRx. + pub rx_interested: bool, + /// Sender half of the bounded channel to the client's writer task. + tx: mpsc::Sender>, +} + +impl ClientSession { + /// Create a new session. Returns the session and the receiver half + /// for the client's writer task to drain. + pub fn new() -> (Self, mpsc::Receiver>) { + let id = NEXT_CLIENT_ID.fetch_add(1, Ordering::Relaxed); + let (tx, rx) = mpsc::channel(SEND_QUEUE_CAP); + let session = Self { + id, + rx_interested: false, + tx, + }; + (session, rx) + } + + /// Human-readable label for logging. + pub fn label(&self) -> String { + format!("client-{}", self.id) + } + + /// Best-effort enqueue of a COBS frame for sending to this client. + /// + /// If the channel is full (slow client), the frame is dropped with a warning. + /// This prevents backpressure from one slow client blocking the entire mux. + pub fn enqueue(&self, frame: Vec) { + if let Err(mpsc::error::TrySendError::Full(_)) = self.tx.try_send(frame) { + warn!("{}: send queue full, dropping frame", self.label()); + } + // Closed channel (client disconnected) is silently ignored — the session + // will be cleaned up when the client handler task exits. + } +}