mqtt_client/client/
types.rs

1//! Public data types for the client API: connection configuration,
2//! results, delivered messages, and the foreign-callback listener trait.
3//! No I/O or connection-state logic lives here — see
4//! [`crate::client::inner`] and [`crate::client::io`].
5
6use std::time::Duration;
7
8use crate::protocol::packet::MAX_PACKET_SIZE;
9use crate::protocol::{MqttVersion, QoS};
10use crate::support::secs_or;
11
12/// Default time to wait for a broker acknowledgement (PUBACK/PUBREC/
13/// PUBCOMP/SUBACK/UNSUBACK) before returning [`crate::error::MqttError::Timeout`].
14pub(super) const DEFAULT_OPERATION_TIMEOUT_SECS: u32 = 15;
15
16/// Last-Will-and-Testament configuration for a [`ConnectOptions`].
17#[derive(Debug, Clone, PartialEq, Eq, uniffi::Record)]
18pub struct WillOptions {
19    pub topic: String,
20    pub payload: Vec<u8>,
21    pub qos: QoS,
22    pub retain: bool,
23}
24
25/// TLS configuration for [`ConnectOptions::tls`]. All certificate/key
26/// material is PEM-encoded bytes (not file paths) so this works
27/// identically across every language binding without assuming a
28/// filesystem layout.
29#[derive(Clone, PartialEq, Eq, uniffi::Record)]
30pub struct TlsOptions {
31    /// Additional PEM-encoded root CA certificate(s) to trust, on top of
32    /// the bundled Mozilla root store (`webpki-roots`). Needed to connect
33    /// to a broker with a self-signed or private-CA certificate.
34    pub ca_cert_pem: Option<Vec<u8>>,
35    /// PEM-encoded client certificate, for mutual TLS (mTLS). Requires
36    /// [`Self::client_key_pem`] too.
37    pub client_cert_pem: Option<Vec<u8>>,
38    /// PEM-encoded client private key, for mutual TLS. Requires
39    /// [`Self::client_cert_pem`] too.
40    pub client_key_pem: Option<Vec<u8>>,
41    /// Skip server certificate verification entirely. **Dangerous** — only
42    /// for local development against a broker with a self-signed
43    /// certificate you can't otherwise easily trust (e.g. one generated
44    /// on the fly for a demo). Never use this in production: it makes the
45    /// connection trivially interceptable.
46    pub insecure_skip_certificate_verification: bool,
47}
48
49// Never print the client private key (or a CA bundle's presence-only detail
50// beyond "is set") when a `TlsOptions` is `Debug`-formatted.
51crate::redacted_debug!(TlsOptions {
52    ca_cert_pem, client_cert_pem, insecure_skip_certificate_verification
53} secret { client_key_pem });
54
55impl TlsOptions {
56    /// TLS with the bundled Mozilla root store and no client certificate —
57    /// the common case of connecting to a broker with a certificate from a
58    /// well-known public CA.
59    pub fn new() -> Self {
60        TlsOptions {
61            ca_cert_pem: None,
62            client_cert_pem: None,
63            client_key_pem: None,
64            insecure_skip_certificate_verification: false,
65        }
66    }
67}
68
69impl Default for TlsOptions {
70    fn default() -> Self {
71        Self::new()
72    }
73}
74
75/// Everything needed to establish an MQTT connection.
76#[derive(Clone, PartialEq, Eq, uniffi::Record)]
77pub struct ConnectOptions {
78    pub host: String,
79    pub port: u16,
80    pub client_id: String,
81    /// Which protocol version to speak. The broker never up/downgrades a
82    /// connection once CONNECT is sent.
83    pub version: MqttVersion,
84    /// MQTT 3.1.1 "Clean Session" / MQTT 5.0 "Clean Start".
85    pub clean_start: bool,
86    /// Keep-alive interval in seconds. `0` disables keep-alive pings.
87    pub keep_alive_secs: u16,
88    pub username: Option<String>,
89    pub password: Option<Vec<u8>>,
90    pub will: Option<WillOptions>,
91    /// Seconds to wait for the TCP connection + CONNACK before failing.
92    pub connect_timeout_secs: u32,
93    /// Seconds to wait for a QoS 1/2 handshake step *per attempt* before
94    /// retrying (with DUP=1) — `MqttClient::publish` retries an unacked
95    /// QoS 1/2 PUBLISH (or QoS 2's PUBREL) up to 3 times before giving up,
96    /// so the worst-case total wait for a single `publish()` call is up to
97    /// 4x this value. `0` selects the built-in default (15s).
98    pub operation_timeout_secs: u32,
99    /// If true, [`crate::client::MqttClient`] automatically reconnects
100    /// (with exponential backoff, see [`Self::reconnect_backoff_secs`] /
101    /// [`Self::reconnect_max_backoff_secs`]) after a connection loss that
102    /// wasn't caused by calling `disconnect()`, replaying every topic this
103    /// client is currently subscribed to (per its own bookkeeping) once
104    /// reconnected. Off by default — the caller must opt in.
105    pub auto_reconnect: bool,
106    /// Initial delay before the first reconnect attempt, doubling after
107    /// each failed attempt up to [`Self::reconnect_max_backoff_secs`].
108    /// `0` selects the built-in default (1s). Ignored unless
109    /// [`Self::auto_reconnect`] is set.
110    pub reconnect_backoff_secs: u32,
111    /// Cap on the exponential reconnect backoff delay. `0` selects the
112    /// built-in default (30s). Ignored unless [`Self::auto_reconnect`] is
113    /// set.
114    pub reconnect_max_backoff_secs: u32,
115    /// If set, connect over TLS (`mqtts`) instead of plain TCP. `None`
116    /// (the default) is plain TCP, unchanged from before TLS support
117    /// existed.
118    pub tls: Option<TlsOptions>,
119    /// Largest packet (fixed header + body, in bytes) this client will
120    /// accept from the broker; a larger one is treated as a protocol error
121    /// and drops the connection instead of being buffered. `0` selects the
122    /// built-in default (the MQTT protocol maximum, ~256 MiB).
123    #[uniffi(default = 0)]
124    pub max_packet_size: u32,
125    /// MQTT 5.0 enhanced authentication (SCRAM, Kerberos, OAuth, …): the
126    /// `Authentication Method` to announce in CONNECT. The broker may then
127    /// send challenges, which are answered by the
128    /// [`MqttAuthHandler`] registered with `MqttClient::set_auth_handler`.
129    /// Requires [`MqttVersion::V5`].
130    #[uniffi(default = None)]
131    pub auth_method: Option<String>,
132    /// Initial `Authentication Data` sent with CONNECT (e.g. a SCRAM
133    /// client-first message). Requires [`Self::auth_method`].
134    #[uniffi(default = None)]
135    pub auth_data: Option<Vec<u8>>,
136}
137
138// Hand-rolled `Debug` so the password never reaches a log line.
139crate::redacted_debug!(ConnectOptions {
140    host, port, client_id, version, clean_start, keep_alive_secs, username, will,
141    connect_timeout_secs, operation_timeout_secs, auto_reconnect,
142    reconnect_backoff_secs, reconnect_max_backoff_secs, tls, max_packet_size, auth_method
143} secret { password, auth_data });
144
145impl ConnectOptions {
146    pub fn new(
147        host: impl Into<String>,
148        port: u16,
149        client_id: impl Into<String>,
150        version: MqttVersion,
151    ) -> Self {
152        ConnectOptions {
153            host: host.into(),
154            port,
155            client_id: client_id.into(),
156            version,
157            clean_start: true,
158            keep_alive_secs: 30,
159            username: None,
160            password: None,
161            will: None,
162            connect_timeout_secs: 10,
163            operation_timeout_secs: 0,
164            auto_reconnect: false,
165            reconnect_backoff_secs: 0,
166            reconnect_max_backoff_secs: 0,
167            tls: None,
168            max_packet_size: 0,
169            auth_method: None,
170            auth_data: None,
171        }
172    }
173
174    pub(super) fn operation_timeout(&self) -> Duration {
175        secs_or(self.operation_timeout_secs, DEFAULT_OPERATION_TIMEOUT_SECS)
176    }
177
178    pub(super) fn connect_timeout(&self) -> Duration {
179        // A zero connect timeout would fail instantly; treat it as "1s".
180        secs_or(self.connect_timeout_secs, 1)
181    }
182
183    pub(super) fn initial_reconnect_backoff(&self) -> Duration {
184        secs_or(self.reconnect_backoff_secs, 1)
185    }
186
187    pub(super) fn max_reconnect_backoff(&self) -> Duration {
188        secs_or(self.reconnect_max_backoff_secs, 30)
189    }
190
191    pub(super) fn max_packet_size(&self) -> usize {
192        match self.max_packet_size {
193            0 => MAX_PACKET_SIZE,
194            n => n as usize,
195        }
196    }
197
198    /// Reject configurations that can never work, before touching the
199    /// network, with a message naming the offending field.
200    pub(super) fn validate(&self) -> crate::error::MqttResult<()> {
201        use crate::error::MqttError::Protocol;
202        use crate::protocol::topic::is_valid_topic_name;
203        if self.host.is_empty() {
204            return Err(Protocol("ConnectOptions.host must not be empty".into()));
205        }
206        if self.client_id.len() > usize::from(u16::MAX) {
207            return Err(Protocol("ConnectOptions.client_id is too long".into()));
208        }
209        if self.password.is_some() && self.username.is_none() && !self.version.is_v5() {
210            // MQTT-3.1.2-22: 3.1.1 forbids a password without a username.
211            return Err(Protocol(
212                "ConnectOptions.password requires a username in MQTT 3.1.1".into(),
213            ));
214        }
215        if self.auth_method.is_some() && !self.version.is_v5() {
216            return Err(Protocol(
217                "enhanced authentication (auth_method) requires MQTT 5.0".into(),
218            ));
219        }
220        if self.auth_data.is_some() && self.auth_method.is_none() {
221            return Err(Protocol("auth_data requires auth_method".into()));
222        }
223        if let Some(will) = &self.will
224            && !is_valid_topic_name(&will.topic)
225        {
226            return Err(Protocol(format!(
227                "invalid will topic {:?}: must be non-empty and wildcard-free",
228                will.topic
229            )));
230        }
231        Ok(())
232    }
233}
234
235/// Outcome of a successful [`crate::client::MqttClient::connect`].
236#[derive(Debug, Clone, Copy, PartialEq, Eq, uniffi::Record)]
237pub struct ConnectResult {
238    pub session_present: bool,
239    /// `0x00` on success. MQTT 5.0 reason code (MQTT 3.1.1 codes are mapped
240    /// onto the equivalent 5.0 code by
241    /// [`ConnectReasonCode`](crate::protocol::connect::ConnectReasonCode)).
242    pub reason_code: u8,
243}
244
245/// A message delivered to a subscriber.
246#[derive(Debug, Clone, PartialEq, Eq, uniffi::Record)]
247pub struct MqttMessage {
248    pub topic: String,
249    pub payload: Vec<u8>,
250    pub qos: QoS,
251    pub retain: bool,
252}
253
254impl From<&crate::protocol::publish::PublishPacket> for MqttMessage {
255    fn from(p: &crate::protocol::publish::PublishPacket) -> Self {
256        MqttMessage {
257            topic: p.topic.clone(),
258            payload: p.payload.to_vec(),
259            qos: p.qos,
260            retain: p.retain,
261        }
262    }
263}
264
265/// Result of a subscribe request: the reason/return code the broker
266/// granted for the requested filter (`< 0x80` is success; MQTT 3.1.1 codes
267/// 0/1/2 map onto the granted QoS).
268#[derive(Debug, Clone, Copy, PartialEq, Eq, uniffi::Record)]
269pub struct SubscribeResult {
270    pub reason_code: u8,
271}
272
273/// Answers the broker's authentication challenges during an MQTT 5.0
274/// enhanced-authentication handshake (see [`ConnectOptions::auth_method`]).
275#[uniffi::export(with_foreign)]
276pub trait MqttAuthHandler: Send + Sync {
277    /// The broker sent `challenge` for `method`; return the response, or
278    /// `None` to abort the connection attempt.
279    fn respond(&self, method: String, challenge: Vec<u8>) -> Option<Vec<u8>>;
280}
281
282/// Callback interface implemented by foreign code to receive events from an
283/// [`crate::client::MqttClient`] without polling.
284#[uniffi::export(with_foreign)]
285pub trait MqttMessageListener: Send + Sync {
286    /// Called for every PUBLISH delivered to a subscription this client holds.
287    fn on_message(&self, message: MqttMessage);
288    /// Called when the connection to the broker is lost or closed, whether
289    /// by the client, the broker, or the network.
290    fn on_disconnected(&self, reason: String);
291}
292
293#[cfg(test)]
294mod tests {
295    use super::*;
296
297    fn opts() -> ConnectOptions {
298        ConnectOptions::new("localhost", 1883, "cid", MqttVersion::V5)
299    }
300
301    #[test]
302    fn defaults_are_sane() {
303        let o = opts();
304        assert!(o.clean_start && !o.auto_reconnect);
305        assert_eq!(o.keep_alive_secs, 30);
306        assert_eq!(o.operation_timeout(), Duration::from_secs(15));
307        assert_eq!(o.initial_reconnect_backoff(), Duration::from_secs(1));
308        assert_eq!(o.max_reconnect_backoff(), Duration::from_secs(30));
309        assert_eq!(o.max_packet_size(), MAX_PACKET_SIZE);
310        assert!(o.validate().is_ok());
311    }
312
313    #[test]
314    fn explicit_values_override_the_zero_defaults() {
315        let mut o = opts();
316        o.operation_timeout_secs = 3;
317        o.reconnect_backoff_secs = 2;
318        o.reconnect_max_backoff_secs = 9;
319        o.max_packet_size = 4096;
320        assert_eq!(o.operation_timeout(), Duration::from_secs(3));
321        assert_eq!(o.initial_reconnect_backoff(), Duration::from_secs(2));
322        assert_eq!(o.max_reconnect_backoff(), Duration::from_secs(9));
323        assert_eq!(o.max_packet_size(), 4096);
324    }
325
326    #[test]
327    fn a_zero_connect_timeout_is_one_second_not_instant_failure() {
328        let mut o = opts();
329        o.connect_timeout_secs = 0;
330        assert_eq!(o.connect_timeout(), Duration::from_secs(1));
331    }
332
333    #[test]
334    fn validation_rejects_each_unusable_configuration() {
335        type Mutation = Box<dyn Fn(&mut ConnectOptions)>;
336        let cases: Vec<(&str, Mutation)> = vec![
337            ("host", Box::new(|o| o.host.clear())),
338            ("client_id", Box::new(|o| o.client_id = "x".repeat(70_000))),
339            (
340                "password requires a username",
341                Box::new(|o| {
342                    o.version = MqttVersion::V311;
343                    o.password = Some(b"pw".to_vec());
344                }),
345            ),
346            (
347                "will topic",
348                Box::new(|o| {
349                    o.will = Some(WillOptions {
350                        topic: "a/+".into(),
351                        payload: vec![],
352                        qos: QoS::AtMostOnce,
353                        retain: false,
354                    })
355                }),
356            ),
357            (
358                "requires MQTT 5.0",
359                Box::new(|o| {
360                    o.version = MqttVersion::V311;
361                    o.auth_method = Some("SCRAM".into());
362                }),
363            ),
364            (
365                "auth_data requires auth_method",
366                Box::new(|o| o.auth_data = Some(vec![1])),
367            ),
368        ];
369        for (needle, mutate) in cases {
370            let mut o = opts();
371            mutate(&mut o);
372            let err = o.validate().expect_err(needle).to_string();
373            assert!(err.contains(needle), "{needle:?} not in {err:?}");
374        }
375    }
376
377    #[test]
378    fn password_without_username_is_fine_in_mqtt5() {
379        let mut o = opts();
380        o.password = Some(b"pw".to_vec());
381        assert!(o.validate().is_ok());
382    }
383
384    #[test]
385    fn debug_redacts_the_auth_data_too() {
386        let mut o = opts();
387        o.auth_method = Some("SCRAM".into());
388        o.auth_data = Some(b"client-first-SECRET".to_vec());
389        let shown = format!("{o:?}");
390        assert!(
391            shown.contains("SCRAM") && !shown.contains("SECRET"),
392            "{shown}"
393        );
394    }
395
396    #[test]
397    fn tls_options_default_matches_new() {
398        let d = TlsOptions::default();
399        assert!(!d.insecure_skip_certificate_verification);
400        assert!(d.ca_cert_pem.is_none());
401    }
402}
mqtt_client/client/types.rs

mqtt_client/client/
types.rs
