Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions native/rust/dimos-module/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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.20", features = ["derive"] }
# Default features pull in ring via QUIC/TLS, which breaks static musl cross-linking.
zenoh = { version = "=1.9.0", default-features = false, features = ["transport_tcp", "transport_udp", "unstable", "shared-memory"] }
Expand Down
82 changes: 81 additions & 1 deletion native/rust/dimos-module/src/lcm.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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};

Expand All @@ -36,9 +37,52 @@ 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"
),
}
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> {
Expand Down Expand Up @@ -119,3 +163,39 @@ 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",
"udpm://239.255.76.67",
"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}");
}
}
}
Loading