wd_pool 0.1.0

use std::future::Future;
use std::marker::PhantomData;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicI32, Ordering};
use std::time::Duration;
use crate::TaskPool;

pub struct TaskPoolContainer<T>{
    status: Arc<AtomicBool>,
    max_wait:i32,
    max:i32,
    active:Arc<AtomicI32>,
    _t:PhantomData<T>,
}

impl<T:Send+Sync + 'static> TaskPoolContainer<T> {
    pub fn set_max_wait(mut self,max_wait:i32)->Self{
        self.max_wait = max_wait;self
    }
    pub fn set_max_parallel(mut self,max:i32)->Self{
        self.max = max;self
    }
    pub fn build(self)-> impl TaskPool<Out=T> {
        self
    }
    fn backoff(&self,i:i32) -> u64 {
        let next_wait_time = i*i;
        if next_wait_time < self.max_wait{
            return next_wait_time as u64
        }
        return self.max_wait as u64
    }
    fn active_reset(&self){
        self.active.store(0,Ordering::Relaxed);
    }
}

impl<T> Default for TaskPoolContainer<T> {
    fn default() -> Self {
        Self{
            status: Arc::new(AtomicBool::new(true)),
            max_wait: 60,
            max: i32::MAX,
            active: Arc::new(Default::default()),
            _t:PhantomData::default(),
        }
    }

}

impl<T> Drop for TaskPoolContainer<T> {
    fn drop(&mut self) {
        self.status.store(false,Ordering::Relaxed);
        // self.active_reset();
        self.active.store(0,Ordering::Relaxed);
    }
}

#[async_trait::async_trait]
impl<T:Send+Sync + 'static> TaskPool for TaskPoolContainer<T> {
    type Out = T;

    async fn push<F: Future<Output=Self::Out> + Send + 'static>(&self, task: F) ->anyhow::Result<Self::Out> {
        let i = 1i32;
        while self.active.load(Ordering::Relaxed) >= self.max {
            if !self.status.load(Ordering::Relaxed) {
                return Err(anyhow::anyhow!("TaskPoolContainer is not normal"))
            }
            tokio::time::sleep(Duration::from_secs(self.backoff(i))).await; //backoff 策略
        }
        if !self.status.load(Ordering::Relaxed) {
            return Err(anyhow::anyhow!("TaskPoolContainer is not normal"))
        }
        self.active.fetch_add(1,Ordering::Relaxed);
        let result = tokio::spawn(task).await?;
        self.active.fetch_sub(1,Ordering::Relaxed);
        return Ok(result);
    }
    async fn close(&self, timeout: Duration) -> anyhow::Result<()> {
        self.status.store(false,Ordering::Relaxed);
        let active = self.active.clone();
        tokio::time::timeout(timeout,async move{
            while active.load(Ordering::Relaxed) > 0 {
                tokio::time::sleep(Duration::from_secs(1)).await;
            }
        }).await?;Ok(())
    }
}