wd_balancing 0.0.4

mod entity;
mod interface;
mod polling;
mod random;
mod minconn;

pub use interface::{BalancingCall,Linker,BalancingStrategy};
pub use polling::Polling;
pub use entity::Balancing;
pub use random::Random;
pub use minconn::MinConn;

#[cfg(test)]
mod test{
    use crate::{BalancingCall, Balancing, Polling, Random, MinConn};
    use rand::Rng;
    use std::sync::Arc;
    use tokio::sync::Mutex;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::ops::DerefMut;

    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn test_main(){
        let t = std::time::Instant::now();
        let rng = Arc::new(Mutex::new(rand::rngs::OsRng::default()));
        let threads = Arc::new(AtomicUsize::new(100));
        for _ in 0..100 {
            let threads = threads.clone();
            let rng = rng.clone();
            tokio::spawn(async{
                let rng = rng;
                let threads = threads;
                for _ in 0..1000 {
                   let mut rng = rng.lock().await;
                   let rng =  rng.deref_mut();
                    let _res = rng.gen_range(0..1000);
                }
                threads.fetch_sub(1,Ordering::Relaxed);
            });
        }
        loop {
            tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
            if threads.load(Ordering::Relaxed) == 0{
                break
            }
        }
        let res = t.elapsed().as_millis();
        let res = (res as f64) / 1000.0;
        wd_log::log_debug_ln!("四个线程生成一百万个随机数需要：{}s",res)
    }
    struct Service {
        name:&'static str
    }
    impl Service{
        fn new(name:&'static str)->Self{
            Self{name}
        }
    }
    #[async_trait::async_trait]
    impl BalancingCall<String, String> for Service{
        async fn call(&self, reqs: String) -> Option<String> {
            tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
            Some(format!("request:{} -> response:{}",reqs,self.name))
        }
    }
    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
    async fn test_polling(){
        wd_log::log_debug_ln!("start test polling");
        let bl = Balancing::new()
            .set_strategy(Polling::new().breadth_first());

        bl.add(1,Service::new("我是第一个节点"),1).await;
        bl.add(2,Service::new("我是第二个节点"),1).await;
        bl.add(3,Service::new("我是第三个节点"),1).await;

        for i in (0..9).map(|x|x as usize) {
            let resp = bl.call(i.to_string()).await;
            if let Some(s) = resp{
                wd_log::log_debug_ln!("-> {}",s)
            }
        }
        wd_log::log_debug_ln!("===================> 移除2号节点===========================");
        for i in (0..9).map(|x|x as usize) {
            let resp = bl.call(i.to_string()).await;
            if let Some(s) = resp{
                wd_log::log_debug_ln!("-> {}",s)
            }
        }
        bl.remove(1).await;
        bl.remove(3).await;
        wd_log::log_debug_ln!("===================> 移除全部节点===========================");
        for i in (0..3).map(|x|x as usize) {
            let resp = bl.call(i.to_string()).await;
            if let Some(s) = resp{
                wd_log::log_debug_ln!("-> {}",s);
            }else{
                wd_log::log_debug_ln!("-> 当前没有存活的节点");
            }
        }
    }
    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
    async fn test_random(){
        wd_log::log_debug_ln!("start test random");
        let bl = Balancing::new()
            .set_strategy(Random::new());

        bl.add(1,Service::new("我是第一个节点"),3).await;
        bl.add(2,Service::new("我是第二个节点"),2).await;
        bl.add(3,Service::new("我是第三个节点"),1).await;

        for i in (0..18).map(|x|x as usize) {
            let resp = bl.call(i.to_string()).await;
            if let Some(s) = resp{
                wd_log::log_debug_ln!("-> {}",s)
            }
        }
        bl.remove(2).await;
        wd_log::log_debug_ln!("===================> 移除2号节点===========================");
        for i in (0..9).map(|x|x as usize) {
            let resp = bl.call(i.to_string()).await;
            if let Some(s) = resp{
                wd_log::log_debug_ln!("-> {}",s)
            }
        }
        bl.remove(1).await;
        bl.remove(3).await;
        wd_log::log_debug_ln!("===================> 移除全部节点===========================");
        for i in (0..3).map(|x|x as usize) {
            let resp = bl.call(i.to_string()).await;
            if let Some(s) = resp{
                wd_log::log_debug_ln!("-> {}",s);
            }else{
                wd_log::log_debug_ln!("-> 当前没有存活的节点");
            }
        }
    }
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn test_minconn(){
        wd_log::log_debug_ln!("start test minconn");
        let bl = Balancing::new()
            .set_strategy_linker(MinConn::new());
        let bl = Arc::new(bl);

        bl.add(1,Service::new("我是第一个节点"),3).await;
        bl.add(2,Service::new("我是第二个节点"),2).await;
        bl.add(3,Service::new("我是第三个节点"),1).await;

        for i in (0..18).map(|x|x as usize) {
            let bbl = bl.clone();
            tokio::spawn(async move{
                let resp = bbl.call(i.to_string()).await;
                if let Some(s) = resp{
                    wd_log::log_debug_ln!("-> {}",s)
                }
            });
        }
        tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;
        bl.remove(2).await;
        wd_log::log_debug_ln!("===================> 移除2号节点===========================");
        for i in (0..9).map(|x|x as usize) {
            let resp = bl.call(i.to_string()).await;
            if let Some(s) = resp{
                wd_log::log_debug_ln!("-> {}",s)
            }
        }
        bl.remove(1).await;
        bl.remove(3).await;
        wd_log::log_debug_ln!("===================> 移除全部节点===========================");
        for i in (0..3).map(|x|x as usize) {
            let resp = bl.call(i.to_string()).await;
            if let Some(s) = resp{
                wd_log::log_debug_ln!("-> {}",s);
            }else{
                wd_log::log_debug_ln!("-> 当前没有存活的节点");
            }
        }
    }
}