remote conf update sub + reworked signals handling (need to fix)

Cargo.toml

@@ -3,6 +3,9 @@ name = "runner-rs"
 version = "0.5.5"
 edition = "2021"
 
+[profile.dev]
+debug = true
+
 [dependencies]
 chrono = "0.4.38"
 env_logger = "0.11.3"

settings.json

@@ -1,5 +1,5 @@
 {
-  "dateOfCreation": "1721381809090",
+  "dateOfCreation": "1721381809101",
   "processes": [
     {
       "name": "web-server",
@@ -55,7 +55,7 @@
             "hostname": "google.com",
             "port": 443,
             "triggers": {
-              "wait": 14,
+              "wait": 5,
               "delay": 1,
               "onLost": "stop"
             }
@@ -64,7 +64,7 @@
             "hostname": "localhost",
             "port": 8080,
             "triggers": {
-              "wait": 10,
+              "wait": 6,
               "delay": 2,
               "onLost": "hold"
             }

src/config.rs

@@ -1,7 +1,10 @@
 use crate::structs::*;
 use log::{error, info, warn};
 use redis::{Client, Commands, Connection, RedisResult};
-use std::fs;
+use std::{env, fs};
+use std::os::unix::process::CommandExt;
+use std::process::Command;
+use std::sync::Arc;
 use tokio::time::Duration;
 
 static CONFIG_PATH: &str = "settings.json";
@@ -22,6 +25,7 @@ pub fn get_actual_config() -> Option<Processes> {
     // let mut local = load_processes(&CONFIG_PATH);
     match load_processes(CONFIG_PATH) {
         Some(local_conf) => {
+            info!("Found local configuration, version - {}", &local_conf.date_of_creation);
             if let Some(remote_conf) = once_get_remote_configuration("redis://localhost") {
                 return match config_comparing(&local_conf, &remote_conf) {
                     ConfigActuality::Local => {
@@ -75,7 +79,10 @@ fn once_get_remote_configuration(serv_info: &str) -> Option<Processes> {
                                     return None;
                                 }
                                 match parse_extern_config(&config[0].1[0].1) {
-                                    Some(prcs) => Some(prcs),
+                                    Some(prcs) => {
+                                        info!("Config {} was pulled from Redis-Server", &prcs.date_of_creation);
+                                        Some(prcs)
+                                    },
                                     None => {
                                         error!("Invalid configuration was pulled (PARSING_ERROR). Check configs state!");
                                         None
@@ -151,14 +158,9 @@ fn get_stream_info_watcher(conn: &mut Connection) {
 }
 fn invalid_config_watcher(conn: &mut Connection) -> Processes {
     loop {
-        let res: Res = conn.xrevrange_count("config_stream", "+", "-", 1);
+        if let Some(prcs) = get_remote_config(conn) {
-        if res.is_ok() {
+            info!("Got new config from Redis-Server, version - {}", &prcs.date_of_creation);
-            let config = &res.unwrap()[0];
+            return prcs;
-            if !config.is_empty() {
-                if let Some(conf) = parse_extern_config(&config[0].1[0].1) {
-                    return conf;
-                }
-            }
         }
         error!("Got INVALID configuration. Update config! Retrying...");
         std::thread::sleep(Duration::from_secs(4));
@@ -167,6 +169,53 @@ fn invalid_config_watcher(conn: &mut Connection) -> Processes {
 
 // ! end of watchers
 
+fn get_remote_config(conn: &mut Connection) -> Option<Processes> {
+    let res: Res = conn.xrevrange_count("config_stream", "+", "-", 1);
+    if res.is_ok() {
+        let config = &res.unwrap()[0];
+        if !config.is_empty() {
+            return parse_extern_config(&config[0].1[0].1);
+        }
+    }
+    None
+}
+
+fn restart_main_thread() -> std::io::Result<()>{
+    let current_exe = env::current_exe()?;
+    Command::new(current_exe)
+        .exec();
+    Ok(())
+}
+pub async fn subscribe_config_stream(actual_prcs: Arc<Processes>) -> Result<(), CustomError> {
+    if let Ok(client) = Client::open("redis://localhost") {
+        if let Ok(mut conn) = client.get_connection() {
+            info!("Runner subscribed on config update");
+            loop {
+                tokio::time::sleep(Duration::from_secs(30)).await;
+                if let Some(prcs) = get_remote_config(&mut conn) {
+                    match config_comparing(&actual_prcs, &prcs) {
+                        ConfigActuality::Remote => {
+                            info!("New config was pulled. Saving and restarting...");
+                            if save_new_config(&prcs, CONFIG_PATH).is_err() {
+                                error!("Error with saving new config to {}", &CONFIG_PATH);
+                                return Err(CustomError::Fatal)
+                            }
+                            if restart_main_thread().is_err() {
+                                error!("Error with restarting Runner. Stopping sub mechanism...");
+                                return Err(CustomError::Fatal)
+                            }
+                        },
+                        _ => continue,
+                    }
+                    return Ok(());
+                }
+            }
+        }
+    }
+    error!("Error with subscribing Redis stream on update. Working only with selected config...");
+    Err(CustomError::Fatal)
+}
+
 fn config_comparing(local: &Processes, remote: &Processes) -> ConfigActuality {
     let local_date: u64 = local.date_of_creation.parse().unwrap();
     let remote_date: u64 = remote.date_of_creation.parse().unwrap();

src/files.rs

@@ -33,6 +33,7 @@ pub async fn file_handler(
             }
             match file.triggers.on_delete.as_str() {
                 "stay" => {
+                    tx.send(9).await.unwrap();
                     continue;
                 }
                 "stop" => {

src/main.rs

@@ -12,6 +12,7 @@ use log::{error, info};
 use logger::setup_logger;
 use signals::set_valid_destructor;
 use std::sync::Arc;
+use std::time::Duration;
 use structs::*;
 use tokio::sync::mpsc;
 use utils::*;
@@ -20,7 +21,7 @@ use utils::*;
 async fn main() {
     let _ = setup_logger();
 
-    log::info!("Runner is configurating...");
+    info!("Runner is configurating...");
 
     // setting up redis connection \
     // then conf checks to choose the most actual \
@@ -29,13 +30,13 @@ async fn main() {
         std::process::exit(101);
     });
 
-    log::info!(
+    info!(
         "Current runner configuration: {}",
         &processes.date_of_creation
     );
-    log::info!("Runner is ready. Initializing...");
+    info!("Runner is ready. Initializing...");
 
-    if processes.processes.len() == 0 {
+    if processes.processes.is_empty() {
         error!("Processes list is null, runner-rs initialization is stopped");
         return;
     }
@@ -44,7 +45,7 @@ async fn main() {
     let mut senders: Vec<Arc<mpsc::Sender<u8>>> = vec![];
 
     for proc in processes.processes.iter() {
-        log::info!(
+        info!(
             "Process '{}' on stage: {}. Depends on {} file(s), {} service(s)",
             proc.name,
             proc.path,
@@ -65,16 +66,28 @@ async fn main() {
         });
         handler.push(event);
     }
+
     // destructor addition
     handler.push(tokio::spawn(async move {
         if let Err(_) = set_valid_destructor(Arc::new(senders)).await {
-            error!("Linux signals handler creation failed. Returning...");
+            error!("Linux signals handler creation failed. Terminating main thread...");
+            return;
+        }
+        // todo: rework this temp construction, use async/await in signals mod
+        tokio::time::sleep(Duration::from_millis(200)).await;
+        info!("End of job. Terminating main thread...");
+        std::process::exit(0);
+    }));
+
+    // remote config update subscription
+    handler.push(tokio::spawn(async move {
+        if let Err(_) = subscribe_config_stream(Arc::new(processes)).await {
             return;
         }
     }));
+
     for i in handler {
         i.await.unwrap();
     }
-    info!("End of job. Terminating main thread...");
     return;
 }

src/services.rs

@@ -31,9 +31,9 @@ pub async fn service_handler(
                     }
                 }
                 "hold" => {
-                    if is_frozen(name).await {
+                    // if is_frozen(name).await {
-                        return Err(CustomError::Fatal);
+                    //     return Err(CustomError::Fatal);
-                    }
+                    // }
                     if looped_service_connecting(name, serv).await.is_err() {
                         tx.send(6).await.unwrap();
                         tokio::time::sleep(Duration::from_millis(400)).await;
@@ -53,6 +53,7 @@ pub async fn service_handler(
 }
 
 async fn looped_service_connecting(name: &str, serv: &Services) -> Result<(), CustomError> {
+    let mut counter = 0;
     if serv.triggers.wait == 0 {
         loop {
             tokio::time::sleep(Duration::from_secs(serv.triggers.delay.into())).await;
@@ -79,8 +80,9 @@ async fn looped_service_connecting(name: &str, serv: &Services) -> Result<(), Cu
         let start = Instant::now();
         while start.elapsed().as_secs() < serv.triggers.wait.into() {
             tokio::time::sleep(Duration::from_secs(serv.triggers.delay.into())).await;
+            counter += 1;
             warn!(
-                "Attempting to connect from {} process to {}:{}",
+                "{counter} Attempting to connect from {} process to {}:{}",
                 &name, &serv.hostname, &serv.port
             );
             match check_service(&serv.hostname, &serv.port).await {

src/utils.rs

@@ -95,7 +95,7 @@ pub async fn run_daemons(
                     },
                     // // 9 - File-dependency change -> staying (after check)
                     9 => {
-                        warn!("File-dependency warning (file changed). Ignoring on {} process...", &proc.name);
+                        warn!("File-dependency warning (file changed). Ignoring event on {} process...", &proc.name);
                     },
 
                     // 10 - Process unfreaze call via file handler