ek-core 0.4.3

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

use crate::registry::EntryStats;
use ek_event::{Event, HandlerFn};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};

use crossbeam_channel::{Sender, TrySendError};

/// 溢出告警节流:累计溢出每越过这么多条,报一条(stderr;不进总线)。
pub const SPILL_REPORT_EVERY: u64 = 128;

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

impl Forwarder {
    pub(crate) fn new(tx: Sender<Event>, stats: Arc<EntryStats>, target: String) -> Arc<Self> {
        Arc::new(Self {
            tx,
            alive: AtomicBool::new(true),
            stats,
            reported: AtomicU64::new(0),
            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()
        {
            // 0.3.0:溢出诊断走 stderr(节流);结构化事实是 stats.spilled,查 mail_stats()
            eprintln!(
                "ek-core: mailbox overflow: plugin {:?} spilled {} events in total",
                self.target, total
            );
        }
    }
}

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