roplat 0.2.0

use std::future::Future;
use std::sync::Arc;
/// 次数节律源实现：循环执行一定次数，次数由创建过程中指定，可以通过内置函数 set_times 修改
use std::sync::atomic::{AtomicUsize, Ordering};

use crate::rhythm::Rhythm;

/// 控制端：提供 set_times 接口来动态修改执行次�?
#[derive(Debug, Clone)]
pub struct CountHandle {
    shared_target: Arc<AtomicUsize>,
}

impl CountHandle {
    /// 设置新的目标执行次数
    ///
    /// # 参数
    /// * `times` - 新的目标执行次数
    pub fn set_times(&self, times: usize) {
        self.shared_target.store(times, Ordering::Relaxed);
    }

    /// 获取当前设置的目标次�?
    pub fn get_times(&self) -> usize {
        self.shared_target.load(Ordering::Relaxed)
    }
}

/// 执行端：实现 Rhythm trait，运行固定次数的循环
pub struct CountRhythm {
    shared_target: Arc<AtomicUsize>,
}

impl CountRhythm {
    /// 创建新的 CountRhythm 实例
    ///
    /// # 参数
    /// * `initial_times` - 初始目标执行次数
    pub fn new(initial_times: usize) -> Self {
        Self { shared_target: Arc::new(AtomicUsize::new(initial_times)) }
    }

    /// �?launch 参数构�?
    ///
    /// YAML 示例:
    /// ```yaml
    /// count: 10
    /// ```
    pub fn from_launch_params(params: &serde_yaml::Value) -> Self {
        let count = params.get("count").and_then(|v| v.as_u64()).unwrap_or(1) as usize;
        Self::new(count)
    }
}

/// 构造器：同时创建控制端和执行端
///
/// # 参数
/// * `initial_times` - 初始目标执行次数
///
/// # 返回
/// 返回一个元�?(CountHandle, CountRhythm)
pub fn create_count_rhythm(initial_times: usize) -> (CountHandle, CountRhythm) {
    let rhythm = CountRhythm::new(initial_times);
    let handle = CountHandle { shared_target: rhythm.shared_target.clone() };
    (handle, rhythm)
}

impl Rhythm for CountRhythm {
    type Yield = usize;
    type Feed = ();
    type Output = ();
    type Error = crate::RoplatError;

    async fn drive<N, F, Fut>(
        &mut self,
        mut nodes: N,
        mut op_domain: F,
    ) -> (Self::Output, N)
    where
        N: Send,
        F: FnMut(N, Self::Yield) -> Fut + Send,
        Fut: Future<Output = (Self::Feed, N)> + Send,
    {
        let mut current_seq = 0;

        loop {
            let target = self.shared_target.load(Ordering::Relaxed);
            if current_seq >= target {
                break;
            }
            let ((), returned_nodes) = op_domain(nodes, current_seq).await;
            nodes = returned_nodes;
            current_seq += 1;
            tokio::task::yield_now().await;
        }
        ((), nodes)
    }
}