use flowsdk::mqtt_client::client::ConnectionResult;
use flowsdk::mqtt_client::tokio_async_client::{
TokioAsyncClientConfig, TokioAsyncMqttClient, TokioMqttEventHandler,
};
use flowsdk::mqtt_client::{MqttClientError, MqttClientOptions};
use flowsdk::mqtt_serde::mqttv5::publishv5::MqttPublish;
use std::io;
use std::sync::Arc;
#[derive(Clone)]
struct BuilderExampleHandler;
#[async_trait::async_trait]
impl TokioMqttEventHandler for BuilderExampleHandler {
async fn on_connected(&mut self, result: &ConnectionResult) {
println!(
"â
Connected to broker - reason_code: {}, session_present: {}",
result.reason_code, result.session_present
);
}
async fn on_message_received(&mut self, publish: &MqttPublish) {
println!(
"đ¨ Message received on '{}': {}",
publish.topic_name,
String::from_utf8_lossy(&publish.payload)
);
}
async fn on_error(&mut self, error: &MqttClientError) {
eprintln!("â ī¸ Error: {}", error.user_message());
}
}
#[allow(clippy::field_reassign_with_default)]
async fn run_example() -> io::Result<()> {
println!("đ MQTT Client Options Builder Pattern Examples\n");
println!("{}", "=".repeat(60));
println!("\nđ Example 1: Minimal Configuration");
println!("{}", "-".repeat(60));
let options_minimal = MqttClientOptions::builder()
.peer("localhost:1883")
.client_id("builder_minimal_client")
.build();
println!("Peer: {}", options_minimal.peer);
println!("Client ID: {}", options_minimal.client_id);
println!("Keep Alive: {} seconds", options_minimal.keep_alive);
println!("Clean Start: {}", options_minimal.clean_start);
println!("Auto ACK: {}", options_minimal.auto_ack);
println!("\nđ Example 2: With Authentication");
println!("{}", "-".repeat(60));
let options_auth = MqttClientOptions::builder()
.peer("localhost:1883")
.client_id("builder_auth_client")
.username("mqtt_user")
.password(b"secret_password".to_vec())
.build();
println!("Peer: {}", options_auth.peer);
println!("Client ID: {}", options_auth.client_id);
println!("Username: {:?}", options_auth.username);
println!("Password: [REDACTED]");
println!("\nđ Example 3: With Session Expiry (1 hour)");
println!("{}", "-".repeat(60));
let options_session = MqttClientOptions::builder()
.peer("localhost:1883")
.client_id("builder_session_client")
.clean_start(false)
.session_expiry_interval(3600) .build();
println!("Peer: {}", options_session.peer);
println!("Client ID: {}", options_session.client_id);
println!("Clean Start: {}", options_session.clean_start);
println!(
"Session Expiry: {:?} seconds",
options_session.session_expiry_interval
);
println!("\nđ Example 4: Full Featured Configuration");
println!("{}", "-".repeat(60));
let options_full = MqttClientOptions::builder()
.peer("localhost:1883")
.client_id("builder_full_client")
.username("admin")
.password(b"admin123".to_vec())
.clean_start(false)
.keep_alive(120)
.reconnect(true)
.auto_ack(true)
.sessionless(false)
.session_expiry_interval(7200) .build();
println!("Peer: {}", options_full.peer);
println!("Client ID: {}", options_full.client_id);
println!("Username: {:?}", options_full.username);
println!("Keep Alive: {} seconds", options_full.keep_alive);
println!("Clean Start: {}", options_full.clean_start);
println!("Reconnect: {}", options_full.reconnect);
println!("Auto ACK: {}", options_full.auto_ack);
println!("Sessionless: {}", options_full.sessionless);
println!(
"Session Expiry: {:?} seconds",
options_full.session_expiry_interval
);
println!("\nđ Example 5: Using Default + Modifications");
println!("{}", "-".repeat(60));
let mut options_default = MqttClientOptions::default();
options_default.peer = "localhost:8883".to_string();
options_default.client_id = "modified_default_client".to_string();
println!("Peer: {}", options_default.peer);
println!("Client ID: {}", options_default.client_id);
println!("Keep Alive: {} seconds", options_default.keep_alive);
println!("\nđ Example 6: Real Client Connection (if broker available)");
println!("{}", "-".repeat(60));
let options = MqttClientOptions::builder()
.peer("broker.emqx.io:1883")
.client_id("builder_example_working_client")
.clean_start(true)
.keep_alive(60)
.reconnect(false)
.auto_ack(true)
.build();
let handler = BuilderExampleHandler;
let config = TokioAsyncClientConfig::default();
match TokioAsyncMqttClient::new(options, Box::new(handler), config).await {
Ok(client) => {
let client = Arc::new(client);
println!("â
Client created successfully");
match client.connect_sync().await {
Ok(result) => {
println!("â
Connected successfully!");
println!(" Reason Code: {}", result.reason_code);
println!(" Session Present: {}", result.session_present);
if let Err(e) = client.disconnect().await {
eprintln!("â ī¸ Error disconnecting: {}", e);
}
}
Err(e) => {
println!(
"âšī¸ Could not connect to broker (expected if not running): {}",
e
);
}
}
}
Err(e) => {
println!("âšī¸ Could not create client: {}", e);
}
}
println!("\n{}", "=".repeat(60));
println!("â
Builder pattern examples completed!");
Ok(())
}
#[tokio::main]
async fn main() -> io::Result<()> {
run_example().await
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn test_example() {
run_example().await.unwrap();
}
}