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
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

use crate::client::{new_client, to_qos, ClientOptions, QoSType};
use async_trait::async_trait;
use pipebase::{
    common::{ConfigInto, FromConfig, FromPath},
    listen::Listen,
};
use rumqttc::{Event, EventLoop, Packet, QoS};
use serde::Deserialize;
use tokio::sync::mpsc::Sender;

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

impl FromPath for MqttSubscriberConfig {}

impl ConfigInto<MqttSubscriber> for MqttSubscriberConfig {}

pub struct MqttSubscriber {
    topic: String,
    qos: QoS,
    client_opts: ClientOptions,
    tx: Option<Sender<Vec<u8>>>,
}

#[async_trait]
impl FromConfig<MqttSubscriberConfig> for MqttSubscriber {
    async fn from_config(config: MqttSubscriberConfig) -> anyhow::Result<Self> {
        let client_opts = config.base;
        let topic = config.topic;
        let qos = to_qos(config.qos);
        Ok(MqttSubscriber {
            topic,
            qos,
            client_opts,
            tx: None,
        })
    }
}

#[async_trait]
impl Listen<Vec<u8>, MqttSubscriberConfig> for MqttSubscriber {
    async fn run(&mut self) -> anyhow::Result<()> {
        let (client, event) = new_client(&self.client_opts);
        client.subscribe(&self.topic, self.qos).await?;
        let tx = self
            .tx
            .as_ref()
            .expect("sender not inited for mqtt listener");
        Self::start_loop(event, tx).await
    }

    fn set_sender(&mut self, sender: Sender<Vec<u8>>) {
        self.tx = Some(sender)
    }
}

impl MqttSubscriber {
    async fn start_loop(mut event: EventLoop, tx: &Sender<Vec<u8>>) -> anyhow::Result<()> {
        loop {
            let event = event.poll().await?;
            let packet = match event {
                Event::Incoming(packet) => packet,
                _ => continue,
            };
            let payload = match packet {
                Packet::Publish(publish) => publish.payload,
                _ => {
                    log::info!("incoming packet {:?}", packet);
                    continue;
                }
            };
            tx.send(payload.to_vec()).await?;
        }
    }
}