Support multiple connections in the client mode.

DEFAULT_CONFIG.json5

@@ -44,6 +44,13 @@
     /// Accepts a single list (e.g. endpoints: ["tcp/10.10.10.10:7447", "tcp/11.11.11.11:7447"])
     /// or different lists for router, peer and client (e.g. endpoints: { router: ["tcp/10.10.10.10:7447"], peer: ["tcp/11.11.11.11:7447"] }).
     ///
+    /// Note that every element in the list can also be a list:
+    /// E.g. endpoints: [{"strategy": "allOf", "locators": ["tcp/10.10.10.10:7447?rel=0", "tcp/10.10.10.10:7447?rel=1"]}, "tcp/11.11.11.11:7447"]
+    /// This can be used in the client mode.
+    /// Since the client mode only allows connecting to a single endpoint, this indicates that we want to build multiple links to the same endpoint.
+    /// It doesn't make any difference for the peer or router mode.
+    /// endpoints: [{"strategy": "allOf", "locators": ["tcp/10.10.10.10:7447?rel=0", "tcp/10.10.10.10:7447?rel=1"]}] is equivalent to endpoints: ["tcp/10.10.10.10:7447?rel=0", "tcp/10.10.10.10:7447?rel=1"].
+    ///
     /// See https://docs.rs/zenoh/latest/zenoh/config/struct.EndPoint.html
     endpoints: [
       // "<proto>/<address>"

commons/zenoh-config/src/lib.rs

@@ -29,6 +29,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,
@@ -52,7 +54,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::{
@@ -447,8 +449,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
 }
 
@@ -523,7 +529,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>,
@@ -1234,6 +1240,34 @@ fn config_deser() {
         })
     );
 
+    let config = Config::from_deserializer(
+        &mut json5::Deserializer::from_str(
+            r#"{
+                mode: "client",
+                connect: {
+                    endpoints: [
+                        { strategy: "allOf", locators: ["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::Locators(zenoh_protocol::core::Locators {
+            strategy: zenoh_protocol::core::LocatorsStrategy::AllOf,
+            locators: vec![
+                EndPoint::from_str("tcp/127.0.0.1:7447?rel=0").unwrap(),
+                EndPoint::from_str("tcp/127.0.0.1:7448?rel=1").unwrap()
+            ]
+        })
+    );
+
     dbg!(Config::from_file("../../DEFAULT_CONFIG.json5").unwrap());
 }
 

commons/zenoh-config/src/mode_dependent.rs

@@ -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;
 
@@ -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> {

commons/zenoh-protocol/Cargo.toml

@@ -53,3 +53,4 @@ zenoh-result = { workspace = true }
 [dev-dependencies]
 lazy_static = { workspace = true }
 rand = { workspace = true, features = ["default"] }
+serde_json = { workspace = true }

commons/zenoh-protocol/src/core/endpoint.rs

@@ -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};
@@ -706,8 +706,143 @@ impl EndPoint {
     }
 }
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum LocatorsStrategy {
+    /// Open links to all locators in the group.
+    AllOf,
+    /// Reserved for future support. The runtime currently only implements
+    /// `AllOf` semantics.
+    OneOf,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+pub struct Locators {
+    pub strategy: LocatorsStrategy,
+    pub locators: Vec<EndPoint>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
+#[serde(untagged)]
+pub enum EndPoints {
+    Single(EndPoint),
+    Locators(Locators),
+}
+impl EndPoints {
+    pub fn flatten(self) -> Vec<EndPoint> {
+        match self {
+            EndPoints::Single(ep) => vec![ep],
+            EndPoints::Locators(l) => l.locators,
+        }
+    }
+
+    pub fn as_vec(&self) -> Vec<EndPoint> {
+        match self {
+            EndPoints::Single(ep) => vec![ep.clone()],
+            EndPoints::Locators(l) => l.locators.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 an object with 'strategy' and 'locators'",
+                )
+            }
+
+            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_map<A>(self, map: A) -> Result<Self::Value, A::Error>
+            where
+                A: serde::de::MapAccess<'de>,
+            {
+                #[derive(serde::Deserialize)]
+                struct LocatorsHelper {
+                    strategy: LocatorsStrategy,
+                    locators: Vec<EndPoint>,
+                }
+
+                let s = serde::Deserialize::deserialize(
+                    serde::de::value::MapAccessDeserializer::new(map),
+                )?;
+                let helper: LocatorsHelper = s;
+                Ok(EndPoints::Locators(Locators {
+                    strategy: helper.strategy,
+                    locators: helper.locators,
+                }))
+            }
+        }
+
+        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>";
+        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())
+    );
+    // Locators
+    let json = r#"{"strategy": "allOf", "locators": ["udp/127.0.0.1:7447?rel=0", "udp/127.0.0.1:7447?rel=1"]}"#;
+    let eps: EndPoints = serde_json::from_str(json).unwrap();
+    assert_eq!(
+        eps,
+        EndPoints::Locators(Locators {
+            strategy: LocatorsStrategy::AllOf,
+            locators: 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());

commons/zenoh-test/Cargo.toml

@@ -14,6 +14,7 @@ zenoh = { workspace = true, features = ["internal", "unstable"] }
 zenoh-config = { workspace = true }
 zenoh-core = { workspace = true }
 zenoh-link = { workspace = true }
+zenoh-protocol = { workspace = true }
 
 [package.metadata.cargo-machete]
 ignored = ["tokio"]

commons/zenoh-test/src/lib.rs

@@ -41,9 +41,10 @@ use std::{
 #[cfg(feature = "internal")]
 use zenoh::internal::runtime::{Runtime, RuntimeBuilder};
 use zenoh::{Session, Wait};
-use zenoh_config::{ModeDependentValue, WhatAmI};
+use zenoh_config::WhatAmI;
 use zenoh_core::ztimeout;
 use zenoh_link::EndPoint;
+use zenoh_protocol::core::EndPoints;
 
 /// Default timeout applied to async operations via [`ztimeout!`].
 pub const TIMEOUT: Duration = Duration::from_secs(60);
@@ -175,17 +176,15 @@ impl TestSessions {
         locators: Vec<EndPoint>,
     ) -> zenoh_config::Config {
         println!("Connecting to {:?}", locators);
+        let endpoint_groups: Vec<EndPoints> = locators.into_iter().map(Into::into).collect();
         let mut config = zenoh_config::Config::default();
         config.scouting.multicast.set_enabled(Some(false)).unwrap();
         config
             .transport
             .unicast
-            .set_max_links(locators.len())
-            .unwrap();
-        config
-            .connect
-            .set_endpoints(ModeDependentValue::Unique(locators))
+            .set_max_links(endpoint_groups.len())
             .unwrap();
+        config.connect.endpoints.set(endpoint_groups).unwrap();
 
         config
     }