stem-mqtt-client 0.8.0

//! Public data types for the client API: connection configuration,
//! results, delivered messages, and the foreign-callback listener trait.
//! No I/O or connection-state logic lives here — see
//! [`crate::client::inner`] and [`crate::client::io`].

use std::time::Duration;

use crate::protocol::packet::MAX_PACKET_SIZE;
use crate::protocol::{MqttVersion, QoS};
use crate::support::secs_or;

/// Default time to wait for a broker acknowledgement (PUBACK/PUBREC/
/// PUBCOMP/SUBACK/UNSUBACK) before returning [`crate::error::MqttError::Timeout`].
pub(super) const DEFAULT_OPERATION_TIMEOUT_SECS: u32 = 15;

/// Last-Will-and-Testament configuration for a [`ConnectOptions`].
#[derive(Debug, Clone, PartialEq, Eq, uniffi::Record)]
pub struct WillOptions {
    pub topic: String,
    pub payload: Vec<u8>,
    pub qos: QoS,
    pub retain: bool,
}

/// TLS configuration for [`ConnectOptions::tls`]. All certificate/key
/// material is PEM-encoded bytes (not file paths) so this works
/// identically across every language binding without assuming a
/// filesystem layout.
#[derive(Clone, PartialEq, Eq, uniffi::Record)]
pub struct TlsOptions {
    /// Additional PEM-encoded root CA certificate(s) to trust, on top of
    /// the bundled Mozilla root store (`webpki-roots`). Needed to connect
    /// to a broker with a self-signed or private-CA certificate.
    pub ca_cert_pem: Option<Vec<u8>>,
    /// PEM-encoded client certificate, for mutual TLS (mTLS). Requires
    /// [`Self::client_key_pem`] too.
    pub client_cert_pem: Option<Vec<u8>>,
    /// PEM-encoded client private key, for mutual TLS. Requires
    /// [`Self::client_cert_pem`] too.
    pub client_key_pem: Option<Vec<u8>>,
    /// Skip server certificate verification entirely. **Dangerous** — only
    /// for local development against a broker with a self-signed
    /// certificate you can't otherwise easily trust (e.g. one generated
    /// on the fly for a demo). Never use this in production: it makes the
    /// connection trivially interceptable.
    pub insecure_skip_certificate_verification: bool,
}

// Never print the client private key (or a CA bundle's presence-only detail
// beyond "is set") when a `TlsOptions` is `Debug`-formatted.
crate::redacted_debug!(TlsOptions {
    ca_cert_pem, client_cert_pem, insecure_skip_certificate_verification
} secret { client_key_pem });

impl TlsOptions {
    /// TLS with the bundled Mozilla root store and no client certificate —
    /// the common case of connecting to a broker with a certificate from a
    /// well-known public CA.
    pub fn new() -> Self {
        TlsOptions {
            ca_cert_pem: None,
            client_cert_pem: None,
            client_key_pem: None,
            insecure_skip_certificate_verification: false,
        }
    }
}

impl Default for TlsOptions {
    fn default() -> Self {
        Self::new()
    }
}

/// Everything needed to establish an MQTT connection.
#[derive(Clone, PartialEq, Eq, uniffi::Record)]
pub struct ConnectOptions {
    pub host: String,
    pub port: u16,
    pub client_id: String,
    /// Which protocol version to speak. The broker never up/downgrades a
    /// connection once CONNECT is sent.
    pub version: MqttVersion,
    /// MQTT 3.1.1 "Clean Session" / MQTT 5.0 "Clean Start".
    pub clean_start: bool,
    /// Keep-alive interval in seconds. `0` disables keep-alive pings.
    pub keep_alive_secs: u16,
    pub username: Option<String>,
    pub password: Option<Vec<u8>>,
    pub will: Option<WillOptions>,
    /// Seconds to wait for the TCP connection + CONNACK before failing.
    pub connect_timeout_secs: u32,
    /// Seconds to wait for a QoS 1/2 handshake step *per attempt* before
    /// retrying (with DUP=1) — `MqttClient::publish` retries an unacked
    /// QoS 1/2 PUBLISH (or QoS 2's PUBREL) up to 3 times before giving up,
    /// so the worst-case total wait for a single `publish()` call is up to
    /// 4x this value. `0` selects the built-in default (15s).
    pub operation_timeout_secs: u32,
    /// If true, [`crate::client::MqttClient`] automatically reconnects
    /// (with exponential backoff, see [`Self::reconnect_backoff_secs`] /
    /// [`Self::reconnect_max_backoff_secs`]) after a connection loss that
    /// wasn't caused by calling `disconnect()`, replaying every topic this
    /// client is currently subscribed to (per its own bookkeeping) once
    /// reconnected. Off by default — the caller must opt in.
    pub auto_reconnect: bool,
    /// Initial delay before the first reconnect attempt, doubling after
    /// each failed attempt up to [`Self::reconnect_max_backoff_secs`].
    /// `0` selects the built-in default (1s). Ignored unless
    /// [`Self::auto_reconnect`] is set.
    pub reconnect_backoff_secs: u32,
    /// Cap on the exponential reconnect backoff delay. `0` selects the
    /// built-in default (30s). Ignored unless [`Self::auto_reconnect`] is
    /// set.
    pub reconnect_max_backoff_secs: u32,
    /// If set, connect over TLS (`mqtts`) instead of plain TCP. `None`
    /// (the default) is plain TCP, unchanged from before TLS support
    /// existed.
    pub tls: Option<TlsOptions>,
    /// Largest packet (fixed header + body, in bytes) this client will
    /// accept from the broker; a larger one is treated as a protocol error
    /// and drops the connection instead of being buffered. `0` selects the
    /// built-in default (the MQTT protocol maximum, ~256 MiB).
    #[uniffi(default = 0)]
    pub max_packet_size: u32,
    /// MQTT 5.0 enhanced authentication (SCRAM, Kerberos, OAuth, …): the
    /// `Authentication Method` to announce in CONNECT. The broker may then
    /// send challenges, which are answered by the
    /// [`MqttAuthHandler`] registered with `MqttClient::set_auth_handler`.
    /// Requires [`MqttVersion::V5`].
    #[uniffi(default = None)]
    pub auth_method: Option<String>,
    /// Initial `Authentication Data` sent with CONNECT (e.g. a SCRAM
    /// client-first message). Requires [`Self::auth_method`].
    #[uniffi(default = None)]
    pub auth_data: Option<Vec<u8>>,
}

// Hand-rolled `Debug` so the password never reaches a log line.
crate::redacted_debug!(ConnectOptions {
    host, port, client_id, version, clean_start, keep_alive_secs, username, will,
    connect_timeout_secs, operation_timeout_secs, auto_reconnect,
    reconnect_backoff_secs, reconnect_max_backoff_secs, tls, max_packet_size, auth_method
} secret { password, auth_data });

impl ConnectOptions {
    pub fn new(
        host: impl Into<String>,
        port: u16,
        client_id: impl Into<String>,
        version: MqttVersion,
    ) -> Self {
        ConnectOptions {
            host: host.into(),
            port,
            client_id: client_id.into(),
            version,
            clean_start: true,
            keep_alive_secs: 30,
            username: None,
            password: None,
            will: None,
            connect_timeout_secs: 10,
            operation_timeout_secs: 0,
            auto_reconnect: false,
            reconnect_backoff_secs: 0,
            reconnect_max_backoff_secs: 0,
            tls: None,
            max_packet_size: 0,
            auth_method: None,
            auth_data: None,
        }
    }

    pub(super) fn operation_timeout(&self) -> Duration {
        secs_or(self.operation_timeout_secs, DEFAULT_OPERATION_TIMEOUT_SECS)
    }

    pub(super) fn connect_timeout(&self) -> Duration {
        // A zero connect timeout would fail instantly; treat it as "1s".
        secs_or(self.connect_timeout_secs, 1)
    }

    pub(super) fn initial_reconnect_backoff(&self) -> Duration {
        secs_or(self.reconnect_backoff_secs, 1)
    }

    pub(super) fn max_reconnect_backoff(&self) -> Duration {
        secs_or(self.reconnect_max_backoff_secs, 30)
    }

    pub(super) fn max_packet_size(&self) -> usize {
        match self.max_packet_size {
            0 => MAX_PACKET_SIZE,
            n => n as usize,
        }
    }

    /// Reject configurations that can never work, before touching the
    /// network, with a message naming the offending field.
    pub(super) fn validate(&self) -> crate::error::MqttResult<()> {
        use crate::error::MqttError::Protocol;
        use crate::protocol::topic::is_valid_topic_name;
        if self.host.is_empty() {
            return Err(Protocol("ConnectOptions.host must not be empty".into()));
        }
        if self.client_id.len() > usize::from(u16::MAX) {
            return Err(Protocol("ConnectOptions.client_id is too long".into()));
        }
        if self.password.is_some() && self.username.is_none() && !self.version.is_v5() {
            // MQTT-3.1.2-22: 3.1.1 forbids a password without a username.
            return Err(Protocol(
                "ConnectOptions.password requires a username in MQTT 3.1.1".into(),
            ));
        }
        if self.auth_method.is_some() && !self.version.is_v5() {
            return Err(Protocol(
                "enhanced authentication (auth_method) requires MQTT 5.0".into(),
            ));
        }
        if self.auth_data.is_some() && self.auth_method.is_none() {
            return Err(Protocol("auth_data requires auth_method".into()));
        }
        if let Some(will) = &self.will
            && !is_valid_topic_name(&will.topic)
        {
            return Err(Protocol(format!(
                "invalid will topic {:?}: must be non-empty and wildcard-free",
                will.topic
            )));
        }
        Ok(())
    }
}

/// Outcome of a successful [`crate::client::MqttClient::connect`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, uniffi::Record)]
pub struct ConnectResult {
    pub session_present: bool,
    /// `0x00` on success. MQTT 5.0 reason code (MQTT 3.1.1 codes are mapped
    /// onto the equivalent 5.0 code by
    /// [`ConnectReasonCode`](crate::protocol::connect::ConnectReasonCode)).
    pub reason_code: u8,
}

/// A message delivered to a subscriber.
#[derive(Debug, Clone, PartialEq, Eq, uniffi::Record)]
pub struct MqttMessage {
    pub topic: String,
    pub payload: Vec<u8>,
    pub qos: QoS,
    pub retain: bool,
}

impl From<&crate::protocol::publish::PublishPacket> for MqttMessage {
    fn from(p: &crate::protocol::publish::PublishPacket) -> Self {
        MqttMessage {
            topic: p.topic.clone(),
            payload: p.payload.to_vec(),
            qos: p.qos,
            retain: p.retain,
        }
    }
}

/// Result of a subscribe request: the reason/return code the broker
/// granted for the requested filter (`< 0x80` is success; MQTT 3.1.1 codes
/// 0/1/2 map onto the granted QoS).
#[derive(Debug, Clone, Copy, PartialEq, Eq, uniffi::Record)]
pub struct SubscribeResult {
    pub reason_code: u8,
}

/// Answers the broker's authentication challenges during an MQTT 5.0
/// enhanced-authentication handshake (see [`ConnectOptions::auth_method`]).
#[uniffi::export(with_foreign)]
pub trait MqttAuthHandler: Send + Sync {
    /// The broker sent `challenge` for `method`; return the response, or
    /// `None` to abort the connection attempt.
    fn respond(&self, method: String, challenge: Vec<u8>) -> Option<Vec<u8>>;
}

/// Callback interface implemented by foreign code to receive events from an
/// [`crate::client::MqttClient`] without polling.
#[uniffi::export(with_foreign)]
pub trait MqttMessageListener: Send + Sync {
    /// Called for every PUBLISH delivered to a subscription this client holds.
    fn on_message(&self, message: MqttMessage);
    /// Called when the connection to the broker is lost or closed, whether
    /// by the client, the broker, or the network.
    fn on_disconnected(&self, reason: String);
}

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

    fn opts() -> ConnectOptions {
        ConnectOptions::new("localhost", 1883, "cid", MqttVersion::V5)
    }

    #[test]
    fn defaults_are_sane() {
        let o = opts();
        assert!(o.clean_start && !o.auto_reconnect);
        assert_eq!(o.keep_alive_secs, 30);
        assert_eq!(o.operation_timeout(), Duration::from_secs(15));
        assert_eq!(o.initial_reconnect_backoff(), Duration::from_secs(1));
        assert_eq!(o.max_reconnect_backoff(), Duration::from_secs(30));
        assert_eq!(o.max_packet_size(), MAX_PACKET_SIZE);
        assert!(o.validate().is_ok());
    }

    #[test]
    fn explicit_values_override_the_zero_defaults() {
        let mut o = opts();
        o.operation_timeout_secs = 3;
        o.reconnect_backoff_secs = 2;
        o.reconnect_max_backoff_secs = 9;
        o.max_packet_size = 4096;
        assert_eq!(o.operation_timeout(), Duration::from_secs(3));
        assert_eq!(o.initial_reconnect_backoff(), Duration::from_secs(2));
        assert_eq!(o.max_reconnect_backoff(), Duration::from_secs(9));
        assert_eq!(o.max_packet_size(), 4096);
    }

    #[test]
    fn a_zero_connect_timeout_is_one_second_not_instant_failure() {
        let mut o = opts();
        o.connect_timeout_secs = 0;
        assert_eq!(o.connect_timeout(), Duration::from_secs(1));
    }

    #[test]
    fn validation_rejects_each_unusable_configuration() {
        type Mutation = Box<dyn Fn(&mut ConnectOptions)>;
        let cases: Vec<(&str, Mutation)> = vec![
            ("host", Box::new(|o| o.host.clear())),
            ("client_id", Box::new(|o| o.client_id = "x".repeat(70_000))),
            (
                "password requires a username",
                Box::new(|o| {
                    o.version = MqttVersion::V311;
                    o.password = Some(b"pw".to_vec());
                }),
            ),
            (
                "will topic",
                Box::new(|o| {
                    o.will = Some(WillOptions {
                        topic: "a/+".into(),
                        payload: vec![],
                        qos: QoS::AtMostOnce,
                        retain: false,
                    })
                }),
            ),
            (
                "requires MQTT 5.0",
                Box::new(|o| {
                    o.version = MqttVersion::V311;
                    o.auth_method = Some("SCRAM".into());
                }),
            ),
            (
                "auth_data requires auth_method",
                Box::new(|o| o.auth_data = Some(vec![1])),
            ),
        ];
        for (needle, mutate) in cases {
            let mut o = opts();
            mutate(&mut o);
            let err = o.validate().expect_err(needle).to_string();
            assert!(err.contains(needle), "{needle:?} not in {err:?}");
        }
    }

    #[test]
    fn password_without_username_is_fine_in_mqtt5() {
        let mut o = opts();
        o.password = Some(b"pw".to_vec());
        assert!(o.validate().is_ok());
    }

    #[test]
    fn debug_redacts_the_auth_data_too() {
        let mut o = opts();
        o.auth_method = Some("SCRAM".into());
        o.auth_data = Some(b"client-first-SECRET".to_vec());
        let shown = format!("{o:?}");
        assert!(
            shown.contains("SCRAM") && !shown.contains("SECRET"),
            "{shown}"
        );
    }

    #[test]
    fn tls_options_default_matches_new() {
        let d = TlsOptions::default();
        assert!(!d.insecure_skip_certificate_verification);
        assert!(d.ca_cert_pem.is_none());
    }
}