fix(native): honor LCM_DEFAULT_URL in the rust LCM transport

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

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<Self> {
-        Ok(Self::wrap(Lcm::new().await?))
+        Self::with_options(options_from_env()).await
     }
 
     pub async fn with_options(opts: LcmOptions) -> io::Result<Self> {
@@ -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}");
+        }
+    }
+}