mqtt-rs 0.20.2

MQTT driver for epics-rs — publish/subscribe MQTT topics as EPICS records
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111

/// MQTT Quality of Service level.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum QoS {
    AtMostOnce = 0,
    #[default]
    AtLeastOnce = 1,
    ExactlyOnce = 2,
}

impl QoS {
    pub fn from_int(v: i32) -> Self {
        match v {
            0 => Self::AtMostOnce,
            2 => Self::ExactlyOnce,
            _ => Self::AtLeastOnce,
        }
    }
}

impl From<QoS> for rumqttc::QoS {
    fn from(q: QoS) -> Self {
        match q {
            QoS::AtMostOnce => rumqttc::QoS::AtMostOnce,
            QoS::AtLeastOnce => rumqttc::QoS::AtLeastOnce,
            QoS::ExactlyOnce => rumqttc::QoS::ExactlyOnce,
        }
    }
}

/// MQTT driver configuration.
#[derive(Debug, Clone)]
pub struct MqttConfig {
    pub broker_host: String,
    pub broker_port: u16,
    pub client_id: String,
    pub qos: QoS,
    pub keep_alive_secs: u64,
    pub clean_session: bool,
}

impl Default for MqttConfig {
    fn default() -> Self {
        Self {
            broker_host: "localhost".into(),
            broker_port: 1883,
            client_id: "epics-mqtt".into(),
            qos: QoS::default(),
            keep_alive_secs: 20,
            clean_session: true,
        }
    }
}

impl MqttConfig {
    /// Parse a broker URL like "mqtt://host:port" or "host:port" or just "host".
    pub fn parse_broker_url(url: &str) -> (String, u16) {
        let stripped = url
            .strip_prefix("mqtt://")
            .or_else(|| url.strip_prefix("tcp://"))
            .unwrap_or(url);

        match stripped.rsplit_once(':') {
            Some((host, port_str)) => {
                let port = port_str.parse().unwrap_or(1883);
                (host.to_string(), port)
            }
            None => (stripped.to_string(), 1883),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn parse_broker_url_full() {
        let (host, port) = MqttConfig::parse_broker_url("mqtt://broker.local:1884");
        assert_eq!(host, "broker.local");
        assert_eq!(port, 1884);
    }

    #[test]
    fn parse_broker_url_no_scheme() {
        let (host, port) = MqttConfig::parse_broker_url("192.168.1.10:1883");
        assert_eq!(host, "192.168.1.10");
        assert_eq!(port, 1883);
    }

    #[test]
    fn parse_broker_url_host_only() {
        let (host, port) = MqttConfig::parse_broker_url("localhost");
        assert_eq!(host, "localhost");
        assert_eq!(port, 1883);
    }

    #[test]
    fn parse_broker_url_tcp_scheme() {
        let (host, port) = MqttConfig::parse_broker_url("tcp://myhost:9883");
        assert_eq!(host, "myhost");
        assert_eq!(port, 9883);
    }

    #[test]
    fn qos_from_int() {
        assert_eq!(QoS::from_int(0), QoS::AtMostOnce);
        assert_eq!(QoS::from_int(1), QoS::AtLeastOnce);
        assert_eq!(QoS::from_int(2), QoS::ExactlyOnce);
        assert_eq!(QoS::from_int(99), QoS::AtLeastOnce); // fallback
    }
}