Skip to content

Support multiple connections in the client mode. #2320

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
OlivierHecart merged 15 commits into from
May 20, 2026
Merged

Support multiple connections in the client mode. #2320

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
6000ab0
Support multiple connections in the client mode.
evshary Dec 17, 2025
d764658
Fix clippy issue.
evshary Dec 17, 2025
cfe752d
Support no-std
evshary Dec 17, 2025
d8ef67d
Don't check the address.
evshary Dec 18, 2025
940852f
Able to accept multiple connections from CLI
evshary Dec 18, 2025
6688c80
Add tests for the client mode config.
evshary Dec 18, 2025
d9095ba
Add the usage inside the config file.
evshary Jan 6, 2026
43854bc
Merge branch 'main' into support_multiclient
evshary Jan 14, 2026
d1abb0d
Revert the change in examples behaviors.
evshary Jan 15, 2026
4316d55
Update the connect config format. Support object instead of array.
evshary Jan 15, 2026
83cd651
Fix the Rust format issue
evshary Jan 15, 2026
a46b494
merge: sync main into support_multiclient
evshary May 1, 2026
2405e55
test(connectivity): cover allOf locator groups
evshary May 4, 2026
4f67d98
Merge branch 'main' into support_multiclient
evshary May 13, 2026
d4d7762
Merge branch 'main' into support_multiclient
evshary May 14, 2026
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
36 changes: 32 additions & 4 deletions commons/zenoh-config/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,8 @@ pub mod wrappers;

#[allow(unused_imports)]
use std::convert::TryFrom;
#[allow(unused_imports)]
use std::str::FromStr;
// This is a false positive from the rust analyser
use std::{
any::Any,
Expand All @@ -51,7 +53,7 @@ use validated_struct::ValidatedMapAssociatedTypes;
pub use validated_struct::{GetError, ValidatedMap};
pub use wrappers::ZenohId;
pub use zenoh_protocol::core::{
whatami, EndPoint, Locator, WhatAmI, WhatAmIMatcher, WhatAmIMatcherVisitor,
whatami, EndPoint, EndPoints, Locator, WhatAmI, WhatAmIMatcher, WhatAmIMatcherVisitor,
};
use zenoh_protocol::{
core::{
Expand Down Expand Up @@ -446,8 +448,12 @@ pub fn peer() -> Config {
pub fn client<I: IntoIterator<Item = T>, T: Into<EndPoint>>(peers: I) -> Config {
let mut config = Config::default();
config.set_mode(Some(WhatAmI::Client)).unwrap();
config.connect.endpoints =
ModeDependentValue::Unique(peers.into_iter().map(|t| t.into()).collect());
config.connect.endpoints = ModeDependentValue::Unique(
peers
.into_iter()
.map(|t| EndPoints::Single(t.into()))
.collect(),
);
config
}

Expand Down Expand Up @@ -478,7 +484,7 @@ validated_struct::validator! {
/// global timeout for full connect cycle
pub timeout_ms: Option<ModeDependentValue<i64>>,
/// The list of endpoints to connect to
pub endpoints: ModeDependentValue<Vec<EndPoint>>,
pub endpoints: ModeDependentValue<Vec<EndPoints>>,
/// if connection timeout exceed, exit from application
pub exit_on_failure: Option<ModeDependentValue<bool>>,
pub retry: Option<connection_retry::ConnectionRetryModeDependentConf>,
Expand Down Expand Up @@ -1197,6 +1203,28 @@ fn config_deser() {
})
);

let config = Config::from_deserializer(
&mut json5::Deserializer::from_str(
r#"{
mode: "client",
connect: {
endpoints: [
["tcp/127.0.0.1:7447?rel=0","tcp/127.0.0.1:7448?rel=1"],
]
}
}"#,
)
.unwrap(),
)
.unwrap();
assert_eq!(*config.mode(), Some(WhatAmI::Client));
let endpoints = config.connect().endpoints().client().unwrap();
assert_eq!(endpoints.len(), 1);
assert_eq!(
endpoints[0],
EndPoints::from_str("[tcp/127.0.0.1:7447?rel=0,tcp/127.0.0.1:7448?rel=1]").unwrap(),
);

dbg!(Config::from_file("../../DEFAULT_CONFIG.json5").unwrap());
}

Expand Down
40 changes: 39 additions & 1 deletion commons/zenoh-config/src/mode_dependent.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@ use serde::{
de::{self, IntoDeserializer, MapAccess, Visitor},
Deserialize, Serialize,
};
use zenoh_protocol::core::{EndPoint, WhatAmI, WhatAmIMatcher, WhatAmIMatcherVisitor};
use zenoh_protocol::core::{EndPoint, EndPoints, WhatAmI, WhatAmIMatcher, WhatAmIMatcherVisitor};

use crate::AutoConnectStrategy;

Expand Down Expand Up @@ -331,6 +331,44 @@ impl<'a> serde::Deserialize<'a> for ModeDependentValue<Vec<EndPoint>> {
}
}

impl<'a> serde::Deserialize<'a> for ModeDependentValue<Vec<EndPoints>> {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'a>,
{
struct UniqueOrDependent<U>(PhantomData<fn() -> U>);

impl<'de> Visitor<'de> for UniqueOrDependent<ModeDependentValue<Vec<EndPoints>>> {
type Value = ModeDependentValue<Vec<EndPoints>>;

fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
formatter.write_str("list of endpoints or mode dependent list of endpoints")
}

fn visit_seq<A>(self, mut seq: A) -> Result<Self::Value, A::Error>
where
A: de::SeqAccess<'de>,
{
let mut v = seq.size_hint().map_or_else(Vec::new, Vec::with_capacity);

while let Some(s) = seq.next_element()? {
v.push(s);
}
Ok(ModeDependentValue::Unique(v))
}

fn visit_map<M>(self, map: M) -> Result<Self::Value, M::Error>
where
M: MapAccess<'de>,
{
ModeValues::deserialize(de::value::MapAccessDeserializer::new(map))
.map(ModeDependentValue::Dependent)
}
}
deserializer.deserialize_any(UniqueOrDependent(PhantomData))
}
}

impl<T> ModeDependent<T> for Option<ModeDependentValue<T>> {
#[inline]
fn router(&self) -> Option<&T> {
Expand Down
113 changes: 112 additions & 1 deletion commons/zenoh-protocol/src/core/endpoint.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
// Contributors:
// ZettaScale Zenoh Team, <zenoh@zettascale.tech>
//
use alloc::{borrow::ToOwned, format, string::String};
use alloc::{borrow::ToOwned, format, string::String, vec, vec::Vec};
use core::{borrow::Borrow, convert::TryFrom, fmt, str::FromStr};

use zenoh_result::{bail, zerror, Error as ZError, ZResult};
Expand Down Expand Up @@ -696,8 +696,119 @@ impl EndPoint {
}
}

#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
pub enum EndPoints {
Single(EndPoint),
Vec(Vec<EndPoint>),
}
impl EndPoints {
pub fn flatten(self) -> Vec<EndPoint> {
match self {
EndPoints::Single(ep) => vec![ep],
EndPoints::Vec(eps) => eps,
}
}

pub fn as_vec(&self) -> Vec<EndPoint> {
match self {
EndPoints::Single(ep) => vec![ep.clone()],
EndPoints::Vec(eps) => eps.clone(),
}
}
}

impl From<EndPoint> for EndPoints {
fn from(ep: EndPoint) -> EndPoints {
EndPoints::Single(ep)
}
}

impl<'de> serde::Deserialize<'de> for EndPoints {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
struct EndPointsVisitor;

impl<'de> serde::de::Visitor<'de> for EndPointsVisitor {
type Value = EndPoints;

fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
formatter.write_str("a single endpoint string or a list of endpoint strings")
}

fn visit_str<E>(self, v: &str) -> Result<Self::Value, E>
where
E: serde::de::Error,
{
EndPoint::from_str(v)
.map(EndPoints::Single)
.map_err(serde::de::Error::custom)
}

fn visit_seq<A>(self, mut seq: A) -> Result<Self::Value, A::Error>
where
A: serde::de::SeqAccess<'de>,
{
let mut vec = Vec::new();
while let Some(elem) = seq.next_element::<EndPoint>()? {
vec.push(elem);
}
Ok(EndPoints::Vec(vec))
}
}

deserializer.deserialize_any(EndPointsVisitor)
}
}

impl TryFrom<String> for EndPoints {
type Error = ZError;

fn try_from(s: String) -> Result<Self, Self::Error> {
const ERR: &str =
"Endpoints must be of the form <endpoint> or [<endpoint>, <endpoint>, ...]";
if s.starts_with('[') && s.ends_with(']') {
let eps: ZResult<Vec<EndPoint>> = s[1..s.len() - 1]
.split(',')
.map(|x| EndPoint::from_str(x.trim()))
.collect();
eps.map(EndPoints::Vec)
} else {
EndPoint::from_str(s.as_str())
.map(EndPoints::Single)
.map_err(|e| zerror!("{}: {}", ERR, e).into())
}
}
}

impl FromStr for EndPoints {
type Err = ZError;

fn from_str(s: &str) -> Result<Self, Self::Err> {
Self::try_from(s.to_owned())
}
}

#[test]
fn endpoints() {
// Single
assert_eq!(
EndPoints::from_str("udp/127.0.0.1:7447").unwrap(),
EndPoints::Single(EndPoint::from_str("udp/127.0.0.1:7447").unwrap())
);
// Vec
assert_eq!(
EndPoints::from_str("[udp/127.0.0.1:7447?rel=0,udp/127.0.0.1:7447?rel=1]").unwrap(),
EndPoints::Vec(vec![
EndPoint::from_str("udp/127.0.0.1:7447?rel=0").unwrap(),
EndPoint::from_str("udp/127.0.0.1:7447?rel=1").unwrap()
])
);
}

#[test]
fn endpoint() {
assert!(EndPoint::from_str("/").is_err());
assert!(EndPoint::from_str("?").is_err());
assert!(EndPoint::from_str("#").is_err());
Expand Down
8 changes: 7 additions & 1 deletion examples/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,8 +55,14 @@ impl From<&CommonArgs> for Config {
}

if !args.connect.is_empty() {
// Able to parse multiple endpoints, e.g. "tcp/127.0.0.1:7447?rel=0,tcp/127.0.0.1:7447?rel=1"

@OlivierHecart OlivierHecart Dec 19, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is a change in behavior.
Coma separated values were already accepted here and were the equivalent of passing multiple -e arguments.
Let's decide if:

  • we assume this behavior change and add this in a minor release
  • find another separator for this case

@evshary evshary Dec 19, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I personally prefer the first one, while the second one will make usage complicated.
However, it's still fine with me if we decide on the second one.

let endpoints: Vec<Vec<String>> = args
.connect
.iter()
.map(|s| s.split(',').map(String::from).collect())
.collect();
config
.insert_json5("connect/endpoints", &json!(args.connect).to_string())
.insert_json5("connect/endpoints", &json!(endpoints).to_string())
.unwrap();
}
if !args.listen.is_empty() {
Expand Down
28 changes: 14 additions & 14 deletions zenoh-ext/tests/advanced.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
//

use zenoh::{key_expr::OwnedNonWildKeyExpr, sample::SampleKind};
use zenoh_config::{EndPoint, ModeDependentValue, WhatAmI};
use zenoh_config::{EndPoint, EndPoints, ModeDependentValue, WhatAmI};
use zenoh_ext::{
AdvancedPublisherBuilderExt, AdvancedSubscriberBuilderExt, CacheConfig, HistoryConfig,
MissDetectionConfig, RecoveryConfig,
Expand Down Expand Up @@ -68,7 +68,7 @@ async fn test_advanced_history_inner(
let mut c = zenoh::Config::default();
c.connect
.endpoints
.set(vec![endpoint.parse::<EndPoint>().unwrap()])
.set(vec![endpoint.parse::<EndPoints>().unwrap()])
.unwrap();
c.scouting.multicast.set_enabled(Some(false)).unwrap();
c.namespace = sub_namespace;
Expand Down Expand Up @@ -182,7 +182,7 @@ async fn test_advanced_retransmission_inner(
let mut c = zenoh::Config::default();
c.connect
.endpoints
.set(vec![endpoint.parse::<EndPoint>().unwrap()])
.set(vec![endpoint.parse::<EndPoints>().unwrap()])
.unwrap();
c.scouting.multicast.set_enabled(Some(false)).unwrap();
c.namespace = pub_namespace;
Expand All @@ -196,7 +196,7 @@ async fn test_advanced_retransmission_inner(
let mut c = zenoh::Config::default();
c.connect
.endpoints
.set(vec![endpoint.parse::<EndPoint>().unwrap()])
.set(vec![endpoint.parse::<EndPoints>().unwrap()])
.unwrap();
c.scouting.multicast.set_enabled(Some(false)).unwrap();
c.namespace = sub_namespace;
Expand Down Expand Up @@ -351,7 +351,7 @@ async fn test_advanced_retransmission_periodic_inner(
let mut c = zenoh::Config::default();
c.connect
.endpoints
.set(vec![endpoint.parse::<EndPoint>().unwrap()])
.set(vec![endpoint.parse::<EndPoints>().unwrap()])
.unwrap();
c.scouting.multicast.set_enabled(Some(false)).unwrap();
c.namespace = pub_namespace;
Expand All @@ -365,7 +365,7 @@ async fn test_advanced_retransmission_periodic_inner(
let mut c = zenoh::Config::default();
c.connect
.endpoints
.set(vec![endpoint.parse::<EndPoint>().unwrap()])
.set(vec![endpoint.parse::<EndPoints>().unwrap()])
.unwrap();
c.scouting.multicast.set_enabled(Some(false)).unwrap();
c.namespace = sub_namespace;
Expand Down Expand Up @@ -513,7 +513,7 @@ async fn test_advanced_sample_miss_inner(
let mut c = zenoh::Config::default();
c.connect
.endpoints
.set(vec![endpoint.parse::<EndPoint>().unwrap()])
.set(vec![endpoint.parse::<EndPoints>().unwrap()])
.unwrap();
c.scouting.multicast.set_enabled(Some(false)).unwrap();
c.namespace = pub_namespace;
Expand All @@ -527,7 +527,7 @@ async fn test_advanced_sample_miss_inner(
let mut c = zenoh::Config::default();
c.connect
.endpoints
.set(vec![endpoint.parse::<EndPoint>().unwrap()])
.set(vec![endpoint.parse::<EndPoints>().unwrap()])
.unwrap();
c.scouting.multicast.set_enabled(Some(false)).unwrap();
c.namespace = sub_namespace;
Expand Down Expand Up @@ -671,7 +671,7 @@ async fn test_advanced_retransmission_sample_miss_inner(
let mut c = zenoh::Config::default();
c.connect
.endpoints
.set(vec![endpoint.parse::<EndPoint>().unwrap()])
.set(vec![endpoint.parse::<EndPoints>().unwrap()])
.unwrap();
c.scouting.multicast.set_enabled(Some(false)).unwrap();
c.namespace = pub_namespace;
Expand All @@ -685,7 +685,7 @@ async fn test_advanced_retransmission_sample_miss_inner(
let mut c = zenoh::Config::default();
c.connect
.endpoints
.set(vec![endpoint.parse::<EndPoint>().unwrap()])
.set(vec![endpoint.parse::<EndPoints>().unwrap()])
.unwrap();
c.scouting.multicast.set_enabled(Some(false)).unwrap();
c.namespace = sub_namespace;
Expand Down Expand Up @@ -826,7 +826,7 @@ async fn test_advanced_late_joiner_inner(
let mut c = zenoh::Config::default();
c.connect
.endpoints
.set(vec![endpoint.parse::<EndPoint>().unwrap()])
.set(vec![endpoint.parse::<EndPoints>().unwrap()])
.unwrap();
c.scouting.multicast.set_enabled(Some(false)).unwrap();
c.timestamping
Expand All @@ -843,7 +843,7 @@ async fn test_advanced_late_joiner_inner(
let mut c = zenoh::Config::default();
c.connect
.endpoints
.set(vec![endpoint.parse::<EndPoint>().unwrap()])
.set(vec![endpoint.parse::<EndPoints>().unwrap()])
.unwrap();
c.scouting.multicast.set_enabled(Some(false)).unwrap();
c.namespace = sub_namespace;
Expand Down Expand Up @@ -986,7 +986,7 @@ async fn test_advanced_retransmission_heartbeat_inner(
let mut c = zenoh::Config::default();
c.connect
.endpoints
.set(vec![endpoint.parse::<EndPoint>().unwrap()])
.set(vec![endpoint.parse::<EndPoints>().unwrap()])
.unwrap();
c.scouting.multicast.set_enabled(Some(false)).unwrap();
c.namespace = pub_namespace;
Expand All @@ -1000,7 +1000,7 @@ async fn test_advanced_retransmission_heartbeat_inner(
let mut c = zenoh::Config::default();
c.connect
.endpoints
.set(vec![endpoint.parse::<EndPoint>().unwrap()])
.set(vec![endpoint.parse::<EndPoints>().unwrap()])
.unwrap();
c.scouting.multicast.set_enabled(Some(false)).unwrap();
c.namespace = sub_namespace;
Expand Down
Loading