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::Arc;
16use std::sync::atomic::{AtomicBool, Ordering};
17use std::sync::mpsc::{RecvTimeoutError, Sender as StdSender};
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(
57        payload: Arc<dyn Any + Send + Sync>,
58    ) -> (Envelope, OneshotReceiver<Box<dyn Any + Send + Sync>>) {
59        let (tx, rx) = oneshot();
60        (Envelope { payload, reply: tx }, rx)
61    }
62
63    /// 接收方标准处理(B 侧唯一入口helper):
64    /// - payload downcast 成功 → `handler(&req)`,回 `Ok(Res)`;
65    /// - payload downcast 失败 → 回 `Err(ReplyErr::InvalidPayload)`;
66    /// - event 载荷不是 Envelope → 静默忽略(错入 type 的兜底);
67    /// - handler panic → 本函数帧先 unwind,reply 未 send → drop 断连(天然错误信令)。
68    pub fn handle<Req, Res>(event: &Event, handler: impl FnOnce(&Req) -> Res)
69    where
70        Req: Any + Send + Sync,
71        Res: Any + Send + Sync,
72    {
73        let Some(env) = event.data.downcast_ref::<Envelope>() else {
74            return;
75        };
76        match Arc::clone(&env.payload).downcast::<Req>() {
77            Ok(req) => {
78                let res = handler(&*req);
79                let _ = env.reply.send(Box::new(Ok::<Res, ReplyErr>(res)));
80            }
81            Err(_) => {
82                let _ = env
83                    .reply
84                    .send(Box::new(Err::<Res, ReplyErr>(ReplyErr::InvalidPayload)));
85            }
86        }
87    }
88}
89
90/// 一次性发送端(共享引用可发;AtomicBool 守卫保证协议上只有第一次 send 生效)。
91///
92/// 设计上的让步(实现期修订):worker 拿到的是 `&Event`,无法把 Envelope.reply move 出来,
93/// 因此 send 取 `&self` + 内部 `sent` 标志 —— 第二次及以后调用直接 `Err(v)` 返回,协议上仍是一发。
94pub struct OneshotSender<T> {
95    tx: StdSender<T>,
96    sent: AtomicBool,
97}
98
99impl<T> OneshotSender<T> {
100    pub(crate) fn new_inner(tx: StdSender<T>) -> Self {
101        Self {
102            tx,
103            sent: AtomicBool::new(false),
104        }
105    }
106    /// 第一次调用生效;接收端已弃收则 `Err(v)` 原样归还;重复调用直接 `Err(v)`。
107    pub fn send(&self, v: T) -> Result<(), T> {
108        if self.sent.swap(true, Ordering::SeqCst) {
109            return Err(v);
110        }
111        self.tx.send(v).map_err(|e| e.0)
112    }
113}
114
115/// 接收端错误。
116#[derive(Debug, Clone, Copy, PartialEq, Eq)]
117pub enum OneshotErr {
118    /// 等满时限仍未回包。
119    Timeout,
120    /// 信道关闭(接收前发送端被 drop / 目标异常退出)。
121    Closed,
122}
123
124/// 一次性接收端(recv 消费自身)。
125pub struct OneshotReceiver<T>(std::sync::mpsc::Receiver<T>);
126
127impl<T> OneshotReceiver<T> {
128    pub(crate) fn new_inner(rx: std::sync::mpsc::Receiver<T>) -> Self {
129        Self(rx)
130    }
131    /// 无限期等待;`Err(OneshotErr::Closed)` = 发送端先消失。
132    pub fn recv(self) -> Result<T, OneshotErr> {
133        self.0.recv().map_err(|_| OneshotErr::Closed)
134    }
135    /// 带时限等待。
136    pub fn recv_timeout(self, d: Duration) -> Result<T, OneshotErr> {
137        match self.0.recv_timeout(d) {
138            Ok(v) => Ok(v),
139            Err(RecvTimeoutError::Timeout) => Err(OneshotErr::Timeout),
140            Err(RecvTimeoutError::Disconnected) => Err(OneshotErr::Closed),
141        }
142    }
143}
144
145/// 构造一对 one-shot 信道。
146pub fn oneshot<T>() -> (OneshotSender<T>, OneshotReceiver<T>) {
147    let (tx, rx) = std::sync::mpsc::channel();
148    (OneshotSender::new_inner(tx), OneshotReceiver::new_inner(rx))
149}
ek_core/envelope.rs

ek_core/
envelope.rs
