gmqtt-client 0.3.1

use fluent_uri::Uri;
use mqttbytes::{v5::SubscribeFilter, QoS};
use rand::{distr::Alphanumeric, rng, Rng};

use super::{
    mqtt::MqttWorker, Message, MqttClient, MqttClientConfig, OnConnectedCallback, OnMessageCallback,
};
use crate::client::config::OnMessageOwnedCallback;
use std::time::Instant;

pub struct MqttClientBuilder {
    mqtt_broker: String,
    client_id: Option<String>,
    clean_session: bool,

    subscribe_filters: Vec<SubscribeFilter>,
    on_message_callback: Option<Box<OnMessageCallback>>,
    on_message_owned_callback: Option<Box<OnMessageOwnedCallback>>,
    on_connected_callback: Option<Box<OnConnectedCallback>>,
}

#[derive(Debug, thiserror::Error)]
pub enum MqttClientBuilderError {
    #[error("Invalid MQTT broker URI: {0}")]
    InvalidUri(#[source] fluent_uri::ParseError),
}

impl MqttClientBuilder {
    pub fn new(uri: &str) -> Self {
        Self {
            mqtt_broker: uri.to_owned(),
            client_id: None,
            clean_session: false,
            subscribe_filters: vec![],
            on_message_callback: None,
            on_message_owned_callback: None,
            on_connected_callback: None,
        }
    }

    pub fn client_id<S: Into<String>>(mut self, client_id: S) -> Self {
        self.client_id = Some(client_id.into());
        self
    }

    /// When the clean session flag is set to true, the client does not want a persistent session.
    /// If the client disconnects for any reason, all information and messages that are queued from a previous persistent session are lost.
    ///
    /// When the clean session flag is set to false, the broker creates a persistent session for the client.
    /// All information and messages are preserved until the next time that the client requests a clean session.
    /// If the clean session flag is set to false and the broker already has a session available for the client,
    /// it uses the existing session and delivers previously queued messages to the client.
    pub fn clean_session(mut self, clean_session: bool) -> Self {
        self.clean_session = clean_session;
        self
    }

    pub fn subscribe<S: Into<String>>(mut self, filter: S, qos: QoS) -> Self {
        if qos == QoS::ExactlyOnce {
            // TODO: Add QoS 2 support for incoming messages
            panic!("Quantity of Service 2 (Exactly Once) is not supported");
        }

        let sf = SubscribeFilter::new(filter.into(), qos);
        self.subscribe_filters.push(sf);
        self
    }

    pub fn on_message_callback<F>(mut self, callback: F) -> Self
    where
        F: FnMut(&Message) + Send + Sync + 'static,
    {
        self.on_message_callback = Some(Box::new(callback));
        self
    }

    pub fn on_connected_callback<F>(mut self, callback: F) -> Self
    where
        F: Fn() + Send + Sync + 'static,
    {
        self.on_connected_callback = Some(Box::new(callback));
        self
    }

    pub fn on_message_owned_callback<F>(mut self, callback: F) -> Self
    where
        F: FnMut(Message, Instant) + Send + 'static,
    {
        self.on_message_owned_callback = Some(Box::new(callback));
        self
    }

    pub fn build(self) -> Result<(MqttClient, MqttWorker), MqttClientBuilderError> {
        let uri = Uri::parse(self.mqtt_broker)
            .map_err(|(err, _u)| MqttClientBuilderError::InvalidUri(err))?;

        let client_id = self.client_id.unwrap_or_else(|| {
            let username = uri
                .authority()
                .and_then(|a| a.userinfo())
                .map(|u| u.as_str().split_once(':').map_or(u.as_str(), |(u, _)| u))
                .unwrap_or("");
            if !username.is_empty() {
                format!("{}-{}", username, random_string(8))
            } else {
                random_string(16)
            }
        });

        let config = MqttClientConfig {
            client_id,
            subscribe_filters: self.subscribe_filters,
            on_message_callback: self.on_message_callback,
            on_connected_callback: self.on_connected_callback,
            clean_session: self.clean_session,
        };

        Ok(MqttClient::new(uri, config, self.on_message_owned_callback))
    }
}

fn random_string(length: usize) -> String {
    rng()
        .sample_iter(&Alphanumeric)
        .take(length)
        .map(char::from)
        .collect()
}