notus-mempool 0.1.1

use log::{debug, error, info, trace};
use serde::{Deserialize, Serialize};
use std::{
    collections::HashMap,
    net::UdpSocket,
    sync::{Arc, Mutex},
    thread,
    time::{SystemTime, UNIX_EPOCH},
};

#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub struct MempoolMessageStruct {
    pub time: u128,
    pub data: Vec<u8>,
}
#[derive(Debug)]
pub struct Mempool {
    // mempool soket adresi
    pub addr: String,

    // node listesini tutan liste
    pub node_list: Arc<Mutex<Vec<String>>>,

    // mesajların sıralı olduğu liste ( hash )
    pub order_list: Arc<Mutex<Vec<String>>>,

    // mesajlara hash ile doğrudan erişilen liste ( key = hash )
    pub data_list: Arc<Mutex<HashMap<String, MempoolMessageStruct>>>,

    // diğer nodelara iletilecek liste ( key = hash )
    pub dist_list: Arc<Mutex<Vec<String>>>,

    // bana dooğrudan veya diğer nodelardan gelen mesajların iletim kontrol listesi
    // pub cont_list: Arc<Mutex<HashMap<String, bool>>>,
}

impl Mempool {
    pub fn new(node_addr:String) -> Self {
        Mempool {
            addr: node_addr.clone(),
            node_list: Arc::new(Mutex::new(Vec::new())),
            order_list: Arc::new(Mutex::new(Vec::new())),
            data_list: Arc::new(Mutex::new(HashMap::new())),
            dist_list: Arc::new(Mutex::new(Vec::new())),
            // cont_list: Arc::new(Mutex::new(HashMap::new())),
        }
    }
    pub fn add_msg(&mut self, raw_data: Vec<u8>) -> bool {
        let msg_time = Mempool::time_stamp();
        let msg_hash = format!("{:x}", md5::compute(&raw_data));
        self.order_list.lock().unwrap().push(msg_hash.clone());
        if self
            .data_list
            .lock()
            .unwrap()
            .contains_key(&msg_hash.clone())
            == false
        {
            self.data_list.lock().unwrap().insert(
                msg_hash.clone(),
                MempoolMessageStruct {
                    time: msg_time,
                    data: raw_data.clone(),
                },
            );
        }
        //self.cont_list.lock().unwrap().insert(msg_hash.clone(), false);
        self.dist_list.lock().unwrap().push(msg_hash.clone());
        true
    }
    pub fn read(&mut self) -> Option<MempoolMessageStruct> {
        loop {
            match self.order_list.lock().unwrap().clone().get(0) {
                Some(msg_data) => {
                    let timestamp = Mempool::time_stamp();
                    if self
                        .data_list
                        .lock()
                        .unwrap()
                        .contains_key(&msg_data.clone())
                    {
                        let inner_struct = self
                            .data_list
                            .lock()
                            .unwrap()
                            .get(&msg_data.clone())
                            .unwrap()
                            .clone();
                        let time_diff = timestamp - inner_struct.time;
                        if time_diff > 100000 {
                            self.order_list.lock().unwrap().remove(0);
                        } else {
                            return Some(inner_struct.clone());
                        }
                    } else {
                        error!("msg could not find > {}", msg_data);
                    }
                }
                None => {
                    return None;
                }
            }
        }
    }
    pub fn start(&self) {
        self.start_listener();
        self.start_distribution();
    }
    fn start_listener(&self) {
        let dist_list_cloned = self.dist_list.clone();
        let data_list_cloned = self.data_list.clone();
        let order_list_cloned = self.order_list.clone();
        let addr_cloned = self.addr.clone();
        let thread_obj = thread::spawn(move || {
            let socket_ret = UdpSocket::bind(addr_cloned.clone());
            if socket_ret.is_ok() {
                let socket = socket_ret.unwrap();
                info!("server listening on {} ", addr_cloned.clone());
                let mut buffer = [0; 1024];
                loop {
                    let recv_ret = socket.recv_from(&mut buffer);
                    if recv_ret.is_ok() {
                        let (size, source) = recv_ret.unwrap();
                        let decoded: Result<MempoolMessageStruct, Box<bincode::ErrorKind>> =
                        bincode::deserialize(&buffer[..size]);
                        if decoded.is_ok() {
                            let in_data = decoded.unwrap();
                            let msg_hash = format!("{:x}", md5::compute(&in_data.data));
                            if data_list_cloned.lock().unwrap().contains_key(&msg_hash.clone())== false
                            {
                                data_list_cloned.lock().unwrap().insert(
                                    msg_hash.clone(),
                                    in_data,
                                );
                                order_list_cloned.lock().unwrap().push(msg_hash.clone());
                                dist_list_cloned.lock().unwrap().push(msg_hash.clone());                                    
                            }
                            let request = format!("{}", String::from_utf8_lossy(&buffer[..size]));
                            debug!("Received request: {} from {}", request, source);
                        }else{
                            error!("wrong packet arrived > {} ", addr_cloned.clone());
                        }
                    }
                }
            } else {
                error!("socket could not start > {} ", addr_cloned.clone());
            }
        });
        thread::spawn(move || {
            let _res = thread_obj.join();
        });
    }
    fn start_distribution(&self) {
        let dist_list_cloned = self.dist_list.clone();
        let data_list_cloned = self.data_list.clone();
        let node_list_cloned = self.node_list.clone();
        //let cont_list_cloned = self.cont_list.clone();
        let thread_obj = thread::spawn(move || loop {
            if node_list_cloned.lock().unwrap().len() > 0
                && dist_list_cloned.lock().unwrap().len() > 0
            {
                let message_to_send = dist_list_cloned.lock().unwrap().get(0).unwrap().to_string();
                dist_list_cloned.lock().unwrap().remove(0);

                trace!("message to send > {}", message_to_send);
                if data_list_cloned
                    .lock()
                    .unwrap()
                    .contains_key(&message_to_send.clone())
                {
                    let inner_struct = data_list_cloned
                    .lock()
                    .unwrap()
                    .get(&message_to_send.clone())
                    .unwrap()
                    .clone();
                    //cont_list_cloned.lock().unwrap().insert(message_to_send.clone(),true);
                    for node_addr in node_list_cloned.lock().unwrap().iter() {
                        Mempool::send_via_udp(node_addr.to_string(), inner_struct.clone());
                    }
                } else {
                    error!("msg could not find > {}", message_to_send);
                }
            }
        });
        thread::spawn(move || {
            let _res = thread_obj.join();
        });
    }
    pub fn send_via_udp(receiver_addr: String, data: MempoolMessageStruct) {
        thread::spawn(move || {
            let socket = UdpSocket::bind("127.0.0.1:0");
            if socket.is_ok() {
                let encoded: Vec<u8> = bincode::serialize(&data).unwrap();
                let socket = socket.unwrap();
                let sended = socket.send_to(&encoded, receiver_addr);
                if sended.is_ok() {
                    trace!("msg sended");
                } else {
                    error!("msg sending error");
                }
            } else {
                error!("socket openning error");
            }
        });
    }
    pub fn time_stamp() -> u128 {
        match SystemTime::now().duration_since(UNIX_EPOCH) {
            Ok(time) => time.as_micros(),
            Err(_) => 0,
        }
    }
    pub fn add_nodes(&mut self, node_list: Vec<String>) {
        for node in node_list.iter() {
            self.add_node(String::from(node));
        }
    }
    pub fn add_node(&mut self, node_addr: String) -> bool {
        for item in self.node_list.lock().unwrap().iter() {
            if item.eq(&node_addr) {
                return true;
            }
        }
        self.node_list.lock().unwrap().push(node_addr);
        return true;
    }
    pub fn remove_node(&mut self, node_addr: String) {
        let mut founded = false;
        let mut founded_index = 0;
        for (index, item) in self.node_list.lock().unwrap().iter().enumerate() {
            if item.eq(&node_addr) {
                founded = true;
                founded_index = index;
                break;
            }
        }
        if founded == true {
            self.node_list.lock().unwrap().remove(founded_index);
        }
    }
}

#[test]
fn full_test() {
    // cargo test  --lib full_test -- --nocapture
    /*
    let mut storage_obj=Storage::new("ornek");
    let result1=storage_obj.add_file("file1.txt");
    if result1==true{
        println!("burada");
        let result2=storage_obj.add_file_with_name("file2.txt","farkli_isim.md");
        if result2==true{
            assert!(true)
        }else {
            assert!(false)
        }
        assert!(true)
    }else {
        assert!(false)
    }
    */
    assert!(true)
}