Rust语言从入门到精通系列 - 物联网消息传输协议MQTT(进阶)

1
use rumqttc::{Client, MqttOptions, QoS};
2
use std::thread;
3

4
fn main() {
5
    let mqtt_options = MqttOptions::new("test-thread", "localhost", 1883);
6

7
    let (mut client, _) = Client::new(mqtt_options, 10);
8

9
    let handle = thread::spawn(move || {
10
        client
11
            .publish("test/topic", QoS::AtLeastOnce, false, "hello world".to_owned())
12
            .unwrap();
13
    });
14

15
    handle.join().unwrap();
16
}

1
use std::thread;
2
use rumqttc::{Client, MqttOptions, QoS, Event, Incoming, Publish};
3

4
fn main() {
5
    let mqtt_options = MqttOptions::new("test-7", "localhost", 1883);
6
    let (mut client, mut connection) = Client::new(mqtt_options, 10);
7
    client.subscribe("test/topic", QoS::AtLeastOnce).unwrap();
8
    let mut storage = Storage::new("mqtt.db").unwrap();
9
    for event in connection.iter() {
10
        match event.unwrap() {
11
            Event::Incoming(Incoming::Publish(publish)) => {
12
                storage.insert_message(&publish).unwrap();
13
                println!("Received message: {} from topic: {}", publish.payload, publish.topic);
14
            },
15
            _ => {},
16
        }
17
    }
18
}
19

20
struct Storage {
21
    conn: rusqlite::Connection,
22
}
23

24
impl Storage {
25
    fn new(path: &str) -> rusqlite::Result<Self> {
26
        let conn = rusqlite::Connection::open(path)?;
27
        conn.execute("CREATE TABLE IF NOT EXISTS messages (id INTEGER PRIMARY KEY, topic TEXT, payload TEXT, qos INTEGER)", [])?;
28
        Ok(Self { conn })
29
    }
30

31
    fn insert_message(&mut self, publish: &Publish) -> rusqlite::Result<()> {
32
        let mut stmt = self.conn.prepare("INSERT INTO messages (topic, payload, qos) VALUES (?, ?, ?)")?;
33
        let qos_val = publish.qos as i32;
34
        stmt.execute(rusqlite::params![publish.topic, publish.payload, qos_val])?;
35
        Ok(())
36
    }
37
}