2 Commits
ec2a85e1ae
...
eb67528d16
2 Commits
ec2a85e1ae
...
eb67528d16
| Author | SHA1 | Message | Date |
|---|---|---|---|
 n0m1s
|
eb67528d16
|
exit gracefully when MQTT server is disconnected | 2 years ago |
|
n0m1s
|
4553a1b65f
|
added logging | 2 years ago |
6 changed files with 222 additions and 20 deletions
Unified View
Diff Options
-
+130 -4Cargo.lock
-
+4 -1Cargo.toml
-
+1 -0example.toml
-
+11 -2src/config.rs
-
+72 -11src/main.rs
-
+4 -2src/mqtt.rs
+ 130
- 4
Cargo.lock
View File
+ 4
- 1
Cargo.toml
View File
+ 1
- 0
example.toml
View File
+ 11
- 2
src/config.rs
View File
+ 72
- 11
src/main.rs
View File
| @ -1,32 +1,93 @@ | |||||
| use std::error::Error; | |||||
| use std::{thread, error::Error, time::Duration}; | |||||
| use log::{error, warn, info, debug}; | |||||
| use simplelog::{ TermLogger, LevelFilter, ConfigBuilder, TerminalMode, ColorChoice}; | |||||
| use paho_mqtt as paho; | |||||
| mod cli_args; | mod cli_args; | ||||
| mod config; | mod config; | ||||
| mod mqtt; | mod mqtt; | ||||
| fn main() -> Result<(), Box<dyn Error>> { | |||||
| let args = cli_args::parse(); | |||||
| let config = config::from_file(&args.configuration)?; | |||||
| //------------------------------------------------------------------------------ | |||||
| println!("{:#?}", config); //TODO: remove | |||||
| fn try_reconnect( mqtt: &paho::Client, timeout: u32 ) -> bool { | |||||
| warn!("MQTT server lost, trying to reconnect..."); | |||||
| for _ in 0..timeout { | |||||
| thread::sleep(Duration::from_secs(1)); | |||||
| if mqtt.reconnect().is_ok() { | |||||
| info!("MQTT server reconnected"); | |||||
| return true; | |||||
| } | |||||
| } | |||||
| error!("MQTT server timed out"); | |||||
| false | |||||
| } | |||||
| //------------------------------------------------------------------------------ | |||||
| fn handle_msg( msg: &paho::Message ) | |||||
| { | |||||
| info!("New message: {}", msg) | |||||
| } | |||||
| //------------------------------------------------------------------------------ | |||||
| fn main_impl( args: &cli_args::Args ) -> Result<(), Box<dyn Error>> { | |||||
| let config = config::from_file(&args.configuration)?; | |||||
| debug!("Config: {:?}", config); | |||||
| // init MQTT | |||||
| let mqtt = mqtt::create( &config.mqtt )?; | let mqtt = mqtt::create( &config.mqtt )?; | ||||
| let rx = mqtt.start_consuming(); | let rx = mqtt.start_consuming(); | ||||
| mqtt::subscribe( &mqtt, &config.topics, args.verbose ); | mqtt::subscribe( &mqtt, &config.topics, args.verbose ); | ||||
| // handle ^C signal to quit gracefully | |||||
| let ctrlc_mqtt = mqtt.clone(); | |||||
| ctrlc::set_handler(move || { | |||||
| ctrlc_mqtt.stop_consuming() | |||||
| })?; | |||||
| // main event loop | |||||
| for msg in rx.iter() { | for msg in rx.iter() { | ||||
| if let Some(msg) = msg { | |||||
| println!("{}", msg); | |||||
| } else if !mqtt.is_connected() { | |||||
| println!("disconnected"); | |||||
| if !mqtt.is_connected() && !try_reconnect(&mqtt, config.mqtt.timeout) { | |||||
| break; | break; | ||||
| } | } | ||||
| if let Some(msg) = msg { | |||||
| handle_msg(&msg) | |||||
| } | |||||
| } | } | ||||
| if mqtt.is_connected() { | if mqtt.is_connected() { | ||||
| println!("Disconnecting"); | |||||
| //mqtt.unsubscribe_many(topics) | |||||
| info!("Disconnecting"); | |||||
| mqtt.disconnect(None)?; | |||||
| } | } | ||||
| Ok(()) | Ok(()) | ||||
| } | } | ||||
| //------------------------------------------------------------------------------ | |||||
| fn main() { | |||||
| let args = cli_args::parse(); | |||||
| // set up logging | |||||
| TermLogger::init( | |||||
| match args.verbose { | |||||
| true => LevelFilter::Debug, | |||||
| false => LevelFilter::Info | |||||
| }, | |||||
| ConfigBuilder::new() | |||||
| .set_time_format_rfc3339() | |||||
| .set_time_offset_to_local().unwrap() | |||||
| .build(), | |||||
| TerminalMode::Mixed, | |||||
| ColorChoice::Auto | |||||
| ).unwrap(); | |||||
| // actually run the program | |||||
| if let Err(e) = main_impl(&args) { | |||||
| error!("Error: {}", e) | |||||
| } | |||||
| } | |||||