ek_core/

context.rs

//! `PluginContext` —— 插件的唯一出口(put / ask / notify / 发现四问)。
//!
//! 纪律(P-H2 审核修订):不暴露 event_engine() 访问器——插件不许自装 handler
//! 绕过 worker/熔断/背压;收信口(on_mail)与发信口(ctx)各只有一个。
//! 0.3.0:宿主零日志 —— write_log 出口移除,业务日志由插件自造自投
//! (`ctx.put` 自己的载荷,业务链上用 ek-data 的 LogData 到 "eLog" 键)。
//!
//! 0.2.0 新增(宿主最小知识原则):plugins / services_of / find_service /
//! ask_service / publish_services —— 宿主只持【纯路由索引】,
//! 方法细节住回插件本身,按需现场取(memoize 仅为性能缓存)。

use crate::envelope::{Envelope, OneshotErr, ReplyErr, e_req_type, e_to_type};
use crate::error::AskError;
use crate::lifecycle::{EVENT_SRV_CHG, EVENT_SRV_CHG_PREFIX};
use crate::main_engine::{PluginInfo, PluginState};
use crate::registry::Registry;
use crate::service::{ProviderRef, ServiceDesc};
use ek_event::{Event, EventEngine, PutError};
use std::any::{Any, type_name};
use std::sync::Arc;
use std::time::Duration;

/// 发放到插件的上下文(on_init 传入;`Clone` 供插件自存)。
#[derive(Clone)]
pub struct PluginContext {
    pub(crate) event: Arc<EventEngine>,
    pub(crate) registry: Arc<Registry>,
    pub(crate) name: String,
}

impl PluginContext {
    pub(crate) fn new(event: Arc<EventEngine>, registry: Arc<Registry>, name: String) -> Self {
        Self {
            event,
            registry,
            name,
        }
    }

    /// 本插件注册名。
    pub fn name(&self) -> &str {
        &self.name
    }

    /// 广播出口:插件自发事件(业务 broadcast / 业务日志 / eTo 均经此)。
    ///
    /// 日志纪律(0.3.0):宿主不提供日志载荷类型;要写业务日志,插件自造
    /// 载荷自理(业务链上用 ek-data 的 LogData 到 "eLog" 键后 put)。
    pub fn put(&self, event: Event) -> Result<(), PutError> {
        self.event.put(event)
    }

    /// 定向通知出口:eTo.<target>,无回包;目标不在线时事件被静默分发为空(尽力而为)。
    pub fn notify(&self, target: &str, payload: Arc<dyn Any + Send + Sync>) {
        let _ = self.event.put(Event::with_data(e_to_type(target), payload));
    }

    // ================ 服务发现四问(0.2.0) ================

    /// 枚举宿主内全部插件(含 Ready;快照含版本)。
    pub fn plugins(&self) -> Vec<PluginInfo> {
        let mut out = Vec::new();
        for name in self.registry.names() {
            if let Some(info) = self.registry.info_of(&name) {
                out.push(info);
            }
        }
        out
    }

    /// 某插件自报的完整服务目录(现场取 + 世代核对 + memoize)。
    /// Ready 未 Running 亦可查(声明在注册时即静态可用)。
    /// 该插件不存在 → None。
    pub fn services_of(&self, name: &str) -> Option<Arc<Vec<ServiceDesc>>> {
        self.registry.services_of(name)
    }

    /// 按【服务 id】全宿主查找提供方(有序:注册序;含非 Running 状态 → 供消费方自裁)。
    /// A 不需要预先知道提供方的注册名。
    pub fn find_service(&self, svc: &str) -> Vec<ProviderRef> {
        let mut out = Vec::new();
        for name in self.registry.index_providers(svc) {
            let Some(list) = self.registry.services_of(&name) else {
                continue;
            };
            let Some(entry) = self.registry.get(&name) else {
                continue;
            };
            if let Some(desc) = list.iter().find(|d| d.id == svc) {
                out.push(ProviderRef {
                    name: entry.name.clone(),
                    version: entry.plugin.version().into_owned(),
                    state: entry.state.get(),
                    service: desc.clone(),
                });
            }
        }
        out
    }

    /// 运行期重新申报自己的服务目录:宿主重织索引、作废缓存,双发 `eSrvChg.<名>`(版本串)
    /// 与 `eSrvChg`(固定键,PlugNotice)。
    ///
    /// ⚠ 纪律(DV4 教训):**先换插件 services() 当前返回,再调用本方法** ——
    /// 宿主细节是现场取的,只申报不改内容会造成索引与 live 细节脱节。
    ///
    /// (0.2.0:Rust 无运行期代码热换;「更新」的运行期形态 = 目录随配置/状态改版。)
    pub fn publish_services(&self, list: Vec<ServiceDesc>) {
        let name = self.name.clone();
        let version = self.registry.publish(&name, list);
        let _ = self.event.put(Event::with_data(
            format!("{EVENT_SRV_CHG_PREFIX}{name}"),
            Arc::new(version.clone()),
        ));
        let _ = self.event.put(Event::with_data(
            EVENT_SRV_CHG,
            Arc::new(crate::service::PlugNotice { name, version }),
        ));
    }

    // ================ 请求应答 ================

    /// 请求出口:eReq.<target> + Envelope + std mpsc 一次性信道。
    ///
    /// **禁止自呼**:ask 指向自身 target 会占满自己的 worker 直到超时
    /// (自己阻塞等自己的回包)——需要自身逻辑请直调。
    ///
    /// 时序:预检目标 Running → put → recv_timeout。
    /// - 预检失败(不存在/未 Running/已 Quarantined)→ 立即 `Err(TargetGone)`;
    /// - 回包 `Ok(Res)` 且契约类型匹配 → `Ok(Res)`;
    /// - 回包 `Err(InvalidPayload)` **或** 回包类型与本侧契约不一致 → `Err(InvalidPayload)`;
    /// - 信道断连(handler panic/信封丢失/目标邮箱溢出被丢)→ `Err(TargetGone)`(尽快,不虚耗超时);
    /// - 等满时限 → `Err(Timeout)`;
    /// - put 失败(总线已停)→ `Err(BusStopped)`。
    pub fn ask<Req, Res>(&self, target: &str, req: Req, timeout: Duration) -> Result<Res, AskError>
    where
        Req: Any + Send + Sync,
        Res: Any + Send + Sync,
    {
        match self.registry.get(target) {
            Some(entry) if entry.state.get() == PluginState::Running => {}
            _ => {
                return Err(AskError::TargetGone {
                    target: target.to_string(),
                });
            }
        }
        self.deliver(target, req, timeout)
    }

    /// 按服务 id 的请求出口(0.2.0):index 命中 → 现场类型闸门 → 发 eReq。
    ///
    /// - 无人提供该服务 → 立即 `Err(UnknownService)`(零成本显错);
    /// - 有提供方但全都非 Running → 立即 `Err(TargetGone)`;
    /// - Req 类型与服务声明的 req_type 不符 → 立即 `Err(InvalidPayload)`
    ///   (把「绕一圈才发现错型」提前到发出之前);
    /// - 多提供方 → 取注册序第一个 Running 者;后续 failover 由发起方在收不到回包后自定。
    ///   其余失败面与 [`ask`](Self::ask) 完全一致。
    pub fn ask_service<Req, Res>(
        &self,
        svc: &str,
        req: Req,
        timeout: Duration,
    ) -> Result<Res, AskError>
    where
        Req: Any + Send + Sync,
        Res: Any + Send + Sync,
    {
        let providers = self.find_service(svc);
        if providers.is_empty() {
            return Err(AskError::UnknownService {
                service: svc.to_string(),
            });
        }
        let Some(slot) = providers.iter().find(|p| p.state == PluginState::Running) else {
            return Err(AskError::TargetGone {
                target: providers[0].name.clone(),
            });
        };
        if slot.service.req_type != type_name::<Req>() {
            return Err(AskError::InvalidPayload {
                target: slot.name.clone(),
            });
        }
        self.deliver(&slot.name.clone(), req, timeout)
    }

    /// ask/ask_service 共用的投递段(Envelope + eReq + recv_timeout + 回包裁决)。
    fn deliver<Req, Res>(&self, target: &str, req: Req, timeout: Duration) -> Result<Res, AskError>
    where
        Req: Any + Send + Sync,
        Res: Any + Send + Sync,
    {
        let (envelope, reply_rx) = Envelope::new(Arc::new(req));
        let put = self
            .event
            .put(Event::with_data(e_req_type(target), Arc::new(envelope)));
        if put.is_err() {
            return Err(AskError::BusStopped);
        }
        let boxed = reply_rx.recv_timeout(timeout).map_err(|e| match e {
            OneshotErr::Timeout => AskError::Timeout {
                target: target.to_string(),
                // 读数:超时那一刻目标邮箱里排了多少帧(目标已摘 → 0)
                pending: self
                    .registry
                    .get(target)
                    .map(|e| e.stats.pending.load(std::sync::atomic::Ordering::Acquire))
                    .unwrap_or(0),
            },
            OneshotErr::Closed => AskError::TargetGone {
                target: target.to_string(),
            },
        })?;
        // 回包为 Box<dyn Any>:downcast 到本侧契约 Result<Res, ReplyErr>
        match boxed.downcast::<Result<Res, ReplyErr>>() {
            Ok(inner) => match *inner {
                Ok(res) => Ok(res),
                Err(_) => Err(AskError::InvalidPayload {
                    target: target.to_string(),
                }),
            },
            // 回包类型与本侧契约不同(双方契约错配)→ 协议违约
            Err(_) => Err(AskError::InvalidPayload {
                target: target.to_string(),
            }),
        }
    }
}