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

zenoh_protocol/core/
endpoint.rs
