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

use rumqttc::{AsyncClient, EventLoop, MqttOptions, QoS};
use serde::Deserialize;

#[derive(Clone, Deserialize)]
pub enum QoSType {
    AtMost,
    AtLeast,
    Exactly,
}

#[derive(Deserialize)]
pub struct Credentials {
    username: String,
    password: String,
}

#[derive(Clone, Deserialize)]
pub struct LastWill {
    topic: String,
    message: String,
    qos: QoSType,
    retain: bool,
}

#[derive(Deserialize)]
pub struct ClientOptions {
    id: String,
    host: String,
    port: u16,
    cap: usize,
    // number of seconds after which client should ping the broker if there is no other data exchange
    keep_alive: u16,
    credentials: Option<Credentials>,
    last_will: Option<LastWill>,
}

pub(crate) fn new_client(options: &ClientOptions) -> (AsyncClient, EventLoop) {
    let mut mqttoptions = MqttOptions::new(&options.id, &options.host, options.port);
    mqttoptions.set_keep_alive(options.keep_alive);
    if let Some(ref credentials) = options.credentials {
        let username = &credentials.username;
        let password = &credentials.password;
        mqttoptions.set_credentials(username, password);
    }
    if let Some(ref last_will) = options.last_will {
        let last_will = to_last_will(last_will.to_owned());
        mqttoptions.set_last_will(last_will);
    }
    AsyncClient::new(mqttoptions, options.cap)
}

pub(crate) fn to_qos(qos: QoSType) -> QoS {
    match qos {
        QoSType::AtLeast => QoS::AtLeastOnce,
        QoSType::AtMost => QoS::AtMostOnce,
        QoSType::Exactly => QoS::ExactlyOnce,
    }
}

pub(crate) fn to_last_will(last_will: LastWill) -> rumqttc::LastWill {
    rumqttc::LastWill {
        topic: last_will.topic,
        message: last_will.message.into(),
        qos: to_qos(last_will.qos),
        retain: last_will.retain,
    }
}