gmqtt-client 0.3.1

use std::time::Duration;

use bytes::Bytes;
use mqttbytes::{
    v5::{Publish, PublishProperties},
    QoS,
};
use std::borrow::Cow;
use tracing::error;

#[derive(Debug)]
pub struct MessageBuilder {
    qos: QoS,
    retain: bool,
    topic: String,
    properties: PublishProperties,
    payload: Bytes,
}

#[derive(Debug, Clone)]
pub struct Message {
    pub(crate) qos: QoS,
    pub(crate) retain: bool,
    pub(crate) topic: String,
    pub(crate) properties: PublishProperties,
    pub(crate) payload: Bytes,
}

impl MessageBuilder {
    pub fn new() -> Self {
        Self {
            qos: QoS::AtMostOnce,
            retain: false,
            topic: String::new(),
            properties: publish_properties_default(),
            payload: Bytes::new(),
        }
    }

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

        self.qos = qos;
        self
    }

    pub fn payload<B: Into<Bytes>>(mut self, payload: B) -> Self {
        self.payload = payload.into();
        self
    }

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

    pub fn retained(mut self, retain: bool) -> Self {
        self.retain = retain;
        self
    }

    pub fn message_expiry_interval(mut self, expiry_interval: Option<Duration>) -> Self {
        let message_expiry_interval = expiry_interval.map(|d| {
            let expiry_interval_secs = d.as_secs();

            if expiry_interval_secs > u32::MAX as u64 {
                error!("Expiry interval too long, using std::u32::MAX");
                return u32::MAX;
            }

            expiry_interval_secs as u32
        });

        self.properties.message_expiry_interval = message_expiry_interval;
        self
    }

    pub fn response_topic<S: Into<String>>(mut self, response_topic: Option<S>) -> Self {
        self.properties.response_topic = response_topic.map(|t| t.into());
        self
    }

    /// Set Payload Format Indicator to 1 (UTF-8 Encoded Character Data) or None (unspecified bytes)
    pub fn utf8(mut self, is_utf8: bool) -> Self {
        if is_utf8 {
            self.properties.payload_format_indicator = Some(1);
        } else {
            self.properties.payload_format_indicator = None;
        }
        self
    }

    pub fn add_user_property<K: Into<String>, V: Into<String>>(mut self, key: K, value: V) -> Self {
        self.properties
            .user_properties
            .push((key.into(), value.into()));
        self
    }

    pub fn correlation_data<B: Into<Bytes>>(mut self, correlation_data: Option<B>) -> Self {
        self.properties.correlation_data = correlation_data.map(|c| c.into());
        self
    }

    pub fn finalize(self) -> Message {
        Message {
            qos: self.qos,
            retain: self.retain,
            topic: self.topic,
            properties: self.properties,
            payload: self.payload,
        }
    }
}

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

impl Message {
    pub fn qos(&self) -> QoS {
        self.qos
    }

    pub fn retain(&self) -> bool {
        self.retain
    }

    pub fn topic(&self) -> &str {
        &self.topic
    }

    pub fn properties(&self) -> &PublishProperties {
        &self.properties
    }

    pub fn payload(&self) -> &[u8] {
        &self.payload
    }

    pub fn payload_str(&self) -> Cow<'_, str> {
        String::from_utf8_lossy(&self.payload)
    }
}

#[cfg(feature = "json")]
impl<'a> Message {
    pub fn payload_json<D: serde::Deserialize<'a>>(&'a self) -> serde_json::error::Result<D> {
        serde_json::from_slice::<D>(&self.payload)
    }
}

impl From<Publish> for Message {
    fn from(publish: Publish) -> Self {
        let properties = publish
            .properties
            .unwrap_or_else(publish_properties_default);

        Self {
            qos: publish.qos,
            retain: publish.retain,
            topic: publish.topic,
            properties,
            payload: publish.payload,
        }
    }
}

fn publish_properties_default() -> PublishProperties {
    PublishProperties {
        payload_format_indicator: None,
        message_expiry_interval: None,
        topic_alias: None,
        response_topic: None,
        correlation_data: None,
        user_properties: Vec::new(),
        subscription_identifiers: Vec::new(),
        content_type: None,
    }
}