zenoh_protocol/core/
endpoint.rs

1//
2// Copyright (c) 2023 ZettaScale Technology
3//
4// This program and the accompanying materials are made available under the
5// terms of the Eclipse Public License 2.0 which is available at
6// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
7// which is available at https://www.apache.org/licenses/LICENSE-2.0.
8//
9// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
10//
11// Contributors:
12//   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
13//
14use alloc::{borrow::ToOwned, format, string::String, vec, vec::Vec};
15use core::{borrow::Borrow, convert::TryFrom, fmt, str::FromStr};
16
17use zenoh_result::{bail, zerror, Error as ZError, ZResult};
18
19use super::{locator::*, parameters};
20
21// Parsing chars
22pub const PROTO_SEPARATOR: char = '/';
23pub const METADATA_SEPARATOR: char = '?';
24pub const CONFIG_SEPARATOR: char = '#';
25
26// Parsing functions
27pub(super) fn protocol(s: &str) -> &str {
28    let pdix = s.find(PROTO_SEPARATOR).unwrap_or(s.len());
29    &s[..pdix]
30}
31
32pub(super) fn address(s: &str) -> &str {
33    let pdix = s.find(PROTO_SEPARATOR).unwrap_or(s.len());
34    let midx = s.find(METADATA_SEPARATOR).unwrap_or(s.len());
35    let cidx = s.find(CONFIG_SEPARATOR).unwrap_or(s.len());
36    &s[pdix + 1..midx.min(cidx)]
37}
38
39pub(super) fn metadata(s: &str) -> &str {
40    match s.find(METADATA_SEPARATOR) {
41        Some(midx) => {
42            let cidx = s.find(CONFIG_SEPARATOR).unwrap_or(s.len());
43            &s[midx + 1..cidx]
44        }
45        None => "",
46    }
47}
48
49pub(super) fn config(s: &str) -> &str {
50    match s.find(CONFIG_SEPARATOR) {
51        Some(cidx) => &s[cidx + 1..],
52        None => "",
53    }
54}
55
56// Protocol
57#[repr(transparent)]
58#[derive(Copy, Clone, PartialEq, Eq, Hash)]
59pub struct Protocol<'a>(pub(super) &'a str);
60
61impl<'a> Protocol<'a> {
62    pub fn as_str(&self) -> &'a str {
63        self.0
64    }
65}
66
67impl AsRef<str> for Protocol<'_> {
68    fn as_ref(&self) -> &str {
69        self.as_str()
70    }
71}
72
73impl fmt::Display for Protocol<'_> {
74    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
75        f.write_str(self.as_str())
76    }
77}
78
79impl fmt::Debug for Protocol<'_> {
80    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
81        write!(f, "{self}")
82    }
83}
84
85#[repr(transparent)]
86#[derive(PartialEq, Eq, Hash)]
87pub struct ProtocolMut<'a>(&'a mut EndPoint);
88
89impl<'a> ProtocolMut<'a> {
90    pub fn as_str(&'a self) -> &'a str {
91        address(self.0.as_str())
92    }
93
94    pub fn set(&mut self, p: &str) -> ZResult<()> {
95        let ep = EndPoint::new(p, self.0.address(), self.0.metadata(), self.0.config())?;
96
97        self.0.inner = ep.inner;
98        Ok(())
99    }
100}
101
102impl AsRef<str> for ProtocolMut<'_> {
103    fn as_ref(&self) -> &str {
104        self.as_str()
105    }
106}
107
108impl fmt::Display for ProtocolMut<'_> {
109    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
110        f.write_str(self.as_str())
111    }
112}
113
114impl fmt::Debug for ProtocolMut<'_> {
115    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
116        write!(f, "{self}")
117    }
118}
119
120// Address
121#[repr(transparent)]
122#[derive(Copy, Clone, PartialEq, Eq, Hash)]
123pub struct Address<'a>(pub(super) &'a str);
124
125impl<'a> Address<'a> {
126    pub fn as_str(&self) -> &'a str {
127        self.0
128    }
129}
130
131impl AsRef<str> for Address<'_> {
132    fn as_ref(&self) -> &str {
133        self.as_str()
134    }
135}
136
137impl fmt::Display for Address<'_> {
138    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
139        f.write_str(self.as_str())
140    }
141}
142
143impl fmt::Debug for Address<'_> {
144    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
145        write!(f, "{self}")
146    }
147}
148
149impl<'a> From<&'a str> for Address<'a> {
150    fn from(value: &'a str) -> Self {
151        Address(value)
152    }
153}
154
155#[repr(transparent)]
156#[derive(PartialEq, Eq, Hash)]
157pub struct AddressMut<'a>(&'a mut EndPoint);
158
159impl<'a> AddressMut<'a> {
160    pub fn as_str(&'a self) -> &'a str {
161        address(self.0.as_str())
162    }
163
164    pub fn set(&'a mut self, a: &str) -> ZResult<()> {
165        let ep = EndPoint::new(self.0.protocol(), a, self.0.metadata(), self.0.config())?;
166
167        self.0.inner = ep.inner;
168        Ok(())
169    }
170}
171
172impl AsRef<str> for AddressMut<'_> {
173    fn as_ref(&self) -> &str {
174        self.as_str()
175    }
176}
177
178impl fmt::Display for AddressMut<'_> {
179    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
180        f.write_str(self.as_str())
181    }
182}
183
184impl fmt::Debug for AddressMut<'_> {
185    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
186        write!(f, "{self}")
187    }
188}
189
190// Metadata
191#[repr(transparent)]
192#[derive(Copy, Clone, PartialEq, Eq, Hash)]
193pub struct Metadata<'a>(pub(super) &'a str);
194
195impl<'a> Metadata<'a> {
196    pub const RELIABILITY: &'static str = "rel";
197    pub const PRIORITIES: &'static str = "prio";
198    pub const MULTISTREAM: &'static str = "multistream";
199    pub const MIXED_RELIABILITY: &'static str = "mixed_rel";
200
201    pub fn as_str(&self) -> &'a str {
202        self.0
203    }
204
205    pub fn is_empty(&'a self) -> bool {
206        self.as_str().is_empty()
207    }
208
209    pub fn iter(&'a self) -> impl DoubleEndedIterator<Item = (&'a str, &'a str)> + Clone {
210        parameters::iter(self.0)
211    }
212
213    pub fn get(&'a self, k: &str) -> Option<&'a str> {
214        parameters::get(self.0, k)
215    }
216
217    pub fn values(&'a self, k: &str) -> impl DoubleEndedIterator<Item = &'a str> {
218        parameters::values(self.0, k)
219    }
220}
221
222impl AsRef<str> for Metadata<'_> {
223    fn as_ref(&self) -> &str {
224        self.as_str()
225    }
226}
227
228impl fmt::Display for Metadata<'_> {
229    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
230        f.write_str(self.as_str())
231    }
232}
233
234impl fmt::Debug for Metadata<'_> {
235    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
236        write!(f, "{self}")
237    }
238}
239
240#[repr(transparent)]
241#[derive(PartialEq, Eq, Hash)]
242pub struct MetadataMut<'a>(&'a mut EndPoint);
243
244impl<'a> MetadataMut<'a> {
245    pub fn as_str(&'a self) -> &'a str {
246        metadata(self.0.as_str())
247    }
248
249    pub fn is_empty(&'a self) -> bool {
250        self.as_str().is_empty()
251    }
252}
253
254impl MetadataMut<'_> {
255    pub fn extend_from_iter<'s, I, K, V>(&mut self, iter: I) -> ZResult<()>
256    where
257        I: Iterator<Item = (&'s K, &'s V)> + Clone,
258        K: Borrow<str> + 's + ?Sized,
259        V: Borrow<str> + 's + ?Sized,
260    {
261        let ep = EndPoint::new(
262            self.0.protocol(),
263            self.0.address(),
264            parameters::from_iter(parameters::sort(parameters::join(
265                self.0.metadata().iter(),
266                iter.map(|(k, v)| (k.borrow(), v.borrow())),
267            ))),
268            self.0.config(),
269        )?;
270
271        self.0.inner = ep.inner;
272        Ok(())
273    }
274
275    pub fn insert<K, V>(&mut self, k: K, v: V) -> ZResult<()>
276    where
277        K: Borrow<str>,
278        V: Borrow<str>,
279    {
280        let ep = EndPoint::new(
281            self.0.protocol(),
282            self.0.address(),
283            parameters::insert_sort(self.0.metadata().as_str(), k.borrow(), v.borrow()).0,
284            self.0.config(),
285        )?;
286
287        self.0.inner = ep.inner;
288        Ok(())
289    }
290
291    pub fn remove<K>(&mut self, k: K) -> ZResult<()>
292    where
293        K: Borrow<str>,
294    {
295        let ep = EndPoint::new(
296            self.0.protocol(),
297            self.0.address(),
298            parameters::remove(self.0.metadata().as_str(), k.borrow()).0,
299            self.0.config(),
300        )?;
301
302        self.0.inner = ep.inner;
303        Ok(())
304    }
305}
306
307impl AsRef<str> for MetadataMut<'_> {
308    fn as_ref(&self) -> &str {
309        self.as_str()
310    }
311}
312
313impl fmt::Display for MetadataMut<'_> {
314    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
315        f.write_str(self.as_str())
316    }
317}
318
319impl fmt::Debug for MetadataMut<'_> {
320    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
321        write!(f, "{self}")
322    }
323}
324
325// Config
326#[repr(transparent)]
327#[derive(Copy, Clone, PartialEq, Eq, Hash)]
328pub struct Config<'a>(pub(super) &'a str);
329
330impl<'a> Config<'a> {
331    pub fn as_str(&self) -> &'a str {
332        self.0
333    }
334
335    pub fn is_empty(&self) -> bool {
336        self.as_str().is_empty()
337    }
338
339    pub fn iter(&self) -> impl DoubleEndedIterator<Item = (&'a str, &'a str)> + Clone {
340        parameters::iter(self.0)
341    }
342
343    pub fn get(&self, k: &str) -> Option<&'a str> {
344        parameters::get(self.0, k)
345    }
346
347    pub fn values(&self, k: &str) -> impl DoubleEndedIterator<Item = &'a str> {
348        parameters::values(self.0, k)
349    }
350}
351
352impl AsRef<str> for Config<'_> {
353    fn as_ref(&self) -> &str {
354        self.as_str()
355    }
356}
357
358impl fmt::Display for Config<'_> {
359    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
360        f.write_str(self.as_str())
361    }
362}
363
364impl fmt::Debug for Config<'_> {
365    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
366        write!(f, "{self}")
367    }
368}
369
370#[repr(transparent)]
371#[derive(PartialEq, Eq, Hash)]
372pub struct ConfigMut<'a>(&'a mut EndPoint);
373
374impl<'a> ConfigMut<'a> {
375    pub fn as_str(&'a self) -> &'a str {
376        config(self.0.as_str())
377    }
378
379    pub fn is_empty(&'a self) -> bool {
380        self.as_str().is_empty()
381    }
382}
383
384impl ConfigMut<'_> {
385    pub fn extend_from_iter<'s, I, K, V>(&mut self, iter: I) -> ZResult<()>
386    where
387        I: Iterator<Item = (&'s K, &'s V)> + Clone,
388        K: Borrow<str> + 's + ?Sized,
389        V: Borrow<str> + 's + ?Sized,
390    {
391        let ep = EndPoint::new(
392            self.0.protocol(),
393            self.0.address(),
394            self.0.metadata(),
395            parameters::from_iter(parameters::sort(parameters::join(
396                self.0.config().iter(),
397                iter.map(|(k, v)| (k.borrow(), v.borrow())),
398            ))),
399        )?;
400
401        self.0.inner = ep.inner;
402        Ok(())
403    }
404
405    pub fn insert<K, V>(&mut self, k: K, v: V) -> ZResult<()>
406    where
407        K: Borrow<str>,
408        V: Borrow<str>,
409    {
410        let ep = EndPoint::new(
411            self.0.protocol(),
412            self.0.address(),
413            self.0.metadata(),
414            parameters::insert_sort(self.0.config().as_str(), k.borrow(), v.borrow()).0,
415        )?;
416
417        self.0.inner = ep.inner;
418        Ok(())
419    }
420
421    pub fn remove<K>(&mut self, k: K) -> ZResult<()>
422    where
423        K: Borrow<str>,
424    {
425        let ep = EndPoint::new(
426            self.0.protocol(),
427            self.0.address(),
428            self.0.metadata(),
429            parameters::remove(self.0.config().as_str(), k.borrow()).0,
430        )?;
431
432        self.0.inner = ep.inner;
433        Ok(())
434    }
435}
436
437impl AsRef<str> for ConfigMut<'_> {
438    fn as_ref(&self) -> &str {
439        self.as_str()
440    }
441}
442
443impl fmt::Display for ConfigMut<'_> {
444    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
445        f.write_str(self.as_str())
446    }
447}
448
449impl fmt::Debug for ConfigMut<'_> {
450    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
451        write!(f, "{self}")
452    }
453}
454
455/// A string that respects the [`EndPoint`] canon form: `<locator>[#<config>]`.
456///
457/// `<locator>` is a valid [`Locator`] and `<config>` is of the form
458/// `<key1>=<value1>;...;<keyN>=<valueN>` where keys are alphabetically sorted. `<config>` is
459/// optional and can be provided to configure some aspects for an [`EndPoint`], e.g. the interface
460/// to listen on or connect to.
461///
462/// A full [`EndPoint`] string is hence in the form of `<proto>/<address>[?<metadata>][#config]`.
463///
464/// ## Metadata
465///
466/// - **`prio`**: a priority range bounded inclusively below and above (e.g. `2-4` signifies
467///   priorities 2, 3 and 4). This value is used to select the link used for transmission based on
468///   the Priority of the message in question.
469///
470///   For example, `tcp/localhost:7447?prio=1-3` assigns priorities
471///   [`Priority::RealTime`](crate::core::Priority::RealTime), [Priority::InteractiveHigh](crate::core::Priority::InteractiveHigh) and
472///   [`Priority::InteractiveLow`](crate::core::Priority::InteractiveLow) to the established link.
473///
474/// - **`rel`**: either "0" for [`Reliability::BestEffort`](crate::core::Reliability::BestEffort) or "1" for
475///   [`Reliability::Reliable`](crate::core::Reliability::Reliable). This value is used to select the link used for
476///   transmission based on the reliability of the message in question.
477///
478///   For example, `tcp/localhost:7447?prio=6-7;rel=0` assigns priorities
479///   [`Priority::DataLow`](crate::core::Priority::DataLow) and [Priority::Background](crate::core::Priority::Background), and
480///   [`Reliability::BestEffort`](crate::core::Reliability::BestEffort) to the established link.
481#[derive(Clone, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
482#[serde(into = "String")]
483#[serde(try_from = "String")]
484pub struct EndPoint {
485    pub(super) inner: String,
486}
487
488impl EndPoint {
489    pub fn new<A, B, C, D>(protocol: A, address: B, metadata: C, config: D) -> ZResult<Self>
490    where
491        A: AsRef<str>,
492        B: AsRef<str>,
493        C: AsRef<str>,
494        D: AsRef<str>,
495    {
496        let p: &str = protocol.as_ref();
497        let a: &str = address.as_ref();
498        let m: &str = metadata.as_ref();
499        let c: &str = config.as_ref();
500
501        let len = p.len() + a.len() + m.len();
502        if len > u8::MAX as usize {
503            bail!("Endpoint too big: {} bytes. Max: {} bytes. ", len, u8::MAX);
504        }
505
506        let s = match (m.is_empty(), c.is_empty()) {
507            (true, true) => format!("{p}{PROTO_SEPARATOR}{a}"),
508            (false, true) => format!("{p}{PROTO_SEPARATOR}{a}{METADATA_SEPARATOR}{m}"),
509            (true, false) => format!("{p}{PROTO_SEPARATOR}{a}{CONFIG_SEPARATOR}{c}"),
510            (false, false) => {
511                format!("{p}{PROTO_SEPARATOR}{a}{METADATA_SEPARATOR}{m}{CONFIG_SEPARATOR}{c}")
512            }
513        };
514
515        Self::try_from(s)
516    }
517
518    pub fn as_str(&self) -> &str {
519        self.inner.as_str()
520    }
521
522    pub fn split(&self) -> (Protocol<'_>, Address<'_>, Metadata<'_>, Config<'_>) {
523        (
524            self.protocol(),
525            self.address(),
526            self.metadata(),
527            self.config(),
528        )
529    }
530
531    pub fn protocol(&self) -> Protocol<'_> {
532        Protocol(protocol(self.inner.as_str()))
533    }
534
535    pub fn protocol_mut(&mut self) -> ProtocolMut<'_> {
536        ProtocolMut(self)
537    }
538
539    pub fn address(&self) -> Address<'_> {
540        Address(address(self.inner.as_str()))
541    }
542
543    pub fn address_mut(&mut self) -> AddressMut<'_> {
544        AddressMut(self)
545    }
546
547    pub fn metadata(&self) -> Metadata<'_> {
548        Metadata(metadata(self.inner.as_str()))
549    }
550
551    pub fn metadata_mut(&mut self) -> MetadataMut<'_> {
552        MetadataMut(self)
553    }
554
555    pub fn config(&self) -> Config<'_> {
556        Config(config(self.inner.as_str()))
557    }
558
559    pub fn config_mut(&mut self) -> ConfigMut<'_> {
560        ConfigMut(self)
561    }
562
563    pub fn to_locator(&self) -> Locator {
564        self.clone().into()
565    }
566
567    /// Constructs an uninitialized empty EndPoint.
568    #[zenoh_macros::internal]
569    pub fn empty() -> Self {
570        EndPoint {
571            inner: String::default(),
572        }
573    }
574}
575
576impl From<Locator> for EndPoint {
577    fn from(val: Locator) -> Self {
578        val.0
579    }
580}
581
582impl fmt::Display for EndPoint {
583    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
584        f.write_str(&self.inner)
585    }
586}
587
588impl fmt::Debug for EndPoint {
589    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
590        write!(f, "{self}")
591    }
592}
593
594impl From<EndPoint> for String {
595    fn from(v: EndPoint) -> String {
596        v.inner
597    }
598}
599
600impl TryFrom<String> for EndPoint {
601    type Error = ZError;
602
603    fn try_from(s: String) -> Result<Self, Self::Error> {
604        const ERR: &str =
605            "Endpoints must be of the form <protocol>/<address>[?<metadata>][#<config>]";
606        const PARAM_ERR: &str =
607            "Endpoint metadata and config must contain at least one valid parameter with a non-empty key";
608
609        let pidx = s
610            .find(PROTO_SEPARATOR)
611            .and_then(|i| (!s[..i].is_empty() && !s[i + 1..].is_empty()).then_some(i))
612            .ok_or_else(|| zerror!("{}: {}", ERR, s))?;
613
614        match (s.find(METADATA_SEPARATOR), s.find(CONFIG_SEPARATOR)) {
615            // No metadata or config at all
616            (None, None) => Ok(EndPoint { inner: s }),
617            // There is some metadata
618            (Some(midx), None) if midx > pidx && !s[midx + 1..].is_empty() => {
619                if !parameters::is_well_formed(&s[midx + 1..]) {
620                    bail!("{}: {}", PARAM_ERR, s);
621                }
622                let mut inner = String::with_capacity(s.len());
623                inner.push_str(&s[..midx + 1]); // Includes metadata separator
624                parameters::from_iter_into(
625                    parameters::sort(parameters::iter(&s[midx + 1..])),
626                    &mut inner,
627                );
628                Ok(EndPoint { inner })
629            }
630            // There is some config
631            (None, Some(cidx)) if cidx > pidx && !s[cidx + 1..].is_empty() => {
632                if !parameters::is_well_formed(&s[cidx + 1..]) {
633                    bail!("{}: {}", PARAM_ERR, s);
634                }
635                let mut inner = String::with_capacity(s.len());
636                inner.push_str(&s[..cidx + 1]); // Includes config separator
637                parameters::from_iter_into(
638                    parameters::sort(parameters::iter(&s[cidx + 1..])),
639                    &mut inner,
640                );
641                Ok(EndPoint { inner })
642            }
643            // There is some metadata and some config
644            (Some(midx), Some(cidx))
645                if midx > pidx
646                    && cidx > midx
647                    && !s[midx + 1..cidx].is_empty()
648                    && !s[cidx + 1..].is_empty() =>
649            {
650                if !parameters::is_well_formed(&s[midx + 1..cidx])
651                    || !parameters::is_well_formed(&s[cidx + 1..])
652                {
653                    bail!("{}: {}", PARAM_ERR, s);
654                }
655                let mut inner = String::with_capacity(s.len());
656                inner.push_str(&s[..midx + 1]); // Includes metadata separator
657
658                parameters::from_iter_into(
659                    parameters::sort(parameters::iter(&s[midx + 1..cidx])),
660                    &mut inner,
661                );
662
663                inner.push(CONFIG_SEPARATOR);
664                parameters::from_iter_into(
665                    parameters::sort(parameters::iter(&s[cidx + 1..])),
666                    &mut inner,
667                );
668
669                Ok(EndPoint { inner })
670            }
671            _ => Err(zerror!("{}: {}", ERR, s).into()),
672        }
673    }
674}
675
676impl FromStr for EndPoint {
677    type Err = ZError;
678
679    fn from_str(s: &str) -> Result<Self, Self::Err> {
680        Self::try_from(s.to_owned())
681    }
682}
683
684impl EndPoint {
685    #[cfg(feature = "test")]
686    #[doc(hidden)]
687    pub fn rand() -> Self {
688        use rand::{
689            distributions::{Alphanumeric, DistString},
690            Rng,
691        };
692
693        const MIN: usize = 2;
694        const MAX: usize = 8;
695
696        let mut rng = rand::thread_rng();
697        let mut endpoint = String::new();
698
699        let len = rng.gen_range(MIN..MAX);
700        let proto = Alphanumeric.sample_string(&mut rng, len);
701        endpoint.push_str(proto.as_str());
702
703        endpoint.push(PROTO_SEPARATOR);
704
705        let len = rng.gen_range(MIN..MAX);
706        let address = Alphanumeric.sample_string(&mut rng, len);
707        endpoint.push_str(address.as_str());
708
709        if rng.gen_bool(0.5) {
710            endpoint.push(METADATA_SEPARATOR);
711            parameters::rand(&mut endpoint);
712        }
713        if rng.gen_bool(0.5) {
714            endpoint.push(CONFIG_SEPARATOR);
715            parameters::rand(&mut endpoint);
716        }
717
718        endpoint.parse().unwrap()
719    }
720}
721
722#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
723#[serde(rename_all = "camelCase")]
724pub enum LocatorsStrategy {
725    /// Open links to all locators in the group.
726    AllOf,
727    /// Reserved for future support. The runtime currently only implements
728    /// `AllOf` semantics.
729    OneOf,
730}
731
732#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
733pub struct Locators {
734    pub strategy: LocatorsStrategy,
735    pub locators: Vec<EndPoint>,
736}
737
738#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
739#[serde(untagged)]
740pub enum EndPoints {
741    Single(EndPoint),
742    Locators(Locators),
743}
744impl EndPoints {
745    pub fn flatten(self) -> Vec<EndPoint> {
746        match self {
747            EndPoints::Single(ep) => vec![ep],
748            EndPoints::Locators(l) => l.locators,
749        }
750    }
751
752    pub fn as_vec(&self) -> Vec<EndPoint> {
753        match self {
754            EndPoints::Single(ep) => vec![ep.clone()],
755            EndPoints::Locators(l) => l.locators.clone(),
756        }
757    }
758}
759
760impl From<EndPoint> for EndPoints {
761    fn from(ep: EndPoint) -> EndPoints {
762        EndPoints::Single(ep)
763    }
764}
765
766impl<'de> serde::Deserialize<'de> for EndPoints {
767    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
768    where
769        D: serde::Deserializer<'de>,
770    {
771        struct EndPointsVisitor;
772
773        impl<'de> serde::de::Visitor<'de> for EndPointsVisitor {
774            type Value = EndPoints;
775
776            fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
777                formatter.write_str(
778                    "a single endpoint string or an object with 'strategy' and 'locators'",
779                )
780            }
781
782            fn visit_str<E>(self, v: &str) -> Result<Self::Value, E>
783            where
784                E: serde::de::Error,
785            {
786                EndPoint::from_str(v)
787                    .map(EndPoints::Single)
788                    .map_err(serde::de::Error::custom)
789            }
790
791            fn visit_map<A>(self, map: A) -> Result<Self::Value, A::Error>
792            where
793                A: serde::de::MapAccess<'de>,
794            {
795                #[derive(serde::Deserialize)]
796                struct LocatorsHelper {
797                    strategy: LocatorsStrategy,
798                    locators: Vec<EndPoint>,
799                }
800
801                let s = serde::Deserialize::deserialize(
802                    serde::de::value::MapAccessDeserializer::new(map),
803                )?;
804                let helper: LocatorsHelper = s;
805                Ok(EndPoints::Locators(Locators {
806                    strategy: helper.strategy,
807                    locators: helper.locators,
808                }))
809            }
810        }
811
812        deserializer.deserialize_any(EndPointsVisitor)
813    }
814}
815
816impl TryFrom<String> for EndPoints {
817    type Error = ZError;
818
819    fn try_from(s: String) -> Result<Self, Self::Error> {
820        const ERR: &str = "Endpoints must be of the form <endpoint>";
821        EndPoint::from_str(s.as_str())
822            .map(EndPoints::Single)
823            .map_err(|e| zerror!("{}: {}", ERR, e).into())
824    }
825}
826
827impl FromStr for EndPoints {
828    type Err = ZError;
829
830    fn from_str(s: &str) -> Result<Self, Self::Err> {
831        Self::try_from(s.to_owned())
832    }
833}
834
835#[test]
836fn endpoints() {
837    // Single
838    assert_eq!(
839        EndPoints::from_str("udp/127.0.0.1:7447").unwrap(),
840        EndPoints::Single(EndPoint::from_str("udp/127.0.0.1:7447").unwrap())
841    );
842    // Locators
843    let json = r#"{"strategy": "allOf", "locators": ["udp/127.0.0.1:7447?rel=0", "udp/127.0.0.1:7447?rel=1"]}"#;
844    let eps: EndPoints = serde_json::from_str(json).unwrap();
845    assert_eq!(
846        eps,
847        EndPoints::Locators(Locators {
848            strategy: LocatorsStrategy::AllOf,
849            locators: vec![
850                EndPoint::from_str("udp/127.0.0.1:7447?rel=0").unwrap(),
851                EndPoint::from_str("udp/127.0.0.1:7447?rel=1").unwrap()
852            ]
853        })
854    );
855}
856
857#[test]
858fn endpoint() {
859    assert!(EndPoint::from_str("/").is_err());
860    assert!(EndPoint::from_str("?").is_err());
861    assert!(EndPoint::from_str("#").is_err());
862
863    assert!(EndPoint::from_str("udp").is_err());
864    assert!(EndPoint::from_str("/udp").is_err());
865    assert!(EndPoint::from_str("udp/").is_err());
866
867    assert!(EndPoint::from_str("udp/127.0.0.1:7447?").is_err());
868    assert!(EndPoint::from_str("udp?127.0.0.1:7447").is_err());
869    assert!(EndPoint::from_str("udp?127.0.0.1:7447/meta").is_err());
870
871    assert!(EndPoint::from_str("udp/127.0.0.1:7447#").is_err());
872    assert!(EndPoint::from_str("udp/127.0.0.1:7447?#").is_err());
873    assert!(EndPoint::from_str("udp/127.0.0.1:7447#?").is_err());
874    assert!(EndPoint::from_str("udp#127.0.0.1:7447/").is_err());
875    assert!(EndPoint::from_str("udp#127.0.0.1:7447/?").is_err());
876    assert!(EndPoint::from_str("udp/127.0.0.1:7447?a=1#").is_err());
877    assert!(EndPoint::from_str("udp/127.0.0.1:7447?;;;").is_err());
878    assert!(EndPoint::from_str("udp/127.0.0.1:7447#;;;").is_err());
879    assert!(EndPoint::from_str("udp/127.0.0.1:7447?a=1#;;;").is_err());
880    assert!(EndPoint::from_str("udp/127.0.0.1:7447?=1").is_err());
881    assert!(EndPoint::from_str("udp/127.0.0.1:7447#=1").is_err());
882    assert!(EndPoint::from_str("udp/127.0.0.1:7447?a=1#=1").is_err());
883
884    let endpoint = EndPoint::from_str("udp/127.0.0.1:7447").unwrap();
885    assert_eq!(endpoint.as_str(), "udp/127.0.0.1:7447");
886    assert_eq!(endpoint.protocol().as_str(), "udp");
887    assert_eq!(endpoint.address().as_str(), "127.0.0.1:7447");
888    assert!(endpoint.metadata().as_str().is_empty());
889    assert_eq!(endpoint.metadata().iter().count(), 0);
890
891    let endpoint = EndPoint::from_str("udp/127.0.0.1:7447?a=1;b=2").unwrap();
892    assert_eq!(endpoint.as_str(), "udp/127.0.0.1:7447?a=1;b=2");
893    assert_eq!(endpoint.protocol().as_str(), "udp");
894    assert_eq!(endpoint.address().as_str(), "127.0.0.1:7447");
895    assert_eq!(endpoint.metadata().as_str(), "a=1;b=2");
896    assert_eq!(endpoint.metadata().iter().count(), 2);
897    endpoint
898        .metadata()
899        .iter()
900        .find(|x| x == &("a", "1"))
901        .unwrap();
902    assert_eq!(endpoint.metadata().get("a"), Some("1"));
903    endpoint
904        .metadata()
905        .iter()
906        .find(|x| x == &("b", "2"))
907        .unwrap();
908    assert_eq!(endpoint.metadata().get("b"), Some("2"));
909    assert!(endpoint.config().as_str().is_empty());
910    assert_eq!(endpoint.config().iter().count(), 0);
911
912    let endpoint = EndPoint::from_str("udp/127.0.0.1:7447?b=2;a=1").unwrap();
913    assert_eq!(endpoint.as_str(), "udp/127.0.0.1:7447?a=1;b=2");
914    assert_eq!(endpoint.protocol().as_str(), "udp");
915    assert_eq!(endpoint.address().as_str(), "127.0.0.1:7447");
916    assert_eq!(endpoint.metadata().as_str(), "a=1;b=2");
917    assert_eq!(endpoint.metadata().iter().count(), 2);
918    endpoint
919        .metadata()
920        .iter()
921        .find(|x| x == &("a", "1"))
922        .unwrap();
923    assert_eq!(endpoint.metadata().get("a"), Some("1"));
924    endpoint
925        .metadata()
926        .iter()
927        .find(|x| x == &("b", "2"))
928        .unwrap();
929    assert_eq!(endpoint.metadata().get("a"), Some("1"));
930    assert!(endpoint.config().as_str().is_empty());
931    assert_eq!(endpoint.config().iter().count(), 0);
932
933    let endpoint = EndPoint::from_str("udp/127.0.0.1:7447#A=1;B=2").unwrap();
934    assert_eq!(endpoint.as_str(), "udp/127.0.0.1:7447#A=1;B=2");
935    assert_eq!(endpoint.protocol().as_str(), "udp");
936    assert_eq!(endpoint.address().as_str(), "127.0.0.1:7447");
937    assert!(endpoint.metadata().as_str().is_empty());
938    assert_eq!(endpoint.metadata().iter().count(), 0);
939    assert_eq!(endpoint.config().as_str(), "A=1;B=2");
940    assert_eq!(endpoint.config().iter().count(), 2);
941    endpoint.config().iter().find(|x| x == &("A", "1")).unwrap();
942    assert_eq!(endpoint.config().get("A"), Some("1"));
943    endpoint.config().iter().find(|x| x == &("B", "2")).unwrap();
944    assert_eq!(endpoint.config().get("B"), Some("2"));
945
946    let endpoint = EndPoint::from_str("udp/127.0.0.1:7447#B=2;A=1").unwrap();
947    assert_eq!(endpoint.as_str(), "udp/127.0.0.1:7447#A=1;B=2");
948    assert_eq!(endpoint.protocol().as_str(), "udp");
949    assert_eq!(endpoint.address().as_str(), "127.0.0.1:7447");
950    assert!(endpoint.metadata().as_str().is_empty());
951    assert_eq!(endpoint.metadata().iter().count(), 0);
952    assert_eq!(endpoint.config().as_str(), "A=1;B=2");
953    assert_eq!(endpoint.config().iter().count(), 2);
954    endpoint.config().iter().find(|x| x == &("A", "1")).unwrap();
955    assert_eq!(endpoint.config().get("A"), Some("1"));
956    endpoint.config().iter().find(|x| x == &("B", "2")).unwrap();
957    assert_eq!(endpoint.config().get("B"), Some("2"));
958
959    let endpoint = EndPoint::from_str("udp/127.0.0.1:7447?a=1;b=2#A=1;B=2").unwrap();
960    assert_eq!(endpoint.as_str(), "udp/127.0.0.1:7447?a=1;b=2#A=1;B=2");
961    assert_eq!(endpoint.protocol().as_str(), "udp");
962    assert_eq!(endpoint.address().as_str(), "127.0.0.1:7447");
963    assert_eq!(endpoint.metadata().as_str(), "a=1;b=2");
964    assert_eq!(endpoint.metadata().iter().count(), 2);
965    endpoint
966        .metadata()
967        .iter()
968        .find(|x| x == &("a", "1"))
969        .unwrap();
970    assert_eq!(endpoint.metadata().get("a"), Some("1"));
971    endpoint
972        .metadata()
973        .iter()
974        .find(|x| x == &("b", "2"))
975        .unwrap();
976    assert_eq!(endpoint.metadata().get("b"), Some("2"));
977    assert_eq!(endpoint.config().as_str(), "A=1;B=2");
978    assert_eq!(endpoint.config().iter().count(), 2);
979    endpoint.config().iter().find(|x| x == &("A", "1")).unwrap();
980    assert_eq!(endpoint.config().get("A"), Some("1"));
981    endpoint.config().iter().find(|x| x == &("B", "2")).unwrap();
982    assert_eq!(endpoint.config().get("B"), Some("2"));
983
984    let endpoint = EndPoint::from_str("udp/127.0.0.1:7447?b=2;a=1#B=2;A=1").unwrap();
985    assert_eq!(endpoint.as_str(), "udp/127.0.0.1:7447?a=1;b=2#A=1;B=2");
986    assert_eq!(endpoint.protocol().as_str(), "udp");
987    assert_eq!(endpoint.address().as_str(), "127.0.0.1:7447");
988    assert_eq!(endpoint.metadata().as_str(), "a=1;b=2");
989    assert_eq!(endpoint.metadata().iter().count(), 2);
990    endpoint
991        .metadata()
992        .iter()
993        .find(|x| x == &("a", "1"))
994        .unwrap();
995    assert_eq!(endpoint.metadata().get("a"), Some("1"));
996    endpoint
997        .metadata()
998        .iter()
999        .find(|x| x == &("b", "2"))
1000        .unwrap();
1001    assert_eq!(endpoint.metadata().get("b"), Some("2"));
1002    assert_eq!(endpoint.config().as_str(), "A=1;B=2");
1003    assert_eq!(endpoint.config().iter().count(), 2);
1004    endpoint.config().iter().find(|x| x == &("A", "1")).unwrap();
1005    assert_eq!(endpoint.config().get("A"), Some("1"));
1006    endpoint.config().iter().find(|x| x == &("B", "2")).unwrap();
1007    assert_eq!(endpoint.config().get("B"), Some("2"));
1008
1009    let mut endpoint = EndPoint::from_str("udp/127.0.0.1:7447?a=1;b=2").unwrap();
1010    endpoint.metadata_mut().insert("c", "3").unwrap();
1011    assert_eq!(endpoint.as_str(), "udp/127.0.0.1:7447?a=1;b=2;c=3");
1012
1013    let mut endpoint = EndPoint::from_str("udp/127.0.0.1:7447?b=2;c=3").unwrap();
1014    endpoint.metadata_mut().insert("a", "1").unwrap();
1015    assert_eq!(endpoint.as_str(), "udp/127.0.0.1:7447?a=1;b=2;c=3");
1016
1017    let mut endpoint = EndPoint::from_str("udp/127.0.0.1:7447?a=1;b=2").unwrap();
1018    endpoint.config_mut().insert("A", "1").unwrap();
1019    assert_eq!(endpoint.as_str(), "udp/127.0.0.1:7447?a=1;b=2#A=1");
1020
1021    let mut endpoint = EndPoint::from_str("udp/127.0.0.1:7447?b=2;c=3#B=2").unwrap();
1022    endpoint.config_mut().insert("A", "1").unwrap();
1023    assert_eq!(endpoint.as_str(), "udp/127.0.0.1:7447?b=2;c=3#A=1;B=2");
1024
1025    let mut endpoint = EndPoint::from_str("udp/127.0.0.1:7447").unwrap();
1026    endpoint
1027        .metadata_mut()
1028        .extend_from_iter([("a", "1"), ("c", "3"), ("b", "2")].iter().copied())
1029        .unwrap();
1030    assert_eq!(endpoint.as_str(), "udp/127.0.0.1:7447?a=1;b=2;c=3");
1031
1032    let mut endpoint = EndPoint::from_str("udp/127.0.0.1:7447").unwrap();
1033    endpoint
1034        .config_mut()
1035        .extend_from_iter([("A", "1"), ("C", "3"), ("B", "2")].iter().copied())
1036        .unwrap();
1037    assert_eq!(endpoint.as_str(), "udp/127.0.0.1:7447#A=1;B=2;C=3");
1038
1039    let endpoint =
1040        EndPoint::from_str("udp/127.0.0.1:7447#iface=en0;join=224.0.0.1|224.0.0.2|224.0.0.3")
1041            .unwrap();
1042    let c = endpoint.config();
1043    assert_eq!(c.get("iface"), Some("en0"));
1044    assert_eq!(c.get("join"), Some("224.0.0.1|224.0.0.2|224.0.0.3"));
1045    assert_eq!(c.values("iface").count(), 1);
1046    let mut i = c.values("iface");
1047    assert_eq!(i.next(), Some("en0"));
1048    assert_eq!(c.values("join").count(), 3);
1049    let mut i = c.values("join");
1050    assert_eq!(i.next(), Some("224.0.0.1"));
1051    assert_eq!(i.next(), Some("224.0.0.2"));
1052    assert_eq!(i.next(), Some("224.0.0.3"));
1053    assert_eq!(i.next(), None);
1054}
zenoh_protocol/core/endpoint.rs

zenoh_protocol/core/
endpoint.rs
