mqttier 0.9.3

A Rust MQTT client library providing an abstracted interface around 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
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

use bytes::Bytes;
use mqttier::{Connection, MqttierClient, MqttierOptionsBuilder};
use std::collections::HashMap;
use std::error::Error;
use stinger_mqtt_trait::message::{MqttMessageBuilder, QoS};
use stinger_mqtt_trait::Mqtt5PubSub;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Initialize tracing to stdout
    tracing_subscriber::fmt()
        .with_writer(std::io::stdout)
        .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
        .init();

    println!("=== MQTT Packet Size Test ===");
    println!("Testing maximum reliable packet size by incrementing payload by 1000 bytes\n");

    // Create a new MQTT client using `MqttierOptionsBuilder`
    let options = MqttierOptionsBuilder::default()
        .connection(Connection::TcpLocalhost(1883))
        .client_id("packet_size_test")
        .ack_timeout_ms(5000u64)
        .keepalive_secs(60u16)
        .session_expiry_interval_secs(1200u16)
        .publish_queue_size(128u16)
        .max_incoming_packet_size(2000u32)
        .max_inflight_messages(100u16)
        .build()
        .expect("Failed to build MqttierOptions");
    let mut client = MqttierClient::new(options)?;

    // Start the MQTT client loop
    println!("Starting MQTT client...");
    client.start().await?;
    println!("Client started successfully\n");

    // Test incrementing packet sizes
    let mut size = 1000;
    let increment = 1000;
    let mut last_successful_size = 0;

    loop {
        // Create payload of the current size
        let payload_data: Vec<u8> = (0..size).map(|i| (i % 256) as u8).collect();
        let payload = Bytes::from(payload_data);

        println!("Testing packet size: {} bytes", size);

        let message = MqttMessageBuilder::default()
            .topic("test/packet_size")
            .payload(payload)
            .qos(QoS::AtLeastOnce)
            .retain(false)
            .user_properties(HashMap::new())
            .build()
            .unwrap();

        // Attempt to publish
        let start = std::time::Instant::now();
        match client.publish(message).await {
            Ok(result) => {
                let elapsed = start.elapsed();
                println!(
                    "✓ Success! Size: {} bytes - Acknowledged in {:?}",
                    size, elapsed
                );
                println!("  Result: {:?}\n", result);
                last_successful_size = size;
            }
            Err(e) => {
                let elapsed = start.elapsed();
                println!("\n========================================");
                println!("✗ ERROR ENCOUNTERED!");
                println!("========================================");
                println!("Packet size: {} bytes", size);
                println!("Time elapsed: {:?}", elapsed);
                println!("Last successful size: {} bytes", last_successful_size);
                println!("\nError details:");
                println!("  Error: {:?}", e);
                println!("  Display: {}", e);
                println!("  Debug representation: {:#?}", e);

                // Try to extract source errors
                let mut source = e.source();
                let mut level = 1;
                while let Some(err) = source {
                    println!("\n  Source error level {}: {}", level, err);
                    println!("  Source error debug: {:?}", err);
                    source = err.source();
                    level += 1;
                }

                println!("\n========================================");
                println!("SUMMARY");
                println!("========================================");
                println!(
                    "Maximum successful packet size: {} bytes",
                    last_successful_size
                );
                println!("Failed at packet size: {} bytes", size);
                println!("========================================\n");

                break;
            }
        }

        // Increment size for next iteration
        size += increment;

        // Optional: add a small delay between publishes to avoid overwhelming the broker
        tokio::time::sleep(tokio::time::Duration::from_millis(1)).await;
    }

    println!("Test completed.");
    Ok(())
}