rainmaker-components 0.1.0

Component abstractions for rainmaker
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
80
81
82
83
84
85
86
87
88
89
90
91
92
93

#![cfg(target_os = "espidf")]

use esp_idf_svc::sys::EspError;

use crate::{error::Error, mqtt::base::*};

impl From<&QoSLevel> for esp_idf_svc::mqtt::client::QoS {
    fn from(value: &QoSLevel) -> Self {
        match value {
            QoSLevel::AtMostOnce => esp_idf_svc::mqtt::client::QoS::AtMostOnce,
            QoSLevel::AtLeastOnce => esp_idf_svc::mqtt::client::QoS::AtLeastOnce,
            QoSLevel::ExactlyOnce => esp_idf_svc::mqtt::client::QoS::ExactlyOnce,
        }
    }
}

impl From<esp_idf_svc::mqtt::client::EventPayload<'_, EspError>> for MqttEvent {
    fn from(value: esp_idf_svc::mqtt::client::EventPayload<'_, EspError>) -> Self {
        match value {
            esp_idf_svc::mqtt::client::EventPayload::Connected(_) => MqttEvent::Connected,
            esp_idf_svc::mqtt::client::EventPayload::Received {
                id: _,
                topic,
                data,
                details: _,
            } => {
                MqttEvent::Received(ReceivedMessage {
                    topic: topic.unwrap().to_string(), // not able to think of a case where this will be null
                    payload: data.to_vec(),
                })
            }
            esp_idf_svc::mqtt::client::EventPayload::Disconnected => MqttEvent::Disconnected,
            esp_idf_svc::mqtt::client::EventPayload::BeforeConnect => MqttEvent::BeforeConnect,
            esp_idf_svc::mqtt::client::EventPayload::Published(_) => MqttEvent::Published,
            esp_idf_svc::mqtt::client::EventPayload::Subscribed(_) => MqttEvent::Subscribed,
            _ => Self::Other,
        }
    }
}

impl<'a> MqttClient<esp_idf_svc::mqtt::client::EspMqttClient<'a>> {
    pub fn new(
        config: &MqttConfiguration,
        tls_certs: &'static TLSconfiguration,
        callback: Box<dyn Fn(MqttEvent) + Send + Sync>,
    ) -> Result<Self, Error> {
        let client_cert = std::ffi::CStr::from_bytes_with_nul(tls_certs.client_cert).unwrap();
        let private_key = std::ffi::CStr::from_bytes_with_nul(tls_certs.private_key).unwrap();
        let server_cert = std::ffi::CStr::from_bytes_with_nul(tls_certs.server_cert).unwrap();

        let options = esp_idf_svc::mqtt::client::MqttClientConfiguration {
            client_id: Some(config.clientid),
            server_certificate: Some(esp_idf_svc::tls::X509::pem(server_cert)),
            client_certificate: Some(esp_idf_svc::tls::X509::pem(client_cert)),
            private_key: Some(esp_idf_svc::tls::X509::pem(private_key)),
            ..Default::default()
        };

        let mut conn_addr = if config.port == 8883 {
            "mqtts://"
        } else {
            "mqtt://"
        }
        .to_string();

        conn_addr += config.host;

        log::info!("connection string: {}", conn_addr);

        let client = esp_idf_svc::mqtt::client::EspMqttClient::new_cb(
            conn_addr.as_str(),
            &options,
            move |req| callback(req.payload().into()),
        )
        .unwrap();

        Ok(Self { client })
    }

    pub fn publish(&mut self, topic: &str, qos: &QoSLevel, payload: Vec<u8>) -> u32 {
        let msg_id = self
            .client
            .publish(topic, qos.into(), false, payload.as_ref())
            .expect("unable to publish");

        msg_id
    }

    pub fn subscribe(&mut self, topic: &str, qos: &QoSLevel) -> Result<(), Error> {
        self.client.subscribe(topic, qos.into())?;
        Ok(())
    }
}