romp 0.5.2

//! Holds the romp server details in memory in such a way that the data can be read by the
//! API used by `xtomp-console`. `xtomp-console` has a frontend HTML/JavaScript component and a backend
//! in nodejs.  This code mimics the backend part inside the romp server itself, since we support a
//! sufficient sub-set of HTTP.  The term "ngin" comes from nginx the base for xtomp. Naming conventions
//! in this code map to similar code in `xtomp-console`.

use std::sync::atomic::{AtomicUsize, AtomicI64, Ordering};
use std::{fs, process};
use std::fs::File;
use std::io::{BufReader, BufRead};

use crate::init::CONFIG;
use crate::init::SERVER;
use crate::message::response::{get_http_404_msg, SERVER_TAG};
use crate::message::stomp_message::{StompMessage, StompCommand, Ownership, Header};
use crate::message::stomp_message::MessageType::Http;
use chrono::Utc;
use crate::workflow::http_router::insert_http_filter;
use crate::workflow::console::filter::ConsoleFilter;
use std::sync::RwLock;

// inner structs

struct DestinationInfo {
    size: usize,
    queue_len: usize,
    delta: usize,
    total: usize,
}

struct NginInfo {
    /// contains pre-formatted JSON for the /ngin-config/ response
    ngin_config_json: String,
    // TODO how does AtomicPtr work, should be able to swap a pointer to DestinationInfo, but I cant get it to work, it will not accept a Box
    destinations: Box<Vec<RwLock<DestinationInfo>>>
}

/// Provides access to the data required for xtomp-console.
pub struct Ngin {
    pub enabled: bool,
    ngin_info: NginInfo,
    connections_count: AtomicUsize,
    connections_total: AtomicUsize,
    messages_total: AtomicUsize,
    lastmod: AtomicI64,
}

impl Ngin {

    /// Create a new Ngin instance, requires CONFIG to be available.
    pub(crate) fn new() -> Ngin {
        let mut ngin_info = NginInfo {
            ngin_config_json: Ngin::serialize_ngin_config(),
            destinations: Box::new(vec![])
        };
        for _destination in CONFIG.destinations.values() {
            ngin_info.destinations.push(RwLock::new(DestinationInfo {
                size: 0,
                queue_len: 0,
                delta: 0,
                total: 0,
            }));
        }
        Ngin {
            enabled: CONFIG.enable_console,
            ngin_info,
            connections_count: AtomicUsize::new(0),
            connections_total: AtomicUsize::new(0),
            messages_total: AtomicUsize::new(0),
            lastmod: AtomicI64::new(0)
        }
    }

    pub(crate) fn init(&self) {
        insert_http_filter(Box::new(ConsoleFilter::new()));
    }

    // mutators //

    /// copy this data once every 60 seconds
    pub(crate) fn set_stats_cc(&self) {
        self.connections_count.store(SERVER.connections_count.load(Ordering::Relaxed), Ordering::Relaxed);
        self.connections_total.store(SERVER.connections_total.load(Ordering::Relaxed), Ordering::Relaxed);
        self.messages_total.store(SERVER.messages_total.load(Ordering::Relaxed), Ordering::Relaxed);
        self.lastmod.store(Utc::now().timestamp(), Ordering::Relaxed);
    }

    pub(crate) fn set_stats_destination(&self, idx: usize, size: usize, q_size: usize, delta: usize, total: usize) {
        if let Ok(mut destination_info) = self.ngin_info.destinations[idx].write() {
            destination_info.size = size;
            destination_info.queue_len = q_size;
            destination_info.delta = delta;
            destination_info.total = total;
        }
    }

    // HTTP responses //

    /// response for /romp/ngin-config/
    pub fn response_ngin_config(&self) -> StompMessage {
        if ! self.enabled {
            return get_http_404_msg();
        }
        // horrible amount of cloning string data here. such is rust.
        let mut message = StompMessage::new(Ownership::Session);
        message.message_type = Http;
        message.command = StompCommand::Ok;
        message.add_header("server", SERVER_TAG);
        message.add_header("access-control-allow-origin", "*");
        message.add_header("content-type", "application/json");
        message.add_header("cache-control", "no-cache");
        let json = &self.ngin_info.ngin_config_json;
        message.push_header(Header {
            name: "content-length".to_string(),
            value: json.len().to_string()
        });
        message.add_body(json.as_bytes());
        message
    }

    /// response for /romp/ngin-info/
    pub fn response_ngin_info(&self) -> StompMessage {
        if ! self.enabled {
            return get_http_404_msg();
        }
        let mut message = StompMessage::new(Ownership::Session);
        message.message_type = Http;
        message.command = StompCommand::Ok;
        message.add_header("server", SERVER_TAG);
        message.add_header("access-control-allow-origin", "*");
        message.add_header("content-type", "application/json");
        message.add_header("cache-control", "no-cache");
        let json = self.serialize_ngin_info();
        message.push_header(Header {
            name: "content-length".to_string(),
            value: json.len().to_string()
        });
        message.add_body(json.as_bytes());
        message
    }

    /// response for /romp/destination-info/?destination=memtop-a
    pub fn response_destination_info(&self, destination: &str) -> StompMessage {
        if ! self.enabled {
            return get_http_404_msg();
        }
        let mut idx: usize = usize::max_value();
        for (_i, (name, _config)) in SERVER.iter().enumerate() {
            if destination.eq(name) {
                idx = 1;
                break;
            }
        }
        if idx == usize::max_value() {
            return get_http_404_msg();
        }

        let mut message = StompMessage::new(Ownership::Session);
        message.message_type = Http;
        message.command = StompCommand::Ok;
        message.add_header("server", SERVER_TAG);
        message.add_header("access-control-allow-origin", "*");
        message.add_header("content-type", "application/json");
        message.add_header("cache-control", "no-cache");
        let json = self.serialize_destination_info(idx);
        message.push_header(Header {
            name: "content-length".to_string(),
            value: json.len().to_string()
        });
        message.add_body(json.as_bytes());
        message
    }

    // ngin-info //

    // the following code presumes kernel `/proc` interfaces are stable.
    // could process the data a little more but this code should match ngin-info.js in xtomp-console

    /// return JSON representation of `/proc/pid/...` data relevant to the `romp` process.
    fn serialize_ngin_info(&self) -> String {
        format!("{{ {}, {}, {}, {}, {} }}",
                Ngin::serialize_proc_pid(),
                Ngin::serialize_proc_cmdline(),
                Ngin::serialize_proc_sockstat(),
                Ngin::serialize_proc_status(),
                self.serialize_cc())
    }

    /// return a JSON string
    fn serialize_proc_pid() -> String {
        format!("\"pid\": {}", process::id())
    }

    /// reads `/proc/pid/cmdline` presuming this is a valid JSON string, and return a JSON string
    fn serialize_proc_cmdline() -> String {
        let mut path = String::from("/proc/");
        path.push_str(process::id().to_string().as_str());
        path.push_str("/cmdline");
        let cmdline = fs::read_to_string(path).expect("Unable /proc/[pid]/cmdline");
        format!("\"cmdline\": \"{}\"", cmdline.trim_matches(char::from(0)).trim())
    }

     /// reads `/proc/pid/net/sockstat` presuming this contains string that do not
     /// need escaping and returns a JSON string
    fn serialize_proc_sockstat() -> String {
        let mut path = String::from("/proc/");
        path.push_str(process::id().to_string().as_str());
        path.push_str("/net/sockstat");
        let f = File::open(path).expect("Unable to read /proc/[pid]/net/sockstat");
        let f = BufReader::new(f);

        let mut json = String::from("\"sock\": {");
        for line in f.lines() {
            let line = line.expect("Unable to read /proc/[pid]/net/sockstat");
            if line.starts_with("sockets") {
                let parts: Vec<&str> = line.split(":").collect();
                if parts.len() == 2 {
                    json.push_str("\"sockets\":\"");
                    json.push_str(parts[1].trim());
                    json.push_str("\",");
                }
            }
            if line.starts_with("TCP") {
                let parts: Vec<&str> = line.split(":").collect();
                if parts.len() == 2 {
                    json.push_str("\"tcp\":\"");
                    json.push_str(parts[1].trim());
                    json.push_str("\",");
                }
            }
        }
        if json.ends_with(",") {
            json.remove(json.len() - 1);
        }
        json.push_str("}");
        json
    }

    fn serialize_proc_status() -> String {
        let mut path = String::from("/proc/");
        path.push_str(process::id().to_string().as_str());
        path.push_str("/status");
        let f = File::open(path).expect("Unable to read /proc/[pid]/status");
        let f = BufReader::new(f);

        let mut json = String::from("\"proc\": {");
        for line in f.lines() {
            let line = line.expect("Unable to read /proc/[pid]/status");
            if line.starts_with("Vm") {
                let parts: Vec<&str>  = line.split(":").collect();
                if parts.len() == 2 {
                    json.push_str("\"");
                    json.push_str(parts[0]);
                    json.push_str("\":\"");
                    json.push_str(parts[1].trim());
                    json.push_str("\",");
                }
            }
        }
        if json.ends_with(",") {
            json.remove(json.len() - 1);
        }
        json.push_str("}");
        json
    }

    fn serialize_cc(&self) -> String {
        let mut json = String::from("\"cc\": {");
        json.push_str(format!("\"sz\": \"{}\",", self.connections_count.load(Ordering::Relaxed)).as_str());
        json.push_str(format!("\"up\": \"{}\",", (SERVER.uptime.load(Ordering::Relaxed)) / 1000).as_str());
        json.push_str(format!("\"Σ\": \"{}\",", self.connections_total.load(Ordering::Relaxed)).as_str());
        json.push_str(format!("\"Σµ\": \"{}\",", self.messages_total.load(Ordering::Relaxed)).as_str());
        json.push_str("\"m\": \"0\",");
        json.push_str(format!("\"lastmod\": \"{}\"", self.lastmod.load(Ordering::Relaxed)).as_str());
        json.push_str("}");
        json
    }


    // ngin-config //

    fn serialize_ngin_config() -> String {
        format!("{{ {}, {}, {}, {} }}",
                Ngin::serialize_listen(),
                Ngin::serialize_timeout(),
                Ngin::serialize_websockets(),
                Ngin::serialize_destinations())
    }

    fn serialize_listen() -> String {
        let mut json = String::from("\"listen\": [");
        for port in &CONFIG.ports {
            json.push_str(port.to_string().as_str());
            json.push_str(",");
        }
        if json.ends_with(",") {
            json.remove(json.len() - 1);
        }
        json.push_str("]");
        json
    }

    fn serialize_timeout() -> String {
        String::from("\"timeout\": 0")
    }

    fn serialize_websockets() -> String {
        format!("\"websockets\": {}", CONFIG.websockets)
    }

    fn serialize_destinations() -> String {
        let mut json = String::from("\"destinations\": [");

        for destination in CONFIG.destinations.values() {
            json.push_str("{");
            json.push_str(format!("\"name\": \"{}\",", destination.name).as_str());
            json.push_str(format!("\"filter\": {},", destination.filter).as_str());
            // currently not supported in romp
            // json.push_str(format!("\"filter_header\": \"{}\",", destination.filter_header).as_str());
            json.push_str(format!("\"max_connections\": {},", destination.max_connections).as_str());
            json.push_str(format!("\"max_messages\": {},", destination.max_messages).as_str());
            json.push_str(format!("\"stats\": {},", destination.stats).as_str());
            json.push_str(format!("\"web_write_block\": {},", destination.web_write_block).as_str());
            json.push_str(format!("\"web_read_block\": {},", destination.web_read_block).as_str());
            json.push_str(format!("\"max_message_size\": {}", destination.max_message_size).as_str());
            json.push_str("},");
        }
        if json.ends_with(",") {
            json.remove(json.len() - 1);
        }
        json.push_str("]");
        json
    }

    // destination-info //

    fn serialize_destination_info(&self, idx: usize) -> String {
        if let Ok(destination_info) = self.ngin_info.destinations[idx].read() {
            return format!("{{\"sz\":{}, \"q\":{}, \"Δ\":{}, \"Σ\":{}}}",
                    destination_info.size,
                    destination_info.queue_len,
                    destination_info.delta,
                    destination_info.total);
        }
        String::from("{}")
    }
}


#[cfg(test)]
mod tests {

    use super::*;

    #[test]
    fn test() {
        println!("{}", Ngin::serialize_proc_pid());
        println!("{}", Ngin::serialize_proc_cmdline());
        println!("{}", Ngin::serialize_proc_sockstat());
        println!("{}", Ngin::serialize_proc_status());
    }

}