gmqtt-client 0.3.1

mod builder;
mod config;
mod message;
mod mqtt;

use std::sync::Arc;

use anyhow::Result;
use fluent_uri::Uri;
use parking_lot::RwLock;

use self::mqtt::PublishStore;

pub use mqttbytes::QoS;

pub use self::{
    builder::{MqttClientBuilder, MqttClientBuilderError},
    config::{OnConnectedCallback, OnMessageCallback},
    message::{Message, MessageBuilder},
    mqtt::MqttWorker,
};
use crate::client::config::{MqttClientConfig, OnMessageOwnedCallback};

#[derive(Clone)]
pub struct MqttClient {
    config: Arc<RwLock<MqttClientConfig>>,
    publish_store: Arc<PublishStore>,
}

impl MqttClient {
    fn new(
        mqtt_broker: Uri<String>,
        config: MqttClientConfig,
        on_message_owned: Option<Box<OnMessageOwnedCallback>>,
    ) -> (Self, MqttWorker) {
        let config = Arc::new(RwLock::new(config));
        let (worker, publish_store) =
            MqttWorker::new(mqtt_broker, config.clone(), on_message_owned);

        let client = Self {
            config,
            publish_store,
        };

        (client, worker)
    }

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

    pub fn set_on_connected_callback<F>(&self, callback: F)
    where
        F: Fn() + Send + Sync + 'static,
    {
        self.config.write().on_connected_callback = Some(Box::new(callback));
    }

    /// Publish MQTT message
    pub fn publish_raw(&self, message: Message) -> Result<()> {
        self.publish_store.insert_to_send(message);

        Ok(())
    }

    /// How many messages are waiting to be send
    pub fn tx_pending(&self) -> usize {
        self.publish_store.tx_pending()
    }

    /// Publish message as json (feature `json` has to be enabled)
    #[cfg(feature = "json")]
    pub fn publish_json<S, T>(
        &self,
        topic: S,
        payload: &T,
        retained: bool,
        qos: QoS,
        expiry_interval: Option<std::time::Duration>,
    ) -> Result<()>
    where
        S: Into<String>,
        T: ?Sized + serde::Serialize,
    {
        let json = serde_json::to_string(payload)?;

        let message = MessageBuilder::new()
            .retained(retained)
            .qos(qos)
            .topic(topic)
            .message_expiry_interval(expiry_interval)
            .utf8(true)
            .payload(json)
            .finalize();

        self.publish_raw(message)
    }
}