pipemqtt 0.1.1

A pipebase plugin using rumqttc
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

use crate::client::{new_client, to_qos, ClientOptions, QoSType};
use async_trait::async_trait;
use pipebase::{
    common::{ConfigInto, FromConfig, FromPath},
    export::Export,
};
use rumqttc::{AsyncClient, QoS};
use serde::Deserialize;

#[derive(Deserialize)]
pub struct MqttPublisherConfig {
    base: ClientOptions,
    topic: String,
    qos: QoSType,
    retain: bool,
}

impl FromPath for MqttPublisherConfig {}

impl ConfigInto<MqttPublisher> for MqttPublisherConfig {}

pub struct MqttPublisher {
    client: AsyncClient,
    topic: String,
    qos: QoS,
    retain: bool,
}

#[async_trait]
impl FromConfig<MqttPublisherConfig> for MqttPublisher {
    async fn from_config(config: MqttPublisherConfig) -> anyhow::Result<Self> {
        let qos = to_qos(config.qos);
        let topic = config.topic;
        let retain = config.retain;
        let (client, mut event) = new_client(&config.base);
        // keep channel open
        tokio::spawn(async move {
            loop {
                let notification = event.poll().await.unwrap();
                log::info!("MqttPublisher received = {:?}", notification);
            }
        });
        Ok(MqttPublisher {
            client,
            topic,
            qos,
            retain,
        })
    }
}

#[async_trait]
impl Export<Vec<u8>, MqttPublisherConfig> for MqttPublisher {
    async fn export(&mut self, payload: Vec<u8>) -> anyhow::Result<()> {
        self.client
            .publish(&self.topic, self.qos, self.retain, payload)
            .await?;
        Ok(())
    }
}