status-monitor/agent/src/mqtt.rs

use common::{metrics::Metrics, StatusMessage, MQTT_TOPIC};
use rumqttc::{AsyncClient, MqttOptions, QoS};
use tokio::task::JoinHandle;
use std::time::Duration;

pub struct MqttHandle {
    pub agent: String,
    pub client: AsyncClient,
    pub eventloop_handle: JoinHandle<()>,
}

impl MqttHandle {
    pub async fn create(agent: String, host: String, port: u16) -> Self {
        let mut mqttoptions = MqttOptions::new(&agent, &host, port);
        mqttoptions.set_keep_alive(Duration::from_secs(5));

        let (client, mut eventloop) = AsyncClient::new(mqttoptions, 10);

        let eventloop_handle = tokio::spawn(async move {
            loop {
                if let Err(e) = eventloop.poll().await {
                    eprintln!("event loop error: {:?}", e);
                    break;
                }
            }
        });

        Self {
            agent,
            client,
            eventloop_handle
        }
    }

    pub async fn send_metrics(&self, metrics: Metrics) -> anyhow::Result<()> {
        let message = StatusMessage::new(&self.agent, metrics).to_string()?;

        self.client
            .publish(MQTT_TOPIC, QoS::AtLeastOnce, false, message.as_bytes())
            .await
            .expect("Failed to publish metrics");

        Ok(())
    }
}