ek-core 0.1.0

//! forwarder(转发器):宿主为插件每条 route 注册的 handler 闭包,
//! 只做「总线事件 → 插件 bounded 邮箱 try_send」,零业务、零阻塞。
//!
//! 溢出:丢弃计数(EntryStats.spilled),每 SPILL_REPORT_EVERY 次发一条节流 eLog;
//! 断连(worker 自退/邮箱关闭):alive 置 false 后短路,不再白调 try_send,
//! 正式注销留给 close 第①步(不在 handler 内 unregister,避免逆序复杂度)。

use crate::log_plugin::{LogData, EVENT_LOG, LOG_LEVEL_WARN};
use crate::registry::EntryStats;
use ek_event::{Event, EventEngine, HandlerFn};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;

use crossbeam_channel::{Sender, TrySendError};

/// 溢出告警节流:累计溢出每越过这么多条,报一条 eLog。
pub const SPILL_REPORT_EVERY: u64 = 128;

pub(crate) struct Forwarder {
    tx: Sender<Event>,
    alive: AtomicBool,
    stats: Arc<EntryStats>,
    reported: AtomicU64,
    bus: Arc<EventEngine>,
    target: String,
}

impl Forwarder {
    pub(crate) fn new(
        tx: Sender<Event>,
        stats: Arc<EntryStats>,
        bus: Arc<EventEngine>,
        target: String,
    ) -> Arc<Self> {
        Arc::new(Self {
            tx,
            alive: AtomicBool::new(true),
            stats,
            reported: AtomicU64::new(0),
            bus,
            target,
        })
    }

    /// 分发线程上执行;永不阻塞。
    pub fn forward(&self, event: &Event) {
        if !self.alive.load(Ordering::Acquire) {
            return;
        }
        // ek-event 0.1.0 的 Event 不实现 Clone:手工复制(data 走 Arc 引用计数,载荷零拷贝)
        let event2 = Event::with_data(event.event_type.clone(), Arc::clone(&event.data));
        match self.tx.try_send(event2) {
            Ok(()) => {}
            Err(TrySendError::Full(_)) => self.on_spill(),
            // worker 自退(rx drop)或邮箱关闭(tx 消费尽):短路口
            Err(TrySendError::Disconnected(_)) => {
                self.alive.store(false, Ordering::Release);
            }
        }
    }

    fn on_spill(&self) {
        let total = self.stats.spilled.fetch_add(1, Ordering::AcqRel) + 1;
        let last = self.reported.load(Ordering::Acquire);
        if total >= last + SPILL_REPORT_EVERY
            && self
                .reported
                .compare_exchange(last, total, Ordering::AcqRel, Ordering::Acquire)
                .is_ok()
        {
            let data = LogData {
                level: LOG_LEVEL_WARN,
                source: "ek-core".to_string(),
                msg: format!(
                    "mailbox overflow: plugin {:?} spilled {} events in total",
                    self.target, total
                ),
            };
            let _ = self.bus.put(Event::with_data(EVENT_LOG, Arc::new(data)));
        }
    }
}

/// 把 Forwarder 包成 ek-event 的 handler。
pub(crate) fn make_forwarder(fwd: Arc<Forwarder>) -> HandlerFn {
    Arc::new(move |event: &Event| fwd.forward(event))
}