wd_pool 0.1.0

A pool crate for rust
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

mod pool_interface;
mod task_pool;
mod obj_pool;

pub use pool_interface::*;
pub use task_pool::*;
pub use obj_pool::*;

#[cfg(test)]
mod test{
    use std::time::Duration;
    use super::*;
    struct Pool;
    #[async_trait::async_trait]
    impl TaskPool for Pool{ type Out = (); }

    /// TaskPool 默认直接同步执行
    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
    async fn test_task_pool_default() {
        let str = "hello";
        Pool.push(async move {
            let str = format!("{} world", str);
            assert_eq!(str, "hello world".to_string(), "failed -> {}", str);
            println!("success--> test_task_pool_default");
        }).await.expect("test_task_pool_default");
    }

    /// TaskPoolContainer 限制任务的并行数，并提供backoff尝试策略，同步模式
    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
    async fn test_task_pool_entity() {
        let pool = TaskPoolContainer::<String>::default().build();
        let s = pool.push(async move {
            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
            println!("success--> test_task_pool_entity");
            return "hello world".to_string();
        }).await.expect("test_task_pool_entity");
        assert_eq!(s,"hello world".to_string(),"test_task_pool_entity test failed");
        let _ = pool.close(Duration::from_secs(1)).await;
    }

    /// TaskPoolChannel 限制并行数，并且由若干消费者消费，发射式调用
    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
    async fn test_task_pool_channel() {
        let str = "hello";
        let pool = TaskPoolChannel::default().set_cache_len(1).build();
        pool.push(async move {
            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
            let str = format!("{} world", str);
            assert_eq!(str, "hello world".to_string(), "failed -> {}", str);
            println!("success--> test_task_pool_channel");
        }).await.expect("test_task_pool_channel");
        println!("test start--> test_task_pool_channel");
        let _ = pool.close(Duration::from_secs(1)).await;
    }

    //------------------------》 对象池

    #[derive(Default)]
    struct ObjectImpl{
        name:String,
    }

    #[async_trait::async_trait]
    impl Object for ObjectImpl {
        async fn init(&mut self) {
            if self.name.is_empty() {
                self.name = String::from("hello world")
            }
        }
        async fn reset(&mut self) {
            self.name = String::from("world hello")
        }
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn test_base(){
        let pool = ObjPoolContainer::<ObjectImpl>::new().build();
        let obj = pool.get().await;
        println!("get success {}",obj.name);
        pool.release(obj).await;
        let obj = pool.get().await;
        println!("set success {}",obj.name);
    }
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn test_function(){
        let pool = ObjPoolContainer::<ObjectImpl>::new().set_pool_max(10);
        let result = pool.function(|x| {
            Box::pin(async move {
                (Some(x),1)
            })
        }).await;
        println!("result = {}",result);
    }
}