diff --git a/Cargo.lock b/Cargo.lock index 22077fb6eb..8627032199 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -608,6 +608,7 @@ dependencies = [ "tracing", "tracing-subscriber", "tracing-test", + "url", "validator", "zenoh", ] diff --git a/native/rust/dimos-module/Cargo.toml b/native/rust/dimos-module/Cargo.toml index 2228e6c3c2..bb60a22bc9 100644 --- a/native/rust/dimos-module/Cargo.toml +++ b/native/rust/dimos-module/Cargo.toml @@ -25,6 +25,7 @@ serde = { version = "1", features = ["derive"] } serde_json = { version = "1", features = ["preserve_order"] } tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["json", "env-filter"] } +url = "2" validator = { version = "0.21", features = ["derive"] } # Default features pull in ring via QUIC/TLS, which breaks static musl cross-linking. zenoh = { version = "=1.10.0", default-features = false, features = ["transport_tcp", "transport_udp", "unstable", "shared-memory"] } diff --git a/native/rust/dimos-module/src/lcm.rs b/native/rust/dimos-module/src/lcm.rs index 5dbf0c3986..6fe3838253 100644 --- a/native/rust/dimos-module/src/lcm.rs +++ b/native/rust/dimos-module/src/lcm.rs @@ -19,6 +19,7 @@ use std::sync::{Arc, Mutex}; use std::time::Duration; use dimos_lcm::{Lcm, LcmOptions}; +use url::Url; use crate::transport::{Dispatch, Transport}; @@ -36,9 +37,55 @@ pub struct LcmTransport { runtime: tokio::runtime::Handle, } +/// liblcm reads this, so python modules follow it and native ones have to as well or the +/// two halves of a pipeline end up on different buses. Format: `udpm://group:port?ttl=n`. +fn options_from_env() -> LcmOptions { + match std::env::var("LCM_DEFAULT_URL") { + Ok(url) => options_from_url(&url), + Err(_) => LcmOptions::default(), + } +} + +fn options_from_url(url: &str) -> LcmOptions { + let mut options = LcmOptions::default(); + let parsed = match Url::parse(url) { + Ok(parsed) if parsed.scheme() == "udpm" => parsed, + _ => { + tracing::warn!( + url, + "LCM_DEFAULT_URL is not a udpm:// url; using the defaults" + ); + return options; + } + }; + // udpm is not a special scheme, so the host stays an opaque string. + let group = parsed.host_str().and_then(|host| host.parse().ok()); + match (group, parsed.port()) { + (Some(group), Some(port)) => { + options.multicast_group = group; + options.port = port; + } + _ => { + tracing::warn!( + url, + "LCM_DEFAULT_URL has no parsable group:port; using the defaults" + ); + return options; + } + } + for (key, value) in parsed.query_pairs() { + if key == "ttl" { + if let Ok(ttl) = value.parse() { + options.ttl = ttl; + } + } + } + options +} + impl LcmTransport { pub async fn new() -> io::Result { - Ok(Self::wrap(Lcm::new().await?)) + Self::with_options(options_from_env()).await } pub async fn with_options(opts: LcmOptions) -> io::Result { @@ -119,3 +166,40 @@ impl Transport for LcmTransport { } } } + +#[cfg(test)] +mod tests { + use super::options_from_url; + use std::net::Ipv4Addr; + + #[test] + fn reads_group_port_and_ttl() { + let options = options_from_url("udpm://239.255.76.67:7712?ttl=0"); + assert_eq!(options.multicast_group, Ipv4Addr::new(239, 255, 76, 67)); + assert_eq!(options.port, 7712); + assert_eq!(options.ttl, 0); + } + + #[test] + fn ttl_is_optional() { + let options = options_from_url("udpm://239.255.76.67:7712"); + assert_eq!(options.port, 7712); + assert_eq!(options.ttl, dimos_lcm::LcmOptions::default().ttl); + } + + #[test] + fn an_unusable_url_leaves_the_defaults() { + let defaults = dimos_lcm::LcmOptions::default(); + for url in [ + "tcp://127.0.0.1:7667", + "udpm://not-an-ip:7667?ttl=42", + "udpm://239.255.76.67?ttl=42", + "239.255.76.67:7667", + ] { + let options = options_from_url(url); + assert_eq!(options.multicast_group, defaults.multicast_group, "{url}"); + assert_eq!(options.port, defaults.port, "{url}"); + assert_eq!(options.ttl, defaults.ttl, "{url}"); + } + } +}