mqtt5 0.32.2

Complete MQTT v5.0 platform with high-performance async client and full-featured broker supporting TCP, TLS, WebSocket, authentication, bridging, and resource monitoring
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
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134

//! Basic connectivity tests using Turmoil
//!
//! These tests verify that the broker can accept connections and handle
//! basic MQTT operations in a deterministic environment.

#[cfg(feature = "turmoil-testing")]
use mqtt5::testing::{TurmoilBroker, TurmoilBrokerConfig, TurmoilClient, TurmoilClientConfig};
#[cfg(feature = "turmoil-testing")]
use mqtt5::time::Duration;

#[cfg(feature = "turmoil-testing")]
#[test]
fn test_basic_broker_startup() {
    let mut sim = turmoil::Builder::new()
        .simulation_duration(Duration::from_secs(10))
        .build();

    // Add broker host
    sim.host("broker", || async {
        let config = TurmoilBrokerConfig::new("0.0.0.0:1883");
        let broker = TurmoilBroker::new(config)
            .await
            .map_err(|e| Box::new(e) as Box<dyn std::error::Error>)?;

        broker
            .run()
            .await
            .map_err(|e| Box::new(e) as Box<dyn std::error::Error>)
    });

    // Run the simulation
    sim.run().unwrap();
}

#[cfg(feature = "turmoil-testing")]
#[test]
fn test_client_connection() {
    let mut sim = turmoil::Builder::new()
        .simulation_duration(Duration::from_secs(15))
        .build();

    // Add broker host
    sim.host("broker", || async {
        let config = TurmoilBrokerConfig::new("0.0.0.0:1883");
        let broker = TurmoilBroker::new(config)
            .await
            .map_err(|e| Box::new(e) as Box<dyn std::error::Error>)?;

        broker
            .run()
            .await
            .map_err(|e| Box::new(e) as Box<dyn std::error::Error>)
    });

    // Add client host
    sim.host("client", || async {
        // Wait a bit for broker to start
        tokio::time::sleep(Duration::from_millis(100)).await;

        let config = TurmoilClientConfig::new("test-client");
        let client = TurmoilClient::new(&config);

        client
            .connect("broker:1883")
            .await
            .map_err(|e| Box::new(e) as Box<dyn std::error::Error>)?;

        // Verify connection
        assert!(client.is_connected().await);

        // Disconnect cleanly
        client
            .disconnect()
            .await
            .map_err(|e| Box::new(e) as Box<dyn std::error::Error>)?;

        Ok::<(), Box<dyn std::error::Error>>(())
    });

    // Run the simulation
    sim.run().unwrap();
}

#[cfg(feature = "turmoil-testing")]
#[test]
fn test_multiple_clients_connection() {
    let mut sim = turmoil::Builder::new()
        .simulation_duration(Duration::from_secs(20))
        .build();

    // Add broker host
    sim.host("broker", || async {
        let config = TurmoilBrokerConfig::new("0.0.0.0:1883");
        let broker = TurmoilBroker::new(config)
            .await
            .map_err(|e| Box::new(e) as Box<dyn std::error::Error>)?;

        broker
            .run()
            .await
            .map_err(|e| Box::new(e) as Box<dyn std::error::Error>)
    });

    // Add multiple client hosts
    for i in 1..=3 {
        let client_name = format!("client{i}");
        sim.host(client_name, move || async move {
            tokio::time::sleep(Duration::from_millis(200)).await;

            let config = TurmoilClientConfig::new(&format!("test-client-{i}"));
            let client = TurmoilClient::new(&config);

            client
                .connect("broker:1883")
                .await
                .map_err(|e| Box::new(e) as Box<dyn std::error::Error>)?;

            assert!(client.is_connected().await);

            // Stay connected for a while
            tokio::time::sleep(Duration::from_secs(5)).await;

            client
                .disconnect()
                .await
                .map_err(|e| Box::new(e) as Box<dyn std::error::Error>)?;

            Ok::<(), Box<dyn std::error::Error>>(())
        });
    }

    // Run the simulation
    sim.run().unwrap();
}