pubsub_config_reciever + cli-local adj

noxis-rs/src/main.rs

@@ -14,6 +14,8 @@ use std::time::Duration;
 use tokio::sync::mpsc;
 use utils::*;
 use options::preboot::PrebootParams;
+use tokio::sync::{broadcast, oneshot};
+use options::config::v2::init_config_mechanism;
 
 #[tokio::main(flavor = "multi_thread")]
 async fn main() -> anyhow::Result<()>{
@@ -21,7 +23,25 @@ async fn main() -> anyhow::Result<()>{
 
     let _ = setup_logger();
 
-    info!("Runner is configurating...");
+    info!("Noxis is configurating...");
+
+    let (tx_brd, mut _rx_brd) = broadcast::channel::<Processes>(1);
+    let (_tx_oneshot, rx_oneshot) = oneshot::channel::<Processes>();
+    let mut handler: Vec<tokio::task::JoinHandle<()>> = vec![];
+
+    let config_module = tokio::spawn(async move {
+        let _ = init_config_mechanism(
+            rx_oneshot,
+            tx_brd,
+            preboot.clone()
+        ).await;
+    });
+
+    handler.push(config_module);
+
+    for i in handler {
+        let _ = i.await;
+    }
 
     // setting up redis connection \
     // then conf checks to choose the most actual \

noxis-rs/src/options/config.rs

@@ -1,6 +1,6 @@
 use super::structs::*;
 use log::{error, info, warn};
-use redis::{Client, Connection, PubSub};
+use redis::{Client, Connection};
 use std::fs::OpenOptions;
 use std::io::Write;
 use std::os::unix::process::CommandExt;
@@ -10,21 +10,32 @@ use std::{env, fs};
 use super::preboot::PrebootParams;
 use tokio::time::{Duration, sleep};
 // use redis::PubSub;
-use tokio::sync::oneshot::Receiver as OneShotReciever;
-use tokio::sync::broadcast::Sender as BroadcastSender;
+use tokio::sync::{ 
+    oneshot::{ Receiver as OneShotReciever, Sender as OneShotSender },
+    broadcast::Sender as BroadcastSender, broadcast::Receiver as BroadcastReceiver };
+use crate::utils::files::create_watcher;
+use std::fs::File;
+use inotify::EventMask;
 
 // const CONFIG_PATH: &str = "settings.json";
 
 pub mod v2 {
+    use std::{fmt::format, path::PathBuf};
+    use crate::utils::get_container_id;
+
     use super::*;
 
     pub async fn init_config_mechanism(
         // to handle cli config changes 
         _cli_oneshot: OneShotReciever<Processes>,
-        // to share local config with PRCS and CLI_PIPELINE modules 
-        _brd_tx : Arc<BroadcastSender<Processes>>
+        // to share local config with PRCS, CLI_PIPELINE and CONFIG modules 
+        brd_tx : BroadcastSender<Processes>,
+        // preboot params (args) 
+        params : Arc<PrebootParams>
         /*...*/
     ) { 
+        // channel for pubsub to handle local config pulling
+        let _local_config_brd_reciever = brd_tx.subscribe();
         /* local + pubsub + cli oneshot check */ 
     }
     pub async fn get_redis_connection(params: Arc<PrebootParams>) -> Option<Connection> {
@@ -34,62 +45,240 @@ pub mod v2 {
         let mut connection_delay:  u64  = 1;
         loop {
             if let Ok(client) = Client::open(format!("redis://{}/", &params.remote_server_url)) {
-                if let Ok(mut conn) = client.get_connection() {
-                    match crate::utils::get_container_id() {
-                        Some(channel_name) => {
-                            // let channel_name = channel_name.trim();
-                            let mut pubsub = conn.as_pubsub();
-                            if pubsub.subscribe(&channel_name.trim()).is_ok() {
-                                
-                                todo!()
-
-                            } else {
-                                error!("Cannot subscribe channel {}. Check Redis Server status", &channel_name);
-                            }
-                        },
-                        None => {
-                            error!("Cannot get channel name");
-                        }
-                    }
+                if let Ok(conn) = client.get_connection() {
+                    info!("Successfully opened Redis connection");
+                    return Some(conn);
                 }
             }
             error!("Error with subscribing Redis stream on update. Retrying in {} secs...", connection_delay);
             sleep(Duration::from_secs(connection_delay)).await;
             connection_delay *= 2;
         }
+    }
 
-        None
+    // loop checking redis pubsub 
+    async fn pubsub_config_reciever(
+        // to stop checking local config
+        local_conf_tx : OneShotSender<bool>,
+        params : Arc<PrebootParams>,
+        tx_brd_local : BroadcastReceiver<Processes>,
+    ) -> anyhow::Result<()>{
+        /*...*/
+        let mut tx_brd_local = tx_brd_local;
+        let mut _local_config = Processes::default();
+        return match get_redis_connection(params.clone()).await {
+            Some(mut conn) => {
+                // 
+                let mut pub_sub = conn.as_pubsub();
+                match pub_sub.subscribe(get_container_id().unwrap_or(String::from("default"))) {
+                    Err(er) => {
+                        error!("Cannot subscribe pubsub channel due to {}", &er);
+                        Err(anyhow::Error::msg(format!("Cannot subscribe pubsub channel due to {}", er)))
+                    },
+                    Ok(_) => {
+                        loop {
+                            // brd check 
+                            // if let Ok(new_lc) = tx_brd_local.recv().await {
+
+                            // }
+                            if !tx_brd_local.is_empty() {
+                                match tx_brd_local.recv().await {
+                                    Ok(lc) => _local_config = lc,
+                                    Err(er) => {
+                                        error!("Cannot get imported local config due to {}", &er);
+                                        return Err(anyhow::Error::msg(
+                                            format!("Cannot get imported local config due to {}", er))
+                                        )
+                                    }
+                                }
+                            }
+                            // pubsub check 
+                            if let Ok(msg) = pub_sub.get_message() {
+                                let payload : Result<String, _> = msg.get_payload(); 
+                                match payload {
+                                    Err(_) => error!("Cannot read new config from Redis channel. Check network or Redis configuration "),
+                                    Ok(payload) => {
+                                        if let Some(remote) = parse_extern_config(&payload) {
+                                            match config_comparing(&_local_config, &remote) {
+                                                ConfigActuality::Local => {
+                                                    warn!("Pulled new config from Redis channel. Current config is more actual ...");
+                                                },
+                                                ConfigActuality::Remote => {
+                                                    info!("Pulled new actual config from Redis channel, version - `{}`", remote.date_of_creation);
+                                                    // to stop watching local config file mechanism  
+                                                    let _ = local_conf_tx.send(true);
+                                                    let config_path = params.config.to_str().unwrap_or("settings.json");
+
+                                                    if save_new_config(&remote, &config_path).is_err() {
+                                                        error!("Error with saving new config to {}. Stopping pubsub mechanism...", config_path);
+                                                        return Err(anyhow::Error::msg(
+                                                            format!("Error with saving new config to {}. Stopping pubsub mechanism...", config_path)
+                                                        ))
+                                                    }
+                                                    return Ok(());
+                                                },
+                                            }
+                                        }
+                                        else {
+                                            warn!("Invalid config was pulled from Redis channel")
+                                        }
+                                    },
+                                }
+                            }
+                            // delay 
+                            sleep(Duration::from_millis(500)).await;
+                        }
+                    },
+                }
+            },
+            None => Err(anyhow::Error::msg("Cannot create Redis connection"))
+        }
     }
     
     // 
-    pub async fn local_config_reciever(
+    async fn local_config_reciever(
         params : Arc<PrebootParams>, 
         pubsub_oneshot : OneShotReciever<bool>,
+        cli_oneshot : OneShotReciever<bool>,
         brd_tx : Arc<BroadcastSender<Processes>>,
         /*...*/
-    ) { 
+    ) -> anyhow::Result<()> { 
         /*...*/
 
-        // {:1} if local config is not exist -> cannot create watcher -> None 
-        // {:2} if local config exists -> load_processes
-        // 									|
-        // 									| [Ok(Processes)]
-        // 									 -> 1) broadcast sending parsed config to PRCS and CLI_PIPELINE
-        // 										2) watcher in loop to deny local changes 
-        // 									|
-        // 									| Err(_)
-        // 									 -> 
-        // ????
+        // borrowing as mut 
+        let mut pubsub_oneshot = pubsub_oneshot; 
+        let mut cli_oneshot = cli_oneshot;
+        // fill with default empty config, mut to change later
+        let mut _current_config = Processes::default();
+        // PathBuf to &str to work with local config path as slice
+        let local_config_path = params
+            .config
+            .to_str()
+            .unwrap_or("settings.json");
+
+        match load_processes(local_config_path) {
+            // if local exists
+            Some(conf) => {
+                info!("Local config `{}` was found.", &conf.date_of_creation);
+                _current_config = conf;
+                if let Err(er) = brd_tx.send(_current_config.clone()) {
+                    error!("Cannot share local config with broadcast due to {}", er);
+                }
+            },
+            // if local is not exist
+            None => {
+                warn!("Local config wasn't found. Waiting for new ...");
+                return Err(anyhow::Error::msg("No local config"));
+                // ...
+            },
+        }
+
+        // 100% local exists here  
+        // create watcher on local config file
+        match create_watcher("", local_config_path).await {
+            Ok(mut watcher) => {
+                loop {
+                    let mut need_to_export_config = false; 
+                    // return situations here 
+                    // 1) oneshot signal 
+                    // 2) if config was deleted -> recreate and fill with current config that is held here
+                    // 3) if config was changed -> fill with current config that is held here
+
+                    // catching signal from pubsub
+                    // it's because pubsub mech pulled new valid and actual config and now it's time to ...
+                    // ... overwrite local config file and restart main thread  
+                    if let Ok(_) = pubsub_oneshot.try_recv() {
+                        sleep(Duration::from_secs(1)).await;
+                        return Ok(());
+                    }
+
+                    // catching signal from cli
+                    // it's because cli mech pulled new valid and actual config and now it's time to ...
+                    // ... overwrite local config file and restart main thread (like in previous mechanism)  
+                    if let Ok(_) = cli_oneshot.try_recv() {
+                        sleep(Duration::from_secs(1)).await;
+                        return Ok(());
+                    }
+
+                    // ! IF NOXIS NEEDS TO RECREATE OR CHANGE LOCAL CONFIG NEED TO DRAIN THIS ACTIVITY ...
+                    // ! ... FROM WATCHER"S BUFFER
+
+                    // existing check
+                    if !params.config.exists() {
+                        warn!("Local config file was deleted or moved. Recreating new one with saved data ...");
+                        need_to_export_config = true;
+                    } else {
+                        // changes check
+                        let mut buffer = [0; 128];
+                        let events  = watcher.read_events(&mut buffer);
+                        if events.is_ok() {
+                            let events: Vec<EventMask> = events
+                                .unwrap()
+                                .map(|mask| mask.mask)
+                                .filter(|mask| {
+                                    *mask == EventMask::MODIFY || *mask == EventMask::DELETE_SELF
+                                })
+                                .collect();
+                            if !events.is_empty() {
+                                warn!("Local config file was overwritten. Discarding changes ...");
+                                need_to_export_config = true;
+                            }
+                        }
+                    }
+                    // exporting data
+                    if need_to_export_config {
+                        if let Err(er) = export_saved_config_data_locally(&params.config, &_current_config).await {
+                            error!("Cannot save actual imported config due to {}", er);
+                        } else {
+                            // recreation watcher (draining activity buffer mechanism)
+                            // if local config file was deleted and recreated
+                            // if local config file was modified locally
+                            match create_watcher("", local_config_path).await {
+                                Ok(new) => watcher = new,
+                                Err(er) => error!("Cannot create new watcher due to {}", er),
+                            }
+                        }
+                    }
+                    sleep(Duration::from_millis(500)).await;
+                }
+            },
+            Err(_) => {
+                error!("Cannot create watcher on local config file `{}`. Deinitializing warding local config mechanism...", local_config_path);
+                return Err(anyhow::Error::msg("Cannot create watcher on local config file"));
+            },
+        }
+
     }
 
     // [:IN-TEST]
-    pub async fn cli_config_reciever(cli_oneshot: OneShotReciever<Processes>) -> Option<Processes> { 
+    async fn from_cli_config_reciever(
+        cli_oneshot: OneShotReciever<Processes>,
+        to_local_tx: OneShotSender<bool>
+    ) -> Option<Processes> { 
         /* match awaits til channel*/
         match cli_oneshot.await {
-            Ok(config_from_cli) => return Some(config_from_cli),
+            Ok(config_from_cli) => {
+                info!("New actual config `{}` from CLI was pulled. Saving and restaring ...", &config_from_cli.date_of_creation);
+                let _ = to_local_tx.send(true);
+                Some(config_from_cli)
+            },
             _ => None,
         }
     }
+
+    async fn export_saved_config_data_locally(
+        config_file_path: &PathBuf, 
+        current_config: &Processes
+    ) -> anyhow::Result<()> {
+
+        let mut file = File::create(config_file_path)?;
+        Ok(
+            file.write_all(
+                serde_json::to_string_pretty(current_config)?.as_bytes()
+            )?
+        )
+        // Ok(())
+    }
 }
 
 
@@ -478,6 +667,9 @@ pub async fn subscribe_config_stream(actual_prcs: Arc<Processes>, params: Arc<Pr
 /// *depends on* : `Processes`, `ConfigActuality`
 /// 
 fn config_comparing(local: &Processes, remote: &Processes) -> ConfigActuality {
+    if local.is_default() {
+        return ConfigActuality::Remote;
+    }
     let local_date: u64 = local.date_of_creation.parse().unwrap();
     let remote_date: u64 = remote.date_of_creation.parse().unwrap();