ek_core/
context.rs

1//! `PluginContext` —— 插件的唯一出口(put / ask / notify / 发现四问)。
2//!
3//! 纪律(P-H2 审核修订):不暴露 event_engine() 访问器——插件不许自装 handler
4//! 绕过 worker/熔断/背压;收信口(on_mail)与发信口(ctx)各只有一个。
5//! 0.3.0:宿主零日志 —— write_log 出口移除,业务日志由插件自造自投
6//! (`ctx.put` 自己的载荷,业务链上用 ek-data 的 LogData 到 "eLog" 键)。
7//!
8//! 0.2.0 新增(宿主最小知识原则):plugins / services_of / find_service /
9//! ask_service / publish_services —— 宿主只持【纯路由索引】,
10//! 方法细节住回插件本身,按需现场取(memoize 仅为性能缓存)。
11
12use crate::envelope::{e_req_type, e_to_type, Envelope, OneshotErr, ReplyErr};
13use crate::error::AskError;
14use crate::lifecycle::{EVENT_SRV_CHG, EVENT_SRV_CHG_PREFIX};
15use crate::main_engine::{PluginInfo, PluginState};
16use crate::registry::Registry;
17use crate::service::{ProviderRef, ServiceDesc};
18use ek_event::{Event, EventEngine, PutError};
19use std::any::{Any, type_name};
20use std::sync::Arc;
21use std::time::Duration;
22
23/// 发放到插件的上下文(on_init 传入;`Clone` 供插件自存)。
24#[derive(Clone)]
25pub struct PluginContext {
26    pub(crate) event: Arc<EventEngine>,
27    pub(crate) registry: Arc<Registry>,
28    pub(crate) name: String,
29}
30
31impl PluginContext {
32    pub(crate) fn new(event: Arc<EventEngine>, registry: Arc<Registry>, name: String) -> Self {
33        Self { event, registry, name }
34    }
35
36    /// 本插件注册名。
37    pub fn name(&self) -> &str {
38        &self.name
39    }
40
41    /// 广播出口:插件自发事件(业务 broadcast / 业务日志 / eTo 均经此)。
42    ///
43    /// 日志纪律(0.3.0):宿主不提供日志载荷类型;要写业务日志,插件自造
44    /// 载荷自理(业务链上用 ek-data 的 LogData 到 "eLog" 键后 put)。
45    pub fn put(&self, event: Event) -> Result<(), PutError> {
46        self.event.put(event)
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    // ================ 服务发现四问(0.2.0) ================
55
56    /// 枚举宿主内全部插件(含 Ready;快照含版本)。
57    pub fn plugins(&self) -> Vec<PluginInfo> {
58        let mut out = Vec::new();
59        for name in self.registry.names() {
60            if let Some(info) = self.registry.info_of(&name) {
61                out.push(info);
62            }
63        }
64        out
65    }
66
67    /// 某插件自报的完整服务目录(现场取 + 世代核对 + memoize)。
68    /// Ready 未 Running 亦可查(声明在注册时即静态可用)。
69    /// 该插件不存在 → None。
70    pub fn services_of(&self, name: &str) -> Option<Arc<Vec<ServiceDesc>>> {
71        self.registry.services_of(name)
72    }
73
74    /// 按【服务 id】全宿主查找提供方(有序:注册序;含非 Running 状态 → 供消费方自裁)。
75    /// A 不需要预先知道提供方的注册名。
76    pub fn find_service(&self, svc: &str) -> Vec<ProviderRef> {
77        let mut out = Vec::new();
78        for name in self.registry.index_providers(svc) {
79            let Some(list) = self.registry.services_of(&name) else { continue };
80            let Some(entry) = self.registry.get(&name) else { continue };
81            if let Some(desc) = list.iter().find(|d| d.id == svc) {
82                out.push(ProviderRef {
83                    name: entry.name.clone(),
84                    version: entry.plugin.version().into_owned(),
85                    state: entry.state.get(),
86                    service: desc.clone(),
87                });
88            }
89        }
90        out
91    }
92
93    /// 运行期重新申报自己的服务目录:宿主重织索引、作废缓存,双发 `eSrvChg.<名>`(版本串)
94    /// 与 `eSrvChg`(固定键,PlugNotice)。
95    ///
96    /// ⚠ 纪律(DV4 教训):**先换插件 services() 当前返回,再调用本方法** ——
97    /// 宿主细节是现场取的,只申报不改内容会造成索引与 live 细节脱节。
98    ///
99    /// (0.2.0:Rust 无运行期代码热换;「更新」的运行期形态 = 目录随配置/状态改版。)
100    pub fn publish_services(&self, list: Vec<ServiceDesc>) {
101        let name = self.name.clone();
102        let version = self.registry.publish(&name, list);
103        let _ = self
104            .event
105            .put(Event::with_data(format!("{EVENT_SRV_CHG_PREFIX}{name}"), Arc::new(version.clone())));
106        let _ = self
107            .event
108            .put(Event::with_data(EVENT_SRV_CHG, Arc::new(crate::service::PlugNotice { name, version })));
109    }
110
111    // ================ 请求应答 ================
112
113    /// 请求出口:eReq.<target> + Envelope + std mpsc 一次性信道。
114    ///
115    /// **禁止自呼**:ask 指向自身 target 会占满自己的 worker 直到超时
116    /// (自己阻塞等自己的回包)——需要自身逻辑请直调。
117    ///
118    /// 时序:预检目标 Running → put → recv_timeout。
119    /// - 预检失败(不存在/未 Running/已 Quarantined)→ 立即 `Err(TargetGone)`;
120    /// - 回包 `Ok(Res)` 且契约类型匹配 → `Ok(Res)`;
121    /// - 回包 `Err(InvalidPayload)` **或** 回包类型与本侧契约不一致 → `Err(InvalidPayload)`;
122    /// - 信道断连(handler panic/信封丢失/目标邮箱溢出被丢)→ `Err(TargetGone)`(尽快,不虚耗超时);
123    /// - 等满时限 → `Err(Timeout)`;
124    /// - put 失败(总线已停)→ `Err(BusStopped)`。
125    pub fn ask<Req, Res>(&self, target: &str, req: Req, timeout: Duration) -> Result<Res, AskError>
126    where
127        Req: Any + Send + Sync,
128        Res: Any + Send + Sync,
129    {
130        match self.registry.get(target) {
131            Some(entry) if entry.state.get() == PluginState::Running => {}
132            _ => return Err(AskError::TargetGone { target: target.to_string() }),
133        }
134        self.deliver(target, req, timeout)
135    }
136
137    /// 按服务 id 的请求出口(0.2.0):index 命中 → 现场类型闸门 → 发 eReq。
138    ///
139    /// - 无人提供该服务 → 立即 `Err(UnknownService)`(零成本显错);
140    /// - 有提供方但全都非 Running → 立即 `Err(TargetGone)`;
141    /// - Req 类型与服务声明的 req_type 不符 → 立即 `Err(InvalidPayload)`
142    ///   (把「绕一圈才发现错型」提前到发出之前);
143    /// - 多提供方 → 取注册序第一个 Running 者;后续 failover 由发起方在收不到回包后自定。
144    ///   其余失败面与 [`ask`](Self::ask) 完全一致。
145    pub fn ask_service<Req, Res>(&self, svc: &str, req: Req, timeout: Duration) -> Result<Res, AskError>
146    where
147        Req: Any + Send + Sync,
148        Res: Any + Send + Sync,
149    {
150        let providers = self.find_service(svc);
151        if providers.is_empty() {
152            return Err(AskError::UnknownService { service: svc.to_string() });
153        }
154        let Some(slot) = providers.iter().find(|p| p.state == PluginState::Running) else {
155            return Err(AskError::TargetGone { target: providers[0].name.clone() });
156        };
157        if slot.service.req_type != type_name::<Req>() {
158            return Err(AskError::InvalidPayload { target: slot.name.clone() });
159        }
160        self.deliver(&slot.name.clone(), req, timeout)
161    }
162
163    /// ask/ask_service 共用的投递段(Envelope + eReq + recv_timeout + 回包裁决)。
164    fn deliver<Req, Res>(&self, target: &str, req: Req, timeout: Duration) -> Result<Res, AskError>
165    where
166        Req: Any + Send + Sync,
167        Res: Any + Send + Sync,
168    {
169        let (envelope, reply_rx) = Envelope::new(Arc::new(req));
170        let put = self.event.put(Event::with_data(e_req_type(target), Arc::new(envelope)));
171        if put.is_err() {
172            return Err(AskError::BusStopped);
173        }
174        let boxed = reply_rx.recv_timeout(timeout).map_err(|e| match e {
175            OneshotErr::Timeout => AskError::Timeout { target: target.to_string() },
176            OneshotErr::Closed => AskError::TargetGone { target: target.to_string() },
177        })?;
178        // 回包为 Box<dyn Any>:downcast 到本侧契约 Result<Res, ReplyErr>
179        match boxed.downcast::<Result<Res, ReplyErr>>() {
180            Ok(inner) => match *inner {
181                Ok(res) => Ok(res),
182                Err(_) => Err(AskError::InvalidPayload { target: target.to_string() }),
183            },
184            // 回包类型与本侧契约不同(双方契约错配)→ 协议违约
185            Err(_) => Err(AskError::InvalidPayload { target: target.to_string() }),
186        }
187    }
188}
ek_core/context.rs

ek_core/
context.rs
