//
// Copyright (c) 2023 ZettaScale Technology
//
// This program and the accompanying materials are made available under the
// terms of the Eclipse Public License 2.0 which is available at
// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
// which is available at https://www.apache.org/licenses/LICENSE-2.0.
//
// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
//
// Contributors:
//   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
//
use super::locator::*;
use alloc::{borrow::ToOwned, format, string::String, vec::Vec};
use core::{convert::TryFrom, fmt, str::FromStr};
use zenoh_result::{bail, zerror, Error as ZError, ZResult};

// Parsing chars
pub const PROTO_SEPARATOR: char = '/';
pub const METADATA_SEPARATOR: char = '?';
pub const LIST_SEPARATOR: char = ';';
pub const FIELD_SEPARATOR: char = '=';
pub const CONFIG_SEPARATOR: char = '#';
pub const VALUE_SEPARATOR: char = '|';

fn split_once(s: &str, c: char) -> (&str, &str) {
    match s.find(c) {
        Some(index) => {
            let (l, r) = s.split_at(index);
            (l, &r[1..])
        }
        None => (s, ""),
    }
}

// Parsing functions
pub(super) fn protocol(s: &str) -> &str {
    let pdix = s.find(PROTO_SEPARATOR).unwrap_or(s.len());
    &s[..pdix]
}

pub(super) fn address(s: &str) -> &str {
    let pdix = s.find(PROTO_SEPARATOR).unwrap_or(s.len());
    let midx = s.find(METADATA_SEPARATOR).unwrap_or(s.len());
    let cidx = s.find(CONFIG_SEPARATOR).unwrap_or(s.len());
    &s[pdix + 1..midx.min(cidx)]
}

pub(super) fn metadata(s: &str) -> &str {
    match s.find(METADATA_SEPARATOR) {
        Some(midx) => {
            let cidx = s.find(CONFIG_SEPARATOR).unwrap_or(s.len());
            &s[midx + 1..cidx]
        }
        None => "",
    }
}

pub(super) fn config(s: &str) -> &str {
    match s.find(CONFIG_SEPARATOR) {
        Some(cidx) => &s[cidx + 1..],
        None => "",
    }
}

pub struct Parameters;

impl Parameters {
    pub fn extend<'s, I>(iter: I, into: &mut String)
    where
        I: Iterator<Item = (&'s str, &'s str)>,
    {
        let mut first = into.is_empty();
        for (k, v) in iter {
            if !first {
                into.push(LIST_SEPARATOR);
            }
            into.push_str(k);
            if !v.is_empty() {
                into.push(FIELD_SEPARATOR);
                into.push_str(v);
            }
            first = false;
        }
    }

    pub fn iter(s: &str) -> impl DoubleEndedIterator<Item = (&str, &str)> {
        s.split(LIST_SEPARATOR).filter_map(|prop| {
            if prop.is_empty() {
                None
            } else {
                Some(split_once(prop, FIELD_SEPARATOR))
            }
        })
    }

    pub fn get<'s>(s: &'s str, k: &str) -> Option<&'s str> {
        Self::iter(s).find(|x| x.0 == k).map(|x| x.1)
    }

    pub fn values<'s>(s: &'s str, k: &str) -> impl DoubleEndedIterator<Item = &'s str> {
        match Self::get(s, k) {
            Some(v) => v.split(VALUE_SEPARATOR),
            None => {
                let mut i = "".split(VALUE_SEPARATOR);
                i.next();
                i
            }
        }
    }

    pub(super) fn insert<'s, I>(iter: I, k: &'s str, v: &'s str) -> String
    where
        I: Iterator<Item = (&'s str, &'s str)>,
    {
        let current = iter.filter(|x| x.0 != k);
        let new = Some((k, v)).into_iter();
        let iter = current.chain(new);

        let mut into = String::new();
        Parameters::extend(iter, &mut into);
        into
    }

    pub(super) fn remove<'s, I>(iter: I, k: &'s str) -> String
    where
        I: Iterator<Item = (&'s str, &'s str)>,
    {
        let iter = iter.filter(|x| x.0 != k);

        let mut into = String::new();
        Parameters::extend(iter, &mut into);
        into
    }
}

// Protocol
#[repr(transparent)]
#[derive(Copy, Clone, PartialEq, Eq, Hash)]
pub struct Protocol<'a>(pub(super) &'a str);

impl<'a> Protocol<'a> {
    pub fn as_str(&self) -> &'a str {
        self.0
    }
}

impl AsRef<str> for Protocol<'_> {
    fn as_ref(&self) -> &str {
        self.as_str()
    }
}

impl fmt::Display for Protocol<'_> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(self.as_str())
    }
}

impl fmt::Debug for Protocol<'_> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self)
    }
}

#[repr(transparent)]
#[derive(PartialEq, Eq, Hash)]
pub struct ProtocolMut<'a>(&'a mut EndPoint);

impl<'a> ProtocolMut<'a> {
    pub fn as_str(&'a self) -> &'a str {
        address(self.0.as_str())
    }

    pub fn set(&mut self, p: &str) -> ZResult<()> {
        let ep = EndPoint::new(p, self.0.address(), self.0.metadata(), self.0.config())?;

        self.0.inner = ep.inner;
        Ok(())
    }
}

impl AsRef<str> for ProtocolMut<'_> {
    fn as_ref(&self) -> &str {
        self.as_str()
    }
}

impl fmt::Display for ProtocolMut<'_> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(self.as_str())
    }
}

impl fmt::Debug for ProtocolMut<'_> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self)
    }
}

// Address
#[repr(transparent)]
#[derive(Copy, Clone, PartialEq, Eq, Hash)]
pub struct Address<'a>(pub(super) &'a str);

impl<'a> Address<'a> {
    pub fn as_str(&self) -> &'a str {
        self.0
    }
}

impl AsRef<str> for Address<'_> {
    fn as_ref(&self) -> &str {
        self.as_str()
    }
}

impl fmt::Display for Address<'_> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(self.as_str())
    }
}

impl fmt::Debug for Address<'_> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self)
    }
}

#[repr(transparent)]
#[derive(PartialEq, Eq, Hash)]
pub struct AddressMut<'a>(&'a mut EndPoint);

impl<'a> AddressMut<'a> {
    pub fn as_str(&'a self) -> &'a str {
        address(self.0.as_str())
    }

    pub fn set(&'a mut self, a: &str) -> ZResult<()> {
        let ep = EndPoint::new(self.0.protocol(), a, self.0.metadata(), self.0.config())?;

        self.0.inner = ep.inner;
        Ok(())
    }
}

impl AsRef<str> for AddressMut<'_> {
    fn as_ref(&self) -> &str {
        self.as_str()
    }
}

impl fmt::Display for AddressMut<'_> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(self.as_str())
    }
}

impl fmt::Debug for AddressMut<'_> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self)
    }
}

// Metadata
#[repr(transparent)]
#[derive(Copy, Clone, PartialEq, Eq, Hash)]
pub struct Metadata<'a>(pub(super) &'a str);

impl<'a> Metadata<'a> {
    pub fn as_str(&self) -> &'a str {
        self.0
    }

    pub fn is_empty(&'a self) -> bool {
        self.as_str().is_empty()
    }

    pub fn iter(&'a self) -> impl DoubleEndedIterator<Item = (&'a str, &'a str)> {
        Parameters::iter(self.0)
    }

    pub fn get(&'a self, k: &str) -> Option<&'a str> {
        Parameters::get(self.0, k)
    }

    pub fn values(&'a self, k: &str) -> impl DoubleEndedIterator<Item = &'a str> {
        Parameters::values(self.0, k)
    }
}

impl AsRef<str> for Metadata<'_> {
    fn as_ref(&self) -> &str {
        self.as_str()
    }
}

impl fmt::Display for Metadata<'_> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(self.as_str())
    }
}

impl fmt::Debug for Metadata<'_> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self)
    }
}

#[repr(transparent)]
#[derive(PartialEq, Eq, Hash)]
pub struct MetadataMut<'a>(&'a mut EndPoint);

impl<'a> MetadataMut<'a> {
    pub fn as_str(&'a self) -> &'a str {
        metadata(self.0.as_str())
    }

    pub fn is_empty(&'a self) -> bool {
        self.as_str().is_empty()
    }
}

impl MetadataMut<'_> {
    pub fn extend<I, K, V>(&mut self, iter: I) -> ZResult<()>
    where
        I: Iterator<Item = (K, V)>,
        K: AsRef<str>,
        V: AsRef<str>,
    {
        for (k, v) in iter {
            let k: &str = k.as_ref();
            let v: &str = v.as_ref();
            self.insert(k, v)?
        }
        Ok(())
    }

    pub fn insert(&mut self, k: &str, v: &str) -> ZResult<()> {
        let ep = EndPoint::new(
            self.0.protocol(),
            self.0.address(),
            Parameters::insert(self.0.metadata().iter(), k, v),
            self.0.config(),
        )?;

        self.0.inner = ep.inner;
        Ok(())
    }

    pub fn remove(&mut self, k: &str) -> ZResult<()> {
        let ep = EndPoint::new(
            self.0.protocol(),
            self.0.address(),
            Parameters::remove(self.0.metadata().iter(), k),
            self.0.config(),
        )?;

        self.0.inner = ep.inner;
        Ok(())
    }
}

impl AsRef<str> for MetadataMut<'_> {
    fn as_ref(&self) -> &str {
        self.as_str()
    }
}

impl fmt::Display for MetadataMut<'_> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(self.as_str())
    }
}

impl fmt::Debug for MetadataMut<'_> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self)
    }
}

// Config
#[repr(transparent)]
#[derive(Copy, Clone, PartialEq, Eq, Hash)]
pub struct Config<'a>(pub(super) &'a str);

impl<'a> Config<'a> {
    pub fn as_str(&self) -> &'a str {
        self.0
    }

    pub fn is_empty(&'a self) -> bool {
        self.as_str().is_empty()
    }

    pub fn iter(&'a self) -> impl DoubleEndedIterator<Item = (&'a str, &'a str)> {
        Parameters::iter(self.0)
    }

    pub fn get(&'a self, k: &str) -> Option<&'a str> {
        Parameters::get(self.0, k)
    }

    pub fn values(&'a self, k: &str) -> impl DoubleEndedIterator<Item = &'a str> {
        Parameters::values(self.0, k)
    }
}

impl AsRef<str> for Config<'_> {
    fn as_ref(&self) -> &str {
        self.as_str()
    }
}

impl fmt::Display for Config<'_> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(self.as_str())
    }
}

impl fmt::Debug for Config<'_> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self)
    }
}

#[repr(transparent)]
#[derive(PartialEq, Eq, Hash)]
pub struct ConfigMut<'a>(&'a mut EndPoint);

impl<'a> ConfigMut<'a> {
    pub fn as_str(&'a self) -> &'a str {
        config(self.0.as_str())
    }

    pub fn is_empty(&'a self) -> bool {
        self.as_str().is_empty()
    }
}

impl ConfigMut<'_> {
    pub fn extend<I, K, V>(&mut self, iter: I) -> ZResult<()>
    where
        I: Iterator<Item = (K, V)>,
        K: AsRef<str>,
        V: AsRef<str>,
    {
        for (k, v) in iter {
            let k: &str = k.as_ref();
            let v: &str = v.as_ref();
            self.insert(k, v)?
        }
        Ok(())
    }

    pub fn insert(&mut self, k: &str, v: &str) -> ZResult<()> {
        let ep = EndPoint::new(
            self.0.protocol(),
            self.0.address(),
            self.0.metadata(),
            Parameters::insert(self.0.config().iter(), k, v),
        )?;

        self.0.inner = ep.inner;
        Ok(())
    }

    pub fn remove(&mut self, k: &str) -> ZResult<()> {
        let ep = EndPoint::new(
            self.0.protocol(),
            self.0.address(),
            self.0.metadata(),
            Parameters::remove(self.0.config().iter(), k),
        )?;

        self.0.inner = ep.inner;
        Ok(())
    }
}

impl AsRef<str> for ConfigMut<'_> {
    fn as_ref(&self) -> &str {
        self.as_str()
    }
}

impl fmt::Display for ConfigMut<'_> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(self.as_str())
    }
}

impl fmt::Debug for ConfigMut<'_> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self)
    }
}

/// A `String` that respects the [`EndPoint`] canon form: `<locator>#<config>`, such that `<locator>` is a valid [`Locator`] `<config>` is of the form `<key1>=<value1>;...;<keyN>=<valueN>` where keys are alphabetically sorted.
#[derive(Clone, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
#[serde(into = "String")]
#[serde(try_from = "String")]
pub struct EndPoint {
    pub(super) inner: String,
}

impl EndPoint {
    pub fn new<A, B, C, D>(protocol: A, address: B, metadata: C, config: D) -> ZResult<Self>
    where
        A: AsRef<str>,
        B: AsRef<str>,
        C: AsRef<str>,
        D: AsRef<str>,
    {
        let p: &str = protocol.as_ref();
        let a: &str = address.as_ref();
        let m: &str = metadata.as_ref();
        let c: &str = config.as_ref();

        let len = p.as_bytes().len() + a.as_bytes().len() + m.as_bytes().len();
        if len > u8::MAX as usize {
            bail!("Endpoint too big: {} bytes. Max: {} bytes. ", len, u8::MAX);
        }

        let s = match (m.is_empty(), c.is_empty()) {
            (true, true) => format!("{p}{PROTO_SEPARATOR}{a}"),
            (false, true) => format!("{p}{PROTO_SEPARATOR}{a}{METADATA_SEPARATOR}{m}"),
            (true, false) => format!("{p}{PROTO_SEPARATOR}{a}{CONFIG_SEPARATOR}{c}"),
            (false, false) => {
                format!("{p}{PROTO_SEPARATOR}{a}{METADATA_SEPARATOR}{m}{CONFIG_SEPARATOR}{c}")
            }
        };

        Self::try_from(s)
    }

    pub fn as_str(&self) -> &str {
        self.inner.as_str()
    }

    pub fn split(&self) -> (Protocol, Address, Metadata, Config) {
        (
            self.protocol(),
            self.address(),
            self.metadata(),
            self.config(),
        )
    }

    pub fn protocol(&self) -> Protocol {
        Protocol(protocol(self.inner.as_str()))
    }

    pub fn protocol_mut(&mut self) -> ProtocolMut {
        ProtocolMut(self)
    }

    pub fn address(&self) -> Address {
        Address(address(self.inner.as_str()))
    }

    pub fn address_mut(&mut self) -> AddressMut {
        AddressMut(self)
    }

    pub fn metadata(&self) -> Metadata {
        Metadata(metadata(self.inner.as_str()))
    }

    pub fn metadata_mut(&mut self) -> MetadataMut {
        MetadataMut(self)
    }

    pub fn config(&self) -> Config {
        Config(config(self.inner.as_str()))
    }

    pub fn config_mut(&mut self) -> ConfigMut {
        ConfigMut(self)
    }

    pub fn to_locator(&self) -> Locator {
        self.clone().into()
    }
}

impl From<Locator> for EndPoint {
    fn from(val: Locator) -> Self {
        val.0
    }
}

impl fmt::Display for EndPoint {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(&self.inner)
    }
}

impl fmt::Debug for EndPoint {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self)
    }
}

impl From<EndPoint> for String {
    fn from(v: EndPoint) -> String {
        v.inner
    }
}

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

    fn try_from(s: String) -> Result<Self, Self::Error> {
        const ERR: &str =
            "Endpoints must be of the form <protocol>/<address>[?<metadata>][#<config>]";

        fn sort_hashmap(from: &str, into: &mut String) {
            let mut from = from
                .split(LIST_SEPARATOR)
                .map(|p| split_once(p, FIELD_SEPARATOR))
                .collect::<Vec<(&str, &str)>>();
            from.sort_by(|(k1, _), (k2, _)| k1.cmp(k2));

            let mut first = true;
            for (k, v) in from.iter() {
                if !first {
                    into.push(LIST_SEPARATOR);
                }
                into.push_str(k);
                if !v.is_empty() {
                    into.push(FIELD_SEPARATOR);
                    into.push_str(v);
                }
                first = false;
            }
        }

        let pidx = s
            .find(PROTO_SEPARATOR)
            .and_then(|i| (!s[..i].is_empty() && !s[i + 1..].is_empty()).then_some(i))
            .ok_or_else(|| zerror!("{}: {}", ERR, s))?;

        match (s.find(METADATA_SEPARATOR), s.find(CONFIG_SEPARATOR)) {
            // No metadata or config at all
            (None, None) => Ok(EndPoint { inner: s }),
            // There is some metadata
            (Some(midx), None) if midx > pidx && !s[midx + 1..].is_empty() => {
                let mut inner = String::with_capacity(s.len());
                inner.push_str(&s[..midx + 1]); // Includes metadata separator
                sort_hashmap(&s[midx + 1..], &mut inner);
                Ok(EndPoint { inner })
            }
            // There is some config
            (None, Some(cidx)) if cidx > pidx && !s[cidx + 1..].is_empty() => {
                let mut inner = String::with_capacity(s.len());
                inner.push_str(&s[..cidx + 1]); // Includes config separator
                sort_hashmap(&s[cidx + 1..], &mut inner);
                Ok(EndPoint { inner })
            }
            // There is some metadata and some config
            (Some(midx), Some(cidx))
                if midx > pidx
                    && cidx > midx
                    && !s[midx + 1..cidx].is_empty()
                    && !s[cidx + 1..].is_empty() =>
            {
                let mut inner = String::with_capacity(s.len());
                inner.push_str(&s[..midx + 1]); // Includes metadata separator

                sort_hashmap(&s[midx + 1..cidx], &mut inner);

                inner.push(CONFIG_SEPARATOR);
                sort_hashmap(&s[cidx + 1..], &mut inner);

                Ok(EndPoint { inner })
            }
            _ => Err(zerror!("{}: {}", ERR, s).into()),
        }
    }
}

impl FromStr for EndPoint {
    type Err = ZError;

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

impl EndPoint {
    #[cfg(feature = "test")]
    pub fn rand() -> Self {
        use rand::{
            distributions::{Alphanumeric, DistString},
            rngs::ThreadRng,
            Rng,
        };

        const MIN: usize = 2;
        const MAX: usize = 8;

        fn gen_hashmap(rng: &mut ThreadRng, endpoint: &mut String) {
            let num = rng.gen_range(MIN..MAX);
            for i in 0..num {
                if i != 0 {
                    endpoint.push(LIST_SEPARATOR);
                }
                let len = rng.gen_range(MIN..MAX);
                let key = Alphanumeric.sample_string(rng, len);
                endpoint.push_str(key.as_str());

                endpoint.push(FIELD_SEPARATOR);

                let len = rng.gen_range(MIN..MAX);
                let value = Alphanumeric.sample_string(rng, len);
                endpoint.push_str(value.as_str());
            }
        }

        let mut rng = rand::thread_rng();
        let mut endpoint = String::new();

        let len = rng.gen_range(MIN..MAX);
        let proto = Alphanumeric.sample_string(&mut rng, len);
        endpoint.push_str(proto.as_str());

        endpoint.push(PROTO_SEPARATOR);

        let len = rng.gen_range(MIN..MAX);
        let address = Alphanumeric.sample_string(&mut rng, len);
        endpoint.push_str(address.as_str());

        if rng.gen_bool(0.5) {
            endpoint.push(METADATA_SEPARATOR);
            gen_hashmap(&mut rng, &mut endpoint);
        }
        if rng.gen_bool(0.5) {
            endpoint.push(CONFIG_SEPARATOR);
            gen_hashmap(&mut rng, &mut endpoint);
        }

        endpoint.parse().unwrap()
    }
}

#[test]
fn endpoints() {
    assert!(EndPoint::from_str("/").is_err());
    assert!(EndPoint::from_str("?").is_err());
    assert!(EndPoint::from_str("#").is_err());

    assert!(EndPoint::from_str("udp").is_err());
    assert!(EndPoint::from_str("/udp").is_err());
    assert!(EndPoint::from_str("udp/").is_err());

    assert!(EndPoint::from_str("udp/127.0.0.1:7447?").is_err());
    assert!(EndPoint::from_str("udp?127.0.0.1:7447").is_err());
    assert!(EndPoint::from_str("udp?127.0.0.1:7447/meta").is_err());

    assert!(EndPoint::from_str("udp/127.0.0.1:7447#").is_err());
    assert!(EndPoint::from_str("udp/127.0.0.1:7447?#").is_err());
    assert!(EndPoint::from_str("udp/127.0.0.1:7447#?").is_err());
    assert!(EndPoint::from_str("udp#127.0.0.1:7447/").is_err());
    assert!(EndPoint::from_str("udp#127.0.0.1:7447/?").is_err());
    assert!(EndPoint::from_str("udp/127.0.0.1:7447?a=1#").is_err());

    let endpoint = EndPoint::from_str("udp/127.0.0.1:7447").unwrap();
    assert_eq!(endpoint.as_str(), "udp/127.0.0.1:7447");
    assert_eq!(endpoint.protocol().as_str(), "udp");
    assert_eq!(endpoint.address().as_str(), "127.0.0.1:7447");
    assert!(endpoint.metadata().as_str().is_empty());
    assert_eq!(endpoint.metadata().iter().count(), 0);

    let endpoint = EndPoint::from_str("udp/127.0.0.1:7447?a=1;b=2").unwrap();
    assert_eq!(endpoint.as_str(), "udp/127.0.0.1:7447?a=1;b=2");
    assert_eq!(endpoint.protocol().as_str(), "udp");
    assert_eq!(endpoint.address().as_str(), "127.0.0.1:7447");
    assert_eq!(endpoint.metadata().as_str(), "a=1;b=2");
    assert_eq!(endpoint.metadata().iter().count(), 2);
    endpoint
        .metadata()
        .iter()
        .find(|x| x == &("a", "1"))
        .unwrap();
    assert_eq!(endpoint.metadata().get("a"), Some("1"));
    endpoint
        .metadata()
        .iter()
        .find(|x| x == &("b", "2"))
        .unwrap();
    assert_eq!(endpoint.metadata().get("b"), Some("2"));
    assert!(endpoint.config().as_str().is_empty());
    assert_eq!(endpoint.config().iter().count(), 0);

    let endpoint = EndPoint::from_str("udp/127.0.0.1:7447?b=2;a=1").unwrap();
    assert_eq!(endpoint.as_str(), "udp/127.0.0.1:7447?a=1;b=2");
    assert_eq!(endpoint.protocol().as_str(), "udp");
    assert_eq!(endpoint.address().as_str(), "127.0.0.1:7447");
    assert_eq!(endpoint.metadata().as_str(), "a=1;b=2");
    assert_eq!(endpoint.metadata().iter().count(), 2);
    endpoint
        .metadata()
        .iter()
        .find(|x| x == &("a", "1"))
        .unwrap();
    assert_eq!(endpoint.metadata().get("a"), Some("1"));
    endpoint
        .metadata()
        .iter()
        .find(|x| x == &("b", "2"))
        .unwrap();
    assert_eq!(endpoint.metadata().get("a"), Some("1"));
    assert!(endpoint.config().as_str().is_empty());
    assert_eq!(endpoint.config().iter().count(), 0);

    let endpoint = EndPoint::from_str("udp/127.0.0.1:7447#A=1;B=2").unwrap();
    assert_eq!(endpoint.as_str(), "udp/127.0.0.1:7447#A=1;B=2");
    assert_eq!(endpoint.protocol().as_str(), "udp");
    assert_eq!(endpoint.address().as_str(), "127.0.0.1:7447");
    assert!(endpoint.metadata().as_str().is_empty());
    assert_eq!(endpoint.metadata().iter().count(), 0);
    assert_eq!(endpoint.config().as_str(), "A=1;B=2");
    assert_eq!(endpoint.config().iter().count(), 2);
    endpoint.config().iter().find(|x| x == &("A", "1")).unwrap();
    assert_eq!(endpoint.config().get("A"), Some("1"));
    endpoint.config().iter().find(|x| x == &("B", "2")).unwrap();
    assert_eq!(endpoint.config().get("B"), Some("2"));

    let endpoint = EndPoint::from_str("udp/127.0.0.1:7447#B=2;A=1").unwrap();
    assert_eq!(endpoint.as_str(), "udp/127.0.0.1:7447#A=1;B=2");
    assert_eq!(endpoint.protocol().as_str(), "udp");
    assert_eq!(endpoint.address().as_str(), "127.0.0.1:7447");
    assert!(endpoint.metadata().as_str().is_empty());
    assert_eq!(endpoint.metadata().iter().count(), 0);
    assert_eq!(endpoint.config().as_str(), "A=1;B=2");
    assert_eq!(endpoint.config().iter().count(), 2);
    endpoint.config().iter().find(|x| x == &("A", "1")).unwrap();
    assert_eq!(endpoint.config().get("A"), Some("1"));
    endpoint.config().iter().find(|x| x == &("B", "2")).unwrap();
    assert_eq!(endpoint.config().get("B"), Some("2"));

    let endpoint = EndPoint::from_str("udp/127.0.0.1:7447?a=1;b=2#A=1;B=2").unwrap();
    assert_eq!(endpoint.as_str(), "udp/127.0.0.1:7447?a=1;b=2#A=1;B=2");
    assert_eq!(endpoint.protocol().as_str(), "udp");
    assert_eq!(endpoint.address().as_str(), "127.0.0.1:7447");
    assert_eq!(endpoint.metadata().as_str(), "a=1;b=2");
    assert_eq!(endpoint.metadata().iter().count(), 2);
    endpoint
        .metadata()
        .iter()
        .find(|x| x == &("a", "1"))
        .unwrap();
    assert_eq!(endpoint.metadata().get("a"), Some("1"));
    endpoint
        .metadata()
        .iter()
        .find(|x| x == &("b", "2"))
        .unwrap();
    assert_eq!(endpoint.metadata().get("b"), Some("2"));
    assert_eq!(endpoint.config().as_str(), "A=1;B=2");
    assert_eq!(endpoint.config().iter().count(), 2);
    endpoint.config().iter().find(|x| x == &("A", "1")).unwrap();
    assert_eq!(endpoint.config().get("A"), Some("1"));
    endpoint.config().iter().find(|x| x == &("B", "2")).unwrap();
    assert_eq!(endpoint.config().get("B"), Some("2"));

    let endpoint = EndPoint::from_str("udp/127.0.0.1:7447?b=2;a=1#B=2;A=1").unwrap();
    assert_eq!(endpoint.as_str(), "udp/127.0.0.1:7447?a=1;b=2#A=1;B=2");
    assert_eq!(endpoint.protocol().as_str(), "udp");
    assert_eq!(endpoint.address().as_str(), "127.0.0.1:7447");
    assert_eq!(endpoint.metadata().as_str(), "a=1;b=2");
    assert_eq!(endpoint.metadata().iter().count(), 2);
    endpoint
        .metadata()
        .iter()
        .find(|x| x == &("a", "1"))
        .unwrap();
    assert_eq!(endpoint.metadata().get("a"), Some("1"));
    endpoint
        .metadata()
        .iter()
        .find(|x| x == &("b", "2"))
        .unwrap();
    assert_eq!(endpoint.metadata().get("b"), Some("2"));
    assert_eq!(endpoint.config().as_str(), "A=1;B=2");
    assert_eq!(endpoint.config().iter().count(), 2);
    endpoint.config().iter().find(|x| x == &("A", "1")).unwrap();
    assert_eq!(endpoint.config().get("A"), Some("1"));
    endpoint.config().iter().find(|x| x == &("B", "2")).unwrap();
    assert_eq!(endpoint.config().get("B"), Some("2"));

    let mut endpoint = EndPoint::from_str("udp/127.0.0.1:7447?a=1;b=2").unwrap();
    endpoint.metadata_mut().insert("c", "3").unwrap();
    assert_eq!(endpoint.as_str(), "udp/127.0.0.1:7447?a=1;b=2;c=3");

    let mut endpoint = EndPoint::from_str("udp/127.0.0.1:7447?b=2;c=3").unwrap();
    endpoint.metadata_mut().insert("a", "1").unwrap();
    assert_eq!(endpoint.as_str(), "udp/127.0.0.1:7447?a=1;b=2;c=3");

    let mut endpoint = EndPoint::from_str("udp/127.0.0.1:7447?a=1;b=2").unwrap();
    endpoint.config_mut().insert("A", "1").unwrap();
    assert_eq!(endpoint.as_str(), "udp/127.0.0.1:7447?a=1;b=2#A=1");

    let mut endpoint = EndPoint::from_str("udp/127.0.0.1:7447?b=2;c=3#B=2").unwrap();
    endpoint.config_mut().insert("A", "1").unwrap();
    assert_eq!(endpoint.as_str(), "udp/127.0.0.1:7447?b=2;c=3#A=1;B=2");

    let mut endpoint = EndPoint::from_str("udp/127.0.0.1:7447").unwrap();
    endpoint
        .metadata_mut()
        .extend([("a", "1"), ("c", "3"), ("b", "2")].iter().copied())
        .unwrap();
    assert_eq!(endpoint.as_str(), "udp/127.0.0.1:7447?a=1;b=2;c=3");

    let mut endpoint = EndPoint::from_str("udp/127.0.0.1:7447").unwrap();
    endpoint
        .config_mut()
        .extend([("A", "1"), ("C", "3"), ("B", "2")].iter().copied())
        .unwrap();
    assert_eq!(endpoint.as_str(), "udp/127.0.0.1:7447#A=1;B=2;C=3");

    let endpoint =
        EndPoint::from_str("udp/127.0.0.1:7447#iface=en0;join=224.0.0.1|224.0.0.2|224.0.0.3")
            .unwrap();
    let c = endpoint.config();
    assert_eq!(c.get("iface"), Some("en0"));
    assert_eq!(c.get("join"), Some("224.0.0.1|224.0.0.2|224.0.0.3"));
    assert_eq!(c.values("iface").count(), 1);
    let mut i = c.values("iface");
    assert_eq!(i.next(), Some("en0"));
    assert_eq!(c.values("join").count(), 3);
    let mut i = c.values("join");
    assert_eq!(i.next(), Some("224.0.0.1"));
    assert_eq!(i.next(), Some("224.0.0.2"));
    assert_eq!(i.next(), Some("224.0.0.3"));
    assert_eq!(i.next(), None);
}