mqrstt 0.4.2

Pure rust MQTTv5 client implementation Smol and Tokio
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
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116

use futures::Future;

use crate::packets::Packet;
/// Handlers are used to deal with packets before they are acknowledged to the broker.
/// This guarantees that the end user has handlded the packet. Additionally, handlers only deal with incoming packets.
///
/// This handler can be used to handle message sequentialy.
///
/// To send messages look at [`crate::MqttClient`]
pub trait AsyncEventHandler {
    fn handle(&mut self, incoming_packet: Packet) -> impl Future<Output = ()> + Send + Sync;
}

/// This is a simple no operation handler.
impl AsyncEventHandler for () {
    async fn handle(&mut self, _: Packet) {}
}

pub trait EventHandler {
    fn handle(&mut self, incoming_packet: Packet);
}

impl EventHandler for () {
    fn handle(&mut self, _: Packet) {}
}

pub mod example_handlers {
    use std::{ops::AddAssign, sync::atomic::AtomicU16};

    use crate::{
        AsyncEventHandler, EventHandler, MqttClient,
        packets::{self, Packet},
    };

    /// Most basic no op handler
    /// This handler performs no operations on incoming messages.
    pub struct NOP {}

    impl AsyncEventHandler for NOP {
        async fn handle(&mut self, _: Packet) {}
    }

    impl EventHandler for NOP {
        fn handle(&mut self, _: Packet) {}
    }

    #[derive(Debug)]
    pub struct PingResp {
        pub client: MqttClient,
        pub ping_resp_received: u32,
    }

    impl PingResp {
        pub fn new(client: MqttClient) -> Self {
            Self { client, ping_resp_received: 0 }
        }
    }

    impl AsyncEventHandler for PingResp {
        async fn handle(&mut self, event: packets::Packet) {
            use Packet::*;
            if event == PingResp {
                self.ping_resp_received += 1;
            }
            println!("Received packet: {}", event);
        }
    }

    impl EventHandler for PingResp {
        fn handle(&mut self, event: Packet) {
            use Packet::*;
            if event == PingResp {
                self.ping_resp_received += 1;
            }
            println!("Received packet: {}", event);
        }
    }

    pub struct PingPong {
        pub client: MqttClient,
        pub number: AtomicU16,
    }

    impl PingPong {
        pub fn new(client: MqttClient) -> Self {
            Self { client, number: AtomicU16::new(0) }
        }
    }

    impl AsyncEventHandler for PingPong {
        async fn handle(&mut self, event: packets::Packet) {
            match event {
                Packet::Publish(p) => {
                    if let Ok(payload) = String::from_utf8(p.payload.to_vec()) {
                        let max_len = payload.len().min(10);
                        let a = &payload[0..max_len];
                        if payload.to_lowercase().contains("ping") {
                            self.client.publish(p.topic.clone(), p.qos, p.retain, b"pong").await.unwrap();
                            println!("Received publish payload: {}", a);

                            if !p.retain {
                                self.number.get_mut().add_assign(1);
                            }

                            println!("DBG: \n {}", &Packet::Publish(p));
                        }
                    }
                }
                Packet::ConnAck(_) => {
                    println!("Connected!")
                }
                _ => (),
            }
        }
    }
}