ek_core/
envelope.rs

1//! 请求应答信封与一次性信道(envelope + Oneshot)。
2//!
3//! 设计要点(P-H 执行期修订,审核记录于 TESTPLAN):
4//! - `Envelope` **非泛型**:`payload: Arc<dyn Any + Send + Sync>` + erasure 回包
5//!   `reply: OneshotSender<Box<dyn Any + Send + Sync>>`。
6//!   原因:若信封双层泛型 `<Req,Res>`,一旦收发双方契约类型不同,收方整封 downcast 失败
7//!   拿不到回包 Sender,"错型必回 InvalidPayload" 无法实现(A 的 Res 只在 A 侧存在)。
8//!   抹平后的契约:B 拆 payload(错型→可回 Err(InvalidPayload)),A 收回包后自己 downcast
9//!   判定(错型同样落 InvalidPayload)。任何路径都不存在类型混淆(全部经 Any TypeId 校验)。
10//! - 一次性信道 = std::sync::mpsc(D5):零依赖;断连=TargetGone、超时=Timeout。
11
12use ek_event::Event;
13use std::any::Any;
14use std::fmt;
15use std::sync::atomic::{AtomicBool, Ordering};
16use std::sync::mpsc::{RecvTimeoutError, Sender as StdSender};
17use std::sync::Arc;
18use std::time::Duration;
19
20/// 寻址请求 type:`eReq.<target>`(精确 key,与 ek-event 精确匹配语义一致)。
21pub fn e_req_type(target: &str) -> String {
22    format!("eReq.{target}")
23}
24
25/// 定向通知 type:`eTo.<target>`。
26pub fn e_to_type(target: &str) -> String {
27    format!("eTo.{target}")
28}
29
30/// 协议级错误(B 明确回)。
31#[derive(Debug, Clone, Copy, PartialEq, Eq)]
32pub enum ReplyErr {
33    /// 请求载荷类型与本插件契约不符。
34    InvalidPayload,
35}
36
37impl fmt::Display for ReplyErr {
38    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
39        match self {
40            Self::InvalidPayload => write!(f, "invalid payload"),
41        }
42    }
43}
44
45/// 寻址信封(非泛型,payload 抹平)。
46pub struct Envelope {
47    /// 请求载荷(由发起方 Box 于 Arc 中;接收方 downcast 到自身契约类型)。
48    pub payload: Arc<dyn Any + Send + Sync>,
49    /// 义务回包:成功 `Box::new(Ok(Res))`,契约错 `Box::new(Err(ReplyErr::InvalidPayload))`。
50    /// *handler panic 时 reply 随信封 drop → 信道关闭 → 发起方立即 TargetGone*。
51    pub reply: OneshotSender<Box<dyn Any + Send + Sync>>,
52}
53
54impl Envelope {
55    /// 发起方构造helper:由 `ctx.ask` 使用。
56    pub fn new(payload: Arc<dyn Any + Send + Sync>) -> (Envelope, OneshotReceiver<Box<dyn Any + Send + Sync>>) {
57        let (tx, rx) = oneshot();
58        (Envelope { payload, reply: tx }, rx)
59    }
60
61    /// 接收方标准处理(B 侧唯一入口helper):
62    /// - payload downcast 成功 → `handler(&req)`,回 `Ok(Res)`;
63    /// - payload downcast 失败 → 回 `Err(ReplyErr::InvalidPayload)`;
64    /// - event 载荷不是 Envelope → 静默忽略(错入 type 的兜底);
65    /// - handler panic → 本函数帧先 unwind,reply 未 send → drop 断连(天然错误信令)。
66    pub fn handle<Req, Res>(event: &Event, handler: impl FnOnce(&Req) -> Res)
67    where
68        Req: Any + Send + Sync,
69        Res: Any + Send + Sync,
70    {
71        let Some(env) = event.data.downcast_ref::<Envelope>() else {
72            return;
73        };
74        match Arc::clone(&env.payload).downcast::<Req>() {
75            Ok(req) => {
76                let res = handler(&*req);
77                let _ = env.reply.send(Box::new(Ok::<Res, ReplyErr>(res)));
78            }
79            Err(_) => {
80                let _ = env.reply.send(Box::new(Err::<Res, ReplyErr>(ReplyErr::InvalidPayload)));
81            }
82        }
83    }
84}
85
86/// 一次性发送端(共享引用可发;AtomicBool 守卫保证协议上只有第一次 send 生效)。
87///
88/// 设计上的让步(实现期修订):worker 拿到的是 `&Event`,无法把 Envelope.reply move 出来,
89/// 因此 send 取 `&self` + 内部 `sent` 标志 —— 第二次及以后调用直接 `Err(v)` 返回,协议上仍是一发。
90pub struct OneshotSender<T> {
91    tx: StdSender<T>,
92    sent: AtomicBool,
93}
94
95impl<T> OneshotSender<T> {
96    pub(crate) fn new_inner(tx: StdSender<T>) -> Self {
97        Self { tx, sent: AtomicBool::new(false) }
98    }
99    /// 第一次调用生效;接收端已弃收则 `Err(v)` 原样归还;重复调用直接 `Err(v)`。
100    pub fn send(&self, v: T) -> Result<(), T> {
101        if self.sent.swap(true, Ordering::SeqCst) {
102            return Err(v);
103        }
104        self.tx.send(v).map_err(|e| e.0)
105    }
106}
107
108/// 接收端错误。
109#[derive(Debug, Clone, Copy, PartialEq, Eq)]
110pub enum OneshotErr {
111    /// 等满时限仍未回包。
112    Timeout,
113    /// 信道关闭(接收前发送端被 drop / 目标异常退出)。
114    Closed,
115}
116
117/// 一次性接收端(recv 消费自身)。
118pub struct OneshotReceiver<T>(std::sync::mpsc::Receiver<T>);
119
120impl<T> OneshotReceiver<T> {
121    pub(crate) fn new_inner(rx: std::sync::mpsc::Receiver<T>) -> Self {
122        Self(rx)
123    }
124    /// 无限期等待;`Err(OneshotErr::Closed)` = 发送端先消失。
125    pub fn recv(self) -> Result<T, OneshotErr> {
126        self.0.recv().map_err(|_| OneshotErr::Closed)
127    }
128    /// 带时限等待。
129    pub fn recv_timeout(self, d: Duration) -> Result<T, OneshotErr> {
130        match self.0.recv_timeout(d) {
131            Ok(v) => Ok(v),
132            Err(RecvTimeoutError::Timeout) => Err(OneshotErr::Timeout),
133            Err(RecvTimeoutError::Disconnected) => Err(OneshotErr::Closed),
134        }
135    }
136}
137
138/// 构造一对 one-shot 信道。
139pub fn oneshot<T>() -> (OneshotSender<T>, OneshotReceiver<T>) {
140    let (tx, rx) = std::sync::mpsc::channel();
141    (OneshotSender::new_inner(tx), OneshotReceiver::new_inner(rx))
142}
ek_core/envelope.rs

ek_core/
envelope.rs
