monitor/noxis-rs/src/options/cli_pipeline.rs

use log::{error, info, warn};
use tokio::net::{ UnixStream, UnixListener };
use tokio::sync::{Mutex, OnceCell};
use tokio::time::{sleep, Duration};
use std::fs;
use std::sync::Arc;
use tokio::io::{ AsyncWriteExt, AsyncReadExt};
use noxis_cli::Cli;
use super::structs::Processes;

use super::preboot::PrebootParams;

type ConfigGateway = tokio::sync::oneshot::Sender<Processes>;
type ProcessedConfigGateway = Arc<Mutex<OnceCell<ConfigGateway>>>;

/// # Fn `init_cli_pipeline`
/// ## for catching all input requests from CLI
/// 
/// *input*      : -
///
/// *output*     :  `anyhow::Result<()>` to wrap errors 
///
/// *initiator*  :  fn `main`
///
/// *managing*   :  `TcpListener` object to handle requests 
///
/// *depends on* :  -
/// 
pub async fn init_cli_pipeline(
    params: Arc<PrebootParams>, 
    config : Arc<Processes>,
    config_gateway : ConfigGateway,
) -> anyhow::Result<()> {
    let socket_path = &params.self_socket;
    let _ = fs::remove_file(socket_path);

    let config_gateway = Arc::new(
        Mutex::new(
            OnceCell::new_with(Some(config_gateway))
        )
    );

    match UnixListener::bind(socket_path) {
        Ok(list) => {
            // TODO: remove `unwrap`s
            info!("Listening on {}", socket_path.display());
            std::env::set_var("NOXIS_SOCKET_PATH", socket_path);
            loop {
                match list.accept().await {
                    Ok((socket, _)) => {
                        // ??? maybe errors on async work with data transfering between modules 
                        let params = params.clone();
                        let config = config.clone();
                        let config_gateway = config_gateway.clone();
                        tokio::spawn(async move {
                            process_connection(socket, params.clone(), config.clone(), config_gateway.clone()).await;
                        });
                    },
                    Err(er) => {
                        error!("Cannot poll connection to CLI due to {}", er);
                        sleep(Duration::from_millis(300)).await;
                    },
                }
            }
            // Ok(())
        },
        Err(er) => {
            error!("Failed to open UnixListener for CLI");
            Err(er.into())
        },
    }
}

/// # Fn `process_connection`
/// ## for processing input CLI requests
/// 
/// *input*      :  mut stream: `TcpStream`
///
/// *output*     :  - 
///
/// *initiator*  :  fn `init_cli_pipeline`
///
/// *managing*   :  mutable object of `TcpStream`
///
/// *depends on* :  `tokio::net::TcpStream`
/// 
async fn process_connection(mut stream: UnixStream, params: Arc<PrebootParams>, config : Arc<Processes>, cfg_gateway : ProcessedConfigGateway) {    
    let mut buf = vec![0; 1024];
    match stream.read(&mut buf).await {
        Ok(0) => {
            info!("Client disconnected ");
        },
        Ok(n) => {
            buf.truncate(n);
            info!("CLI have sent {} bytes", n);
            match serde_json::from_slice::<Cli>(&buf) {
                Ok(cli) => {
                    info!("Received CLI request: {:?}", cli);
                    let response = match process_cli_cmd(cli, params.clone(), config, cfg_gateway.clone()).await {
                        Ok(response) => {
                            response
                        },
                        Err(er) => {
                            let error_msg = format!("Error: {}", er);
                            error!("{}", &error_msg);
                            error_msg
                        },
                    };
                    if let Err(e) = stream.write_all(response.as_bytes()).await {
                        error!("Failed to send response: {}", e);
                    }
                }
                Err(e) => {
                    error!("Failed to parse CLI request: {}", e);
                }
            }
        },
        Err(e) => error!("Failed to read from socket: {}", e),
    }
    let _ = stream.shutdown().await;
}


async fn process_cli_cmd(cli : Cli, params: Arc<PrebootParams>, global_config : Arc<Processes>, cfg_gateway: ProcessedConfigGateway) -> anyhow::Result<String> {
    use noxis_cli::{Commands, ConfigAction};
    return match cli.command {
        Commands::Config(config) => {
            match config.action {
                ConfigAction::Show(env ) => {
                    if env.is_env {
                        Ok(serde_json::to_string_pretty(params.as_ref())?)
                    } else {
                        /* */
                        Ok(serde_json::to_string_pretty(global_config.as_ref())?)
                    }
                },
                ConfigAction::Reset => {
                    Err(anyhow::Error::msg("It's temporarly forbidden to reset current config using CLI-util"))
                },
                ConfigAction::Local(cfg) => {
                    if cfg.is_json {
                        /* */
                        let new_config = serde_json::from_str::<Processes>(&cfg.config)?;
                        let new_version = new_config.get_version().to_string();

                        use super::{config::config_comparing, structs::ConfigActuality};
                        
                        return match config_comparing(&global_config, &new_config) {
                            ConfigActuality::Remote => {
                                let cfg_gateway = cfg_gateway.clone();
                                tokio::spawn(async move {
                                    let mut lock = cfg_gateway.lock().await;
                                    match lock.take() {
                                        Some(channel) => {
                                            let _ = channel.send(new_config);
                                        },
                                        None => error!("Cannot update confif due to channel sender loss"),
                                    }
                                });
                                Ok(format!("Ok. Saving and reloading with version {}", new_version))
                            },
                            _ => Err(anyhow::Error::msg(format!("Local config (version: {}) is more actual", global_config.get_version()))),
                        }
                    } else {
                        Err(anyhow::Error::msg("It's temporarly forbidden to set config in non-json mode"))
                    } 
                },
                ConfigAction::Remote => {Ok(params.remote_server_url.clone())},
                /* */
                _ => Err(anyhow::Error::msg("Unrecognized command from CLI"))
            }
        },
        /* */
        Commands::Status => Ok(String::from("Ok")),
        _ => Ok(String::from("Ok"))
    }
}