ek-station 0.4.20

//! 测试探针(每测试自有的宿主外观测面;零 dev-dep,全手写,与 M5 fixture 纪律同源)。
//!
//! - [`Probe`]:订阅给定 type 列表,把 event_type 全记进共享槽(ePlug*/ePlugDown.* 闭环)。
//! - [`Asker` + `Driver`]:收到 `eGo.` 触发一步 ask 脚本(步骤由 cursor 指定),
//!   回执序列化文本进共享队列 —— main 线程 `pop_output` 轮询(不忙等)。
//! - [`Stopper`]:on_stop 把注册名记进共享槽(close 顺序断言)。
//! - [`wait_until`]:限时轮询(超时 panic,毫秒步进)。

use std::borrow::Cow;
use std::collections::VecDeque;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::{Duration, Instant};

use ek_core::context::PluginContext;
use ek_core::plugin::Plugin;
use ek_event::Event;

/// 事件日志探针。
pub struct Probe {
    name: &'static str,
    types: Vec<String>,
    events: Arc<Mutex<Vec<String>>>,
}

impl Probe {
    /// 注册名 + 订阅 type 面 + 共享日志槽。
    pub fn with_types(name: &'static str, types: &[&str], events: Arc<Mutex<Vec<String>>>) -> Self {
        Self {
            name,
            types: types.iter().map(|t| (*t).to_string()).collect(),
            events,
        }
    }
}

impl Plugin for Probe {
    fn name(&self) -> &str {
        self.name
    }

    fn routes(&self) -> Vec<Cow<'static, str>> {
        self.types.iter().map(|t| Cow::Owned(t.clone())).collect()
    }

    fn on_mail(&self, event: &Event) {
        self.events.lock().unwrap().push(event.event_type.clone());
    }
}

/// ask 脚本的一步:执行体拿 `PluginContext`(插件 worker 线程),返回回执快照文本。
pub struct AskStep {
    pub exec: Box<dyn Fn(&PluginContext) -> String + Send + Sync>,
}

/// 步进驱动:cursor 当前步;`eGo.` 触发执行当前步并前移。
pub struct Driver {
    cursor: AtomicUsize,
    steps: Vec<AskStep>,
    outputs: Mutex<VecDeque<String>>,
}

impl Driver {
    /// 初始步 + 步表。
    pub fn with_steps(initial: usize, steps: Vec<AskStep>) -> Self {
        Self {
            cursor: AtomicUsize::new(initial),
            steps,
            outputs: Mutex::new(VecDeque::new()),
        }
    }

    /// 非阻塞取出一条回执快照(poll 循环里反复调)。
    pub fn pop_output(&self) -> Option<String> {
        self.outputs.lock().unwrap().pop_front()
    }

    /// 队列剩余快照(wait 谓词用:只读非消费)。
    pub fn outputs_left(&self) -> Vec<String> {
        self.outputs.lock().unwrap().iter().cloned().collect()
    }

    fn on_go(&self, ctx: &PluginContext) {
        let idx = self.cursor.fetch_add(1, Ordering::AcqRel);
        if let Some(step) = self.steps.get(idx) {
            let out = (step.exec)(ctx);
            self.outputs.lock().unwrap().push_back(out);
        }
    }
}

/// ask 驱动插件:收到 `eGo.` 走 driver 当前步。
pub struct Asker {
    name: &'static str,
    driver: Arc<Driver>,
    ctx: OnceLock<PluginContext>,
}

impl Asker {
    /// 注册名 + 驱动共享句柄。
    pub fn driven(name: &'static str, driver: Arc<Driver>) -> Self {
        Self {
            name,
            driver,
            ctx: OnceLock::new(),
        }
    }
}

impl Plugin for Asker {
    fn name(&self) -> &str {
        self.name
    }

    fn routes(&self) -> Vec<Cow<'static, str>> {
        vec![Cow::Borrowed("eGo.")]
    }

    fn on_init(&self, ctx: &PluginContext) {
        let _ = self.ctx.set(ctx.clone());
    }

    fn on_mail(&self, event: &Event) {
        if event.event_type != "eGo." {
            return;
        }
        if let Some(ctx) = self.ctx.get() {
            self.driver.on_go(ctx);
        }
    }
}

/// on_stop 记名插件(close 顺序断言用)。
pub struct Stopper {
    name: &'static str,
    deps: Vec<&'static str>,
    stops: Arc<Mutex<Vec<String>>>,
}

impl Stopper {
    /// 注册名 + 显式 deps + 落槽。
    pub fn with_deps(
        name: &'static str,
        deps: &[&'static str],
        stops: Arc<Mutex<Vec<String>>>,
    ) -> Self {
        Self {
            name,
            deps: deps.to_vec(),
            stops,
        }
    }
}

impl Plugin for Stopper {
    fn name(&self) -> &str {
        self.name
    }

    fn deps(&self) -> &[&str] {
        &self.deps
    }

    fn on_stop(&self) {
        self.stops.lock().unwrap().push(self.name.to_string());
    }

    fn on_mail(&self, _event: &Event) {} // 无业务面;纯 stop 探针
}

/// 限时轮询;超时 panic(带上下文标签)。
/// 纪律:谓词只读非消费(计数/存在检查),不调 `pop_output` 等"取走即消失"面。
pub fn wait_until(tag: &str, secs: u64, mut f: impl FnMut() -> bool) {
    let deadline = Instant::now() + Duration::from_secs(secs);
    while Instant::now() < deadline {
        if f() {
            return;
        }
        std::thread::sleep(Duration::from_millis(10));
    }
    panic!("wait_until timeout: {tag}");
}