Skip to main content

ebus_discovery/
broker.rs

1//! Broker endpoints, ranking and selection (ebus.py:565-795).
2
3use alloc::borrow::Cow;
4use alloc::format;
5use alloc::string::{String, ToString};
6use alloc::vec::Vec;
7use core::fmt;
8use core::net::IpAddr;
9
10use crate::addr::{ScopedAddr, candidate_addresses};
11use crate::instance::Instance;
12use crate::txt::split_list;
13use crate::{BrokerMode, BrokerType, is_local_name, same_host, strip_dot};
14
15/// A broker to connect to, discovered or configured.
16///
17/// Built with [`from_instance`](Self::from_instance),
18/// [`configured`](Self::configured) or [`from_url`](Self::from_url).
19#[derive(Clone, Debug, PartialEq, Eq, Hash)]
20#[non_exhaustive]
21pub struct BrokerEndpoint {
22    /// The transport.
23    pub broker_type: BrokerType,
24    /// The TXT `broker` value, else the SRV target, without a trailing dot
25    /// (ebus.py:599-614); the configured URL's host for a configured broker.
26    pub host: String,
27    /// The port.
28    pub port: u16,
29    /// The SRV target, without a trailing dot.
30    pub server: Option<String>,
31    /// The TXT `device_id` value, as advertised.
32    pub device_id: Option<String>,
33    /// The advertised addresses worth trying, in candidate order
34    /// ([`candidate_addresses`]).
35    pub addrs: Vec<ScopedAddr>,
36    /// The instance name; `None` for a configured broker.
37    pub instance: Option<String>,
38}
39
40/// A configured broker URL that cannot be used.
41#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
42#[non_exhaustive]
43pub enum UrlError {
44    /// A scheme other than `mqtt://` or `mqtts://`.
45    Scheme,
46    /// No host.
47    NoHost,
48    /// A port that is not a number from 0 to 65535.
49    Port,
50}
51
52impl fmt::Display for UrlError {
53    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
54        f.write_str(match self {
55            Self::Scheme => "unsupported broker URL: expected mqtt:// or mqtts://",
56            Self::NoHost => "broker URL has no host",
57            Self::Port => "broker URL has an invalid port",
58        })
59    }
60}
61
62impl core::error::Error for UrlError {}
63
64impl BrokerEndpoint {
65    /// The endpoint a browsed broker instance gives (ebus.py:598-614): `host`
66    /// is the TXT `broker` value, trimmed and with one trailing dot removed,
67    /// else the SRV target; port 0 becomes the type's default port. `None`
68    /// for an instance that is not a broker type.
69    pub fn from_instance(instance: &Instance) -> Option<Self> {
70        let broker_type = BrokerType::parse(&instance.key.service_type)?;
71        let server = strip_dot(&instance.server);
72        let txt_broker = instance.txt.get("broker");
73        let txt_broker = txt_broker.as_deref().map_or("", str::trim);
74        let host = if txt_broker.is_empty() {
75            server
76        } else {
77            strip_dot(txt_broker)
78        };
79        let port = match instance.port {
80            0 => broker_type.default_port(),
81            p => p,
82        };
83        Some(Self {
84            broker_type,
85            host: host.into(),
86            port,
87            server: (!server.is_empty()).then(|| server.into()),
88            device_id: instance
89                .txt
90                .get("device_id")
91                .filter(|d| !d.is_empty())
92                .map(Cow::into_owned),
93            addrs: candidate_addresses(&instance.addrs),
94            instance: Some(instance.key.name.clone()),
95        })
96    }
97
98    /// A configured broker from an `mqtt://` or `mqtts://` URL
99    /// (ebus.py:616-633): the port defaults to 1883 or 8883, and the host is
100    /// lowercased, as python's `urlsplit` gives it, except an IPv6 zone.
101    pub fn from_url(url: &str) -> Result<Self, UrlError> {
102        let url = url.trim();
103        let (scheme, rest) = url.split_once(':').ok_or(UrlError::Scheme)?;
104        let broker_type = if scheme.eq_ignore_ascii_case("mqtt") {
105            BrokerType::Mqtt
106        } else if scheme.eq_ignore_ascii_case("mqtts") {
107            BrokerType::SecureMqtt
108        } else {
109            return Err(UrlError::Scheme);
110        };
111        let netloc = rest.strip_prefix("//").ok_or(UrlError::NoHost)?;
112        let netloc = netloc
113            .find(['/', '?', '#'])
114            .map_or(netloc, |at| &netloc[..at]);
115        let hostinfo = netloc.rsplit_once('@').map_or(netloc, |(_, h)| h);
116        let (host, port) = match hostinfo.split_once('[') {
117            Some((_, bracketed)) => {
118                let (host, after) = bracketed.split_once(']').unwrap_or((bracketed, ""));
119                (host, after.split_once(':').map_or("", |(_, p)| p))
120            }
121            None => hostinfo.split_once(':').unwrap_or((hostinfo, "")),
122        };
123        if host.is_empty() {
124            return Err(UrlError::NoHost);
125        }
126        let port = match port {
127            "" => 0,
128            p if p.bytes().all(|b| b.is_ascii_digit()) => p.parse().map_err(|_| UrlError::Port)?,
129            _ => return Err(UrlError::Port),
130        };
131        // urlsplit lowercases the host but not an IPv6 zone.
132        let host = match host.split_once('%') {
133            Some((h, zone)) => format!("{}%{zone}", h.to_lowercase()),
134            None => host.to_lowercase(),
135        };
136        Ok(Self::configured(
137            broker_type,
138            &host,
139            match port {
140                0 => broker_type.default_port(),
141                p => p,
142            },
143        ))
144    }
145
146    /// A configured broker of `broker_type` at `host` and `port`.
147    pub fn configured(broker_type: BrokerType, host: &str, port: u16) -> Self {
148        Self {
149            broker_type,
150            host: host.into(),
151            port,
152            server: None,
153            device_id: None,
154            addrs: Vec::new(),
155            instance: None,
156        }
157    }
158
159    /// `scheme://host:port`, with an IPv6 host in brackets (ebus.py:592-596).
160    pub fn url(&self) -> String {
161        if self.host.contains(':') {
162            format!(
163                "{}://[{}]:{}",
164                self.broker_type.scheme(),
165                self.host,
166                self.port
167            )
168        } else {
169            format!(
170                "{}://{}:{}",
171                self.broker_type.scheme(),
172                self.host,
173                self.port
174            )
175        }
176    }
177
178    /// The host to connect to: `host` when it is not a `.local` name, else the
179    /// first IPv4 candidate address, else `host`, for TLS and plain transports
180    /// alike.
181    pub fn connect_host(&self) -> Cow<'_, str> {
182        if !is_local_name(&self.host) {
183            return Cow::Borrowed(&self.host);
184        }
185        match self.addrs.iter().find(|a| a.ip.is_ipv4()) {
186            Some(a) => Cow::Owned(a.ip.to_string()),
187            None => Cow::Borrowed(&self.host),
188        }
189    }
190
191    /// The TXT `device_id` items.
192    pub fn device_ids(&self) -> Vec<String> {
193        self.device_id
194            .as_deref()
195            .map(split_list)
196            .unwrap_or_default()
197    }
198}
199
200/// The particular broker a host wants among those discovered.
201#[derive(Clone, Debug, PartialEq, Eq, Hash)]
202pub enum Wanted {
203    /// The broker with this instance name, compared ignoring case.
204    Instance(String),
205    /// The broker whose TXT `device_id` lists this id, under any of its
206    /// transports: an endpoint matches when its own `device_id` lists the
207    /// id, or when a sighted endpoint of the same broker (sharing a host name
208    /// or SRV target), of any broker type, lists it.
209    DeviceId(String),
210}
211
212impl Wanted {
213    /// True when `ep` is the wanted broker. `sighted` is every broker
214    /// endpoint seen, of every type, `ep` among them or not; a
215    /// [`Wanted::DeviceId`] reads the `device_id` of `ep` and of the
216    /// endpoints in `sighted` that share its host name or SRV target.
217    pub fn matches(&self, ep: &BrokerEndpoint, sighted: &[BrokerEndpoint]) -> bool {
218        match self {
219            Self::Instance(name) => ep.instance.as_deref().is_some_and(|i| same_host(i, name)),
220            Self::DeviceId(id) => {
221                let lists = |e: &BrokerEndpoint| e.device_ids().iter().any(|d| d == id);
222                lists(ep) || sighted.iter().any(|s| lists(s) && same_broker(s, ep))
223            }
224        }
225    }
226}
227
228fn norm_host(name: &str) -> String {
229    strip_dot(name).to_lowercase()
230}
231
232/// The normalized host name and SRV target of `ep`, those not empty.
233fn host_keys(ep: &BrokerEndpoint) -> impl Iterator<Item = String> + '_ {
234    core::iter::once(&ep.host)
235        .chain(ep.server.as_ref())
236        .map(|h| norm_host(h))
237        .filter(|h| !h.is_empty())
238}
239
240/// True when `a` and `b` share a host name or SRV target.
241fn same_broker(a: &BrokerEndpoint, b: &BrokerEndpoint) -> bool {
242    host_keys(a).any(|ka| host_keys(b).any(|kb| ka == kb))
243}
244
245/// The names that make two endpoints one broker (ebus.py:665-672).
246fn broker_keys(ep: &BrokerEndpoint) -> Vec<(bool, String)> {
247    let mut keys: Vec<(bool, String)> = host_keys(ep).map(|h| (false, h)).collect();
248    if let Some(d) = &ep.device_id {
249        keys.push((true, d.clone()));
250    }
251    keys
252}
253
254/// One endpoint per distinct broker, ranked in the client's preference order
255/// (ebus.py:675-702).
256///
257/// `order` lists the broker types the client accepts, most preferred first;
258/// a type listed twice keeps its first position. Endpoints of other types are
259/// dropped. The rest are sorted by their type's position in `order`, then
260/// lowercased host, then port. Endpoints sharing a host name, SRV target or
261/// `device_id` are one broker, which keeps its transport earliest in `order`.
262/// With [`BROKER_TYPES`](crate::BROKER_TYPES), narrowed to python's accepted
263/// types, as `order`, the ranking is python's.
264pub fn rank_brokers(
265    endpoints: impl IntoIterator<Item = BrokerEndpoint>,
266    order: &[BrokerType],
267) -> Vec<BrokerEndpoint> {
268    let mut candidates: Vec<(usize, String, BrokerEndpoint)> = endpoints
269        .into_iter()
270        .filter_map(|e| {
271            let at = order.iter().position(|t| *t == e.broker_type)?;
272            Some((at, norm_host(&e.host), e))
273        })
274        .collect();
275    candidates.sort_by(|(ta, ha, a), (tb, hb, b)| (ta, ha, a.port).cmp(&(tb, hb, b.port)));
276    let mut kept: Vec<(BrokerEndpoint, Vec<(bool, String)>)> = Vec::new();
277    for (_, _, ep) in candidates {
278        let keys = broker_keys(&ep);
279        match kept
280            .iter_mut()
281            .find(|(_, seen)| seen.iter().any(|k| keys.contains(k)))
282        {
283            Some((_, seen)) => {
284                for k in keys {
285                    if !seen.contains(&k) {
286                        seen.push(k);
287                    }
288                }
289            }
290            None => kept.push((ep, keys)),
291        }
292    }
293    kept.into_iter().map(|(ep, _)| ep).collect()
294}
295
296fn parse_ip(text: &str) -> Option<IpAddr> {
297    text.split('%').next()?.parse().ok()
298}
299
300/// (index into `ranked`, endpoint to use) for the first match.
301fn find_configured(
302    configured: &BrokerEndpoint,
303    ranked: &[BrokerEndpoint],
304) -> Option<(usize, BrokerEndpoint)> {
305    let want_ip = parse_ip(&configured.host);
306    ranked.iter().enumerate().find_map(|(i, ep)| {
307        let by_name = same_host(&configured.host, &ep.host)
308            || ep
309                .server
310                .as_deref()
311                .is_some_and(|s| same_host(&configured.host, s));
312        if by_name {
313            return Some((i, ep.clone()));
314        }
315        let ip = want_ip?;
316        if !ep.addrs.iter().any(|a| a.ip == ip) {
317            return None;
318        }
319        let mut use_ep = ep.clone();
320        // A TLS client verifies the certificate against the discovered name.
321        if !ep.broker_type.uses_tls() {
322            use_ep.host.clone_from(&configured.host);
323        }
324        Some((i, use_ep))
325    })
326}
327
328/// The discovered broker that is the configured one (ebus.py:712-742).
329///
330/// A discovered broker matches when the configured host equals its TXT
331/// `broker` name or SRV target (case-insensitive, trailing dot ignored), or
332/// when a configured IP address equals one of its candidate addresses. The
333/// first match in `ranked` is returned with its discovered port and
334/// transport; a match by address keeps the configured address as `host` on
335/// a plain transport and the discovered name on TLS.
336pub fn match_configured(
337    configured: &BrokerEndpoint,
338    ranked: &[BrokerEndpoint],
339) -> Option<BrokerEndpoint> {
340    find_configured(configured, ranked).map(|(_, ep)| ep)
341}
342
343/// What [`select_broker`] chose.
344pub(crate) enum Choice {
345    None,
346    Configured,
347    /// The endpoint to use, and the index of the discovered one it came from.
348    Discovered(usize, BrokerEndpoint),
349}
350
351pub(crate) fn choose(
352    mode: BrokerMode,
353    configured: Option<&BrokerEndpoint>,
354    ranked: &[BrokerEndpoint],
355    allow_unmatched: bool,
356) -> Choice {
357    let fallback = |c: Option<&BrokerEndpoint>| {
358        if c.is_some() {
359            Choice::Configured
360        } else {
361            Choice::None
362        }
363    };
364    match mode {
365        BrokerMode::ConfiguredOnly => fallback(configured),
366        _ if ranked.is_empty() => match mode {
367            BrokerMode::DiscoveryWithFallback => fallback(configured),
368            _ => Choice::None,
369        },
370        BrokerMode::DiscoveryWithFallback if configured.is_some() => {
371            match configured.and_then(|c| find_configured(c, ranked)) {
372                Some((i, ep)) => Choice::Discovered(i, ep),
373                None if allow_unmatched => Choice::Discovered(0, ranked[0].clone()),
374                None => Choice::Configured,
375            }
376        }
377        _ => Choice::Discovered(0, ranked[0].clone()),
378    }
379}
380
381/// Log a choice as python's `select_broker` does (ebus.py:782-794).
382pub(crate) fn log_choice(
383    choice: &Choice,
384    configured: Option<&BrokerEndpoint>,
385    ranked: &[BrokerEndpoint],
386) {
387    let (picked, chosen_url) = match (choice, configured) {
388        (Choice::Discovered(i, ep), _) => (Some(*i), ep.url()),
389        (Choice::Configured, Some(c)) => (None, c.url()),
390        _ => return,
391    };
392    if ranked.is_empty() {
393        return;
394    }
395    for (i, ep) in ranked.iter().enumerate() {
396        if Some(i) != picked {
397            crate::log::info!("brokerNotChosen", url = ep.url(), chosen = chosen_url);
398        }
399    }
400    if picked.is_none() {
401        crate::log::info!("configuredBrokerNotDiscovered", url = chosen_url);
402    } else if ranked.len() > 1 {
403        let all: Vec<String> = ranked.iter().map(BrokerEndpoint::url).collect();
404        crate::log::info!(
405            "brokerChosenAmongSeveral",
406            count = ranked.len(),
407            chosen = chosen_url,
408            candidates = all.join("|")
409        );
410    }
411}
412
413/// Choose the broker to connect to (ebus.py:745-795).
414///
415/// - `ConfiguredOnly`: `configured`.
416/// - `DiscoveryOnly`: the first of `ranked`, or `None`.
417/// - `DiscoveryWithFallback`: the discovered broker [`match_configured`]
418///   pairs with `configured`, else `configured`; an unmatched discovered
419///   broker, the first of `ranked`, only with `allow_unmatched`. With no
420///   `configured`, the first of `ranked`.
421///
422/// Each discovered broker not chosen is logged with `reason=brokerNotChosen`.
423pub fn select_broker(
424    mode: BrokerMode,
425    configured: Option<&BrokerEndpoint>,
426    ranked: &[BrokerEndpoint],
427    allow_unmatched: bool,
428) -> Option<BrokerEndpoint> {
429    let choice = choose(mode, configured, ranked, allow_unmatched);
430    log_choice(&choice, configured, ranked);
431    match choice {
432        Choice::None => None,
433        Choice::Configured => configured.cloned(),
434        Choice::Discovered(_, ep) => Some(ep),
435    }
436}
437
438#[cfg(test)]
439mod tests {
440    use super::*;
441    use crate::instance::InstanceKey;
442    use crate::txt::TxtRecord;
443    use crate::{BROKER_TYPES, TCP_BROKER_TYPES};
444    use alloc::vec;
445
446    fn inst(
447        t: BrokerType,
448        name: &str,
449        server: &str,
450        port: u16,
451        txt: &[&str],
452        addrs: &[&str],
453    ) -> Instance {
454        Instance {
455            key: InstanceKey {
456                service_type: t.as_str().into(),
457                name: name.into(),
458            },
459            server: server.into(),
460            port,
461            txt: TxtRecord::from_strings(txt),
462            addrs: addrs
463                .iter()
464                .map(|a| ScopedAddr::new(a.parse().unwrap()))
465                .collect(),
466        }
467    }
468
469    fn ep(t: BrokerType, name: &str, server: &str, txt: &[&str], addrs: &[&str]) -> BrokerEndpoint {
470        BrokerEndpoint::from_instance(&inst(t, name, server, 0, txt, addrs)).unwrap()
471    }
472
473    #[test]
474    fn from_instance() {
475        let e = BrokerEndpoint::from_instance(&inst(
476            BrokerType::SecureMqtt,
477            "panel",
478            "Panel.local.",
479            0,
480            &["broker= cert.example.com. ", "device_id=p1"],
481            &["127.0.0.1", "192.168.1.5"],
482        ))
483        .unwrap();
484        assert_eq!(e.host, "cert.example.com");
485        assert_eq!(e.server.as_deref(), Some("Panel.local"));
486        assert_eq!(e.port, 8883);
487        assert_eq!(e.device_id.as_deref(), Some("p1"));
488        assert_eq!(e.addrs, [ScopedAddr::new("192.168.1.5".parse().unwrap())]);
489        assert_eq!(e.instance.as_deref(), Some("panel"));
490        assert_eq!(e.url(), "mqtts://cert.example.com:8883");
491        let mut other = inst(BrokerType::Mqtt, "x", "", 1884, &[], &[]);
492        other.key.service_type = "_ebus._tcp".into();
493        assert_eq!(BrokerEndpoint::from_instance(&other), None);
494    }
495
496    #[test]
497    fn from_url() {
498        let e = BrokerEndpoint::from_url(" MQTTS://user:pw@Broker.Local:8884/x?y ").unwrap();
499        assert_eq!(
500            (e.broker_type, e.host.as_str(), e.port),
501            (BrokerType::SecureMqtt, "broker.local", 8884)
502        );
503        assert_eq!(e.instance, None);
504        let e = BrokerEndpoint::from_url("mqtt://[FE80::1%25En0]").unwrap();
505        assert_eq!((e.host.as_str(), e.port), ("fe80::1%25En0", 1883));
506        assert_eq!(e.url(), "mqtt://[fe80::1%25En0]:1883");
507        assert_eq!(BrokerEndpoint::from_url("mqtt://h:0").unwrap().port, 1883);
508        assert_eq!(BrokerEndpoint::from_url("mqtt://h:").unwrap().port, 1883);
509        assert_eq!(BrokerEndpoint::from_url("ws://h"), Err(UrlError::Scheme));
510        assert_eq!(BrokerEndpoint::from_url("h:1883"), Err(UrlError::Scheme));
511        assert_eq!(
512            BrokerEndpoint::from_url("mqtt://:1883"),
513            Err(UrlError::NoHost)
514        );
515        assert_eq!(BrokerEndpoint::from_url("mqtt:h"), Err(UrlError::NoHost));
516        assert_eq!(
517            BrokerEndpoint::from_url("mqtt://h:65536"),
518            Err(UrlError::Port)
519        );
520        assert_eq!(BrokerEndpoint::from_url("mqtt://h:x"), Err(UrlError::Port));
521        assert_eq!(BrokerEndpoint::from_url("mqtt://h:+1"), Err(UrlError::Port));
522    }
523
524    #[test]
525    fn connect_host() {
526        let dns = ep(
527            BrokerType::SecureMqtt,
528            "a",
529            "a.local.",
530            &["broker=mqtt.example.com"],
531            &["192.168.1.5"],
532        );
533        assert_eq!(dns.connect_host(), "mqtt.example.com");
534        for t in [BrokerType::SecureMqtt, BrokerType::Mqtt] {
535            let local = ep(
536                t,
537                "a",
538                "a.local.",
539                &[],
540                &["fd00::5", "192.168.1.5", "10.0.0.2"],
541            );
542            assert_eq!(local.connect_host(), "10.0.0.2");
543            let v6 = ep(t, "a", "a.local.", &[], &["fd00::5"]);
544            assert_eq!(v6.connect_host(), "a.local");
545        }
546        let configured = BrokerEndpoint::from_url("mqtts://b.local").unwrap();
547        assert_eq!(configured.connect_host(), "b.local");
548    }
549
550    #[test]
551    fn ranking_merges_transports_and_device_ids() {
552        let ranked = rank_brokers(
553            vec![
554                ep(BrokerType::Mqtt, "b", "b.local.", &[], &[]),
555                ep(BrokerType::Mqtt, "a", "A.local.", &[], &[]),
556                ep(
557                    BrokerType::SecureMqtt,
558                    "a",
559                    "a.local.",
560                    &["device_id=x"],
561                    &[],
562                ),
563                ep(
564                    BrokerType::SecureMqtt,
565                    "c",
566                    "c.local.",
567                    &["broker=c.example.com", "device_id=x"],
568                    &[],
569                ),
570                ep(BrokerType::MqttWs, "d", "d.local.", &[], &[]),
571            ],
572            BROKER_TYPES,
573        );
574        let got: Vec<String> = ranked.iter().map(BrokerEndpoint::url).collect();
575        assert_eq!(
576            got,
577            [
578                "mqtts://a.local:8883",
579                "ws://d.local:9001",
580                "mqtt://b.local:1883"
581            ]
582        );
583        let tcp = rank_brokers(ranked, TCP_BROKER_TYPES);
584        assert_eq!(tcp.len(), 2);
585    }
586
587    #[test]
588    fn ranking_orders_by_host_then_port() {
589        let mut a = ep(BrokerType::Mqtt, "a2", "a2.local.", &[], &[]);
590        a.port = 1884;
591        a.host = "a.local".into();
592        let b = ep(BrokerType::Mqtt, "b", "B.local.", &[], &[]);
593        let ranked = rank_brokers(vec![b, a], BROKER_TYPES);
594        assert_eq!(ranked[0].host, "a.local");
595    }
596
597    #[test]
598    fn match_by_name_and_address() {
599        let ranked = vec![
600            ep(
601                BrokerType::SecureMqtt,
602                "s",
603                "s.local.",
604                &[],
605                &["192.168.1.7"],
606            ),
607            ep(BrokerType::Mqtt, "p", "p.local.", &[], &["192.168.1.8"]),
608        ];
609        let by_name = BrokerEndpoint::from_url("mqtt://P.LOCAL.").unwrap();
610        assert_eq!(
611            match_configured(&by_name, &ranked)
612                .unwrap()
613                .instance
614                .as_deref(),
615            Some("p")
616        );
617        let tls_ip = BrokerEndpoint::from_url("mqtt://192.168.1.7").unwrap();
618        let m = match_configured(&tls_ip, &ranked).unwrap();
619        assert_eq!(
620            (m.host.as_str(), m.port, m.broker_type),
621            ("s.local", 8883, BrokerType::SecureMqtt)
622        );
623        let plain_ip = BrokerEndpoint::from_url("mqtts://192.168.1.8:9999").unwrap();
624        let m = match_configured(&plain_ip, &ranked).unwrap();
625        assert_eq!(
626            (m.host.as_str(), m.port, m.broker_type),
627            ("192.168.1.8", 1883, BrokerType::Mqtt)
628        );
629        let none = BrokerEndpoint::from_url("mqtt://other").unwrap();
630        assert_eq!(match_configured(&none, &ranked), None);
631    }
632
633    #[test]
634    fn selection_modes() {
635        let ranked = vec![
636            ep(BrokerType::Mqtt, "a", "a.local.", &[], &[]),
637            ep(BrokerType::Mqtt, "b", "b.local.", &[], &[]),
638        ];
639        let conf_b = BrokerEndpoint::from_url("mqtt://b.local").unwrap();
640        let conf_x = BrokerEndpoint::from_url("mqtt://x.local").unwrap();
641        let pick = |m, c: Option<&BrokerEndpoint>, r: &[BrokerEndpoint], u| {
642            select_broker(m, c, r, u).map(|e| e.host)
643        };
644        use BrokerMode::*;
645        assert_eq!(
646            pick(ConfiguredOnly, Some(&conf_x), &ranked, false).as_deref(),
647            Some("x.local")
648        );
649        assert_eq!(pick(ConfiguredOnly, None, &ranked, false), None);
650        assert_eq!(
651            pick(DiscoveryOnly, Some(&conf_b), &ranked, false).as_deref(),
652            Some("a.local")
653        );
654        assert_eq!(pick(DiscoveryOnly, None, &[], false), None);
655        assert_eq!(
656            pick(DiscoveryWithFallback, Some(&conf_b), &ranked, false).as_deref(),
657            Some("b.local")
658        );
659        assert_eq!(
660            pick(DiscoveryWithFallback, Some(&conf_x), &ranked, false).as_deref(),
661            Some("x.local")
662        );
663        assert_eq!(
664            pick(DiscoveryWithFallback, Some(&conf_x), &ranked, true).as_deref(),
665            Some("a.local")
666        );
667        assert_eq!(
668            pick(DiscoveryWithFallback, Some(&conf_x), &[], true).as_deref(),
669            Some("x.local")
670        );
671        assert_eq!(
672            pick(DiscoveryWithFallback, None, &ranked, false).as_deref(),
673            Some("a.local")
674        );
675        assert_eq!(pick(DiscoveryWithFallback, None, &[], false), None);
676    }
677
678    #[test]
679    fn ranking_follows_the_client_order() {
680        let eps = || {
681            vec![
682                ep(BrokerType::SecureMqtt, "a", "a.local.", &[], &[]),
683                ep(BrokerType::Mqtt, "a", "a.local.", &[], &[]),
684                ep(BrokerType::SecureMqtt, "b", "b.local.", &[], &[]),
685                ep(BrokerType::Mqtt, "c", "c.local.", &[], &[]),
686            ]
687        };
688        let urls = |order: &[BrokerType]| -> Vec<String> {
689            rank_brokers(eps(), order)
690                .iter()
691                .map(BrokerEndpoint::url)
692                .collect()
693        };
694        assert_eq!(
695            urls(&[BrokerType::Mqtt, BrokerType::SecureMqtt]),
696            [
697                "mqtt://a.local:1883",
698                "mqtt://c.local:1883",
699                "mqtts://b.local:8883"
700            ]
701        );
702        assert_eq!(
703            urls(TCP_BROKER_TYPES),
704            [
705                "mqtts://a.local:8883",
706                "mqtts://b.local:8883",
707                "mqtt://c.local:1883"
708            ]
709        );
710        assert_eq!(
711            urls(&[BrokerType::Mqtt, BrokerType::SecureMqtt, BrokerType::Mqtt]),
712            urls(&[BrokerType::Mqtt, BrokerType::SecureMqtt])
713        );
714        assert!(urls(&[]).is_empty());
715    }
716
717    #[test]
718    fn wanted() {
719        let e = ep(
720            BrokerType::SecureMqtt,
721            "Panel-A",
722            "a.local.",
723            &["device_id=p1, p2"],
724            &[],
725        );
726        assert!(Wanted::Instance("panel-a".into()).matches(&e, &[]));
727        assert!(Wanted::DeviceId("p2".into()).matches(&e, &[]));
728        assert!(!Wanted::DeviceId("p".into()).matches(&e, &[]));
729        assert!(!Wanted::Instance("panel".into()).matches(&e, &[]));
730    }
731
732    #[test]
733    fn device_id_on_any_type_and_from_the_same_broker() {
734        let p1 = Wanted::DeviceId("p1".into());
735        // A plain record's own device_id.
736        let plain = ep(BrokerType::Mqtt, "a", "a.local.", &["device_id=p1"], &[]);
737        assert!(p1.matches(&plain, &[]));
738        // The id on a sibling of the same SRV target, the sibling's TXT
739        // `broker` naming another host.
740        let bare = ep(BrokerType::Mqtt, "a", "A.local.", &[], &[]);
741        let secure = ep(
742            BrokerType::SecureMqtt,
743            "a",
744            "a.local",
745            &["broker=cert.example.com", "device_id=p1"],
746            &[],
747        );
748        let ws = ep(BrokerType::MqttWs, "w", "a.local.", &["device_id=p1"], &[]);
749        assert!(!p1.matches(&bare, &[]));
750        assert!(p1.matches(&bare, &[bare.clone(), secure.clone()]));
751        assert!(p1.matches(&bare, core::slice::from_ref(&ws)));
752        // A sibling by TXT `broker` name.
753        let named = ep(
754            BrokerType::Mqtt,
755            "n",
756            "n.local.",
757            &["broker=cert.example.com"],
758            &[],
759        );
760        assert!(p1.matches(&named, core::slice::from_ref(&secure)));
761        // Another broker's id, and a different id on the same broker.
762        let other = ep(
763            BrokerType::SecureMqtt,
764            "o",
765            "o.local.",
766            &["device_id=p1"],
767            &[],
768        );
769        assert!(!p1.matches(&bare, core::slice::from_ref(&other)));
770        assert!(!Wanted::DeviceId("p2".into()).matches(&bare, &[secure]));
771        assert!(!Wanted::Instance("x".into()).matches(&bare, &[ws]));
772    }
773}