ek_core/
context.rs

1//! `PluginContext` —— 插件的唯一出口(put / ask / notify / write_log)。
2//!
3//! 纪律(P-H2 审核修订):不再暴露 event_engine() 访问器——插件不许自装 handler
4//! 绕过 worker/熔断/背压;收信口(on_mail)与发信口(ctx)各只有一个。
5
6use crate::envelope::{e_req_type, e_to_type, Envelope, OneshotErr, ReplyErr};
7use crate::error::AskError;
8use crate::log_plugin::{LogData, EVENT_LOG};
9use crate::main_engine::PluginState;
10use crate::registry::Registry;
11use ek_event::{Event, EventEngine, PutError};
12use std::any::Any;
13use std::sync::Arc;
14use std::time::Duration;
15
16/// 发放到插件的上下文(on_init 传入;`Clone` 供插件自存)。
17#[derive(Clone)]
18pub struct PluginContext {
19    pub(crate) event: Arc<EventEngine>,
20    pub(crate) registry: Arc<Registry>,
21    pub(crate) name: String,
22}
23
24impl PluginContext {
25    pub(crate) fn new(event: Arc<EventEngine>, registry: Arc<Registry>, name: String) -> Self {
26        Self { event, registry, name }
27    }
28
29    /// 本插件注册名。
30    pub fn name(&self) -> &str {
31        &self.name
32    }
33
34    /// 广播出口:插件自发事件(业务 broadcast / eTo / eLog 均经此)。
35    pub fn put(&self, event: Event) -> Result<(), PutError> {
36        self.event.put(event)
37    }
38
39    /// 日志出口(eLog → 内置 LogPlugin)。
40    pub fn write_log(&self, level: u8, msg: &str) {
41        let data = LogData {
42            level,
43            source: self.name.clone(),
44            msg: msg.to_string(),
45        };
46        let _ = self.event.put(Event::with_data(EVENT_LOG, Arc::new(data)));
47    }
48
49    /// 定向通知出口:eTo.<target>,无回包;目标不在线时事件被静默分发为空(尽力而为)。
50    pub fn notify(&self, target: &str, payload: Arc<dyn Any + Send + Sync>) {
51        let _ = self.event.put(Event::with_data(e_to_type(target), payload));
52    }
53
54    /// 请求出口:eReq.<target> + Envelope{payload, reply} + std mpsc 一次性信道。
55    ///
56    /// **禁止自呼**:ask 指向自身 target 会占满自己的 worker 直到超时
57    /// (自己阻塞等自己的回包,worker 无法转去处理请求)——需要自身逻辑请直调。
58    ///
59    /// 时序:预检目标 Running → put → recv_timeout。
60    /// - 预检失败(不存在/未 Running/已 Quarantined)→ 立即 `Err(TargetGone)`;
61    /// - 回包 `Ok(Res)` 且契约类型匹配 → `Ok(Res)`;
62    /// - 回包 `Err(InvalidPayload)` **或** 回包类型与本侧契约不一致 → `Err(InvalidPayload)`;
63    /// - 信道断连(handler panic/信封丢失/目标邮箱溢出被丢)→ `Err(TargetGone)`(尽快,不虚耗超时);
64    ///   注:目标邮箱溢出时 forwarder 弃整个请求事件,信封随之 drop → 回包信道立即关闭,
65    ///   发起方走 TargetGone 快失败;同时宿主侧有溢出丢弃 eLog 节流告警(可追溯);
66    /// - 等满时限 → `Err(Timeout)`;
67    /// - put 失败(总线已停)→ `Err(BusStopped)`。
68    ///
69    /// TargetGone 的即时性为「尽力而为」:预检 + 断连快失败覆盖绝大多数场合;
70    /// 极端 TOCTOU 窗口(预检后瞬间目标下线且断连未被感知)由 Timeout 兜底。
71    pub fn ask<Req, Res>(&self, target: &str, req: Req, timeout: Duration) -> Result<Res, AskError>
72    where
73        Req: Any + Send + Sync,
74        Res: Any + Send + Sync,
75    {
76        match self.registry.get(target) {
77            Some(entry) if entry.state.get() == PluginState::Running => {}
78            _ => return Err(AskError::TargetGone { target: target.to_string() }),
79        }
80        let (envelope, reply_rx) = Envelope::new(Arc::new(req));
81        let put = self.event.put(Event::with_data(e_req_type(target), Arc::new(envelope)));
82        if put.is_err() {
83            return Err(AskError::BusStopped);
84        }
85        let boxed = reply_rx.recv_timeout(timeout).map_err(|e| match e {
86            OneshotErr::Timeout => AskError::Timeout { target: target.to_string() },
87            OneshotErr::Closed => AskError::TargetGone { target: target.to_string() },
88        })?;
89        // 回包为 Box<dyn Any>: downcast 到本侧契约 Result<Res, ReplyErr>
90        match boxed.downcast::<Result<Res, ReplyErr>>() {
91            Ok(inner) => match *inner {
92                Ok(res) => Ok(res),
93                Err(_) => Err(AskError::InvalidPayload { target: target.to_string() }),
94            },
95            // 回包类型与本侧契约不同(双方契约错配)→ 协议违约
96            Err(_) => Err(AskError::InvalidPayload { target: target.to_string() }),
97        }
98    }
99}
ek_core/context.rs

ek_core/
context.rs
