绝尘
22964a2e20
- B-06: DagExecutor 接收 execution_id 下沉到 NodeContext，消除 dummy-execution-id 硬编码 - B-07: NodeContext.node_status 共享 self.state_machine.clone()（is_cancelled 可工作） - B-03a: HumanNode execute 重写为 subscribe→send→select! 循环（响应/超时/取消 + execution_id+node_id 双键 + Lagged 容忍），删假返回"同意" - eventbus.rs 删 try_recv_human_approval 死代码 - 新增 human_node 7 单测（正常/双键过滤/超时/非法决策/自由文本） 照 B-03-人工审批响应机制.md 设计。cargo check 0 error，df-nodes 14 + df-workflow 11 test 全过
62 lines
1.5 KiB
Rust
62 lines
1.5 KiB
Rust
//! 事件总线 — 基于 tokio::sync::broadcast 的发布/订阅
|
|
|
|
use tokio::sync::broadcast;
|
|
use tokio::sync::broadcast::error::SendError;
|
|
use df_core::events::WorkflowEvent;
|
|
|
|
/// 事件总线
|
|
#[derive(Debug)]
|
|
pub struct EventBus {
|
|
sender: broadcast::Sender<WorkflowEvent>,
|
|
}
|
|
|
|
/// 事件订阅者
|
|
pub type EventSubscriber = broadcast::Receiver<WorkflowEvent>;
|
|
|
|
/// 默认事件通道容量
|
|
const DEFAULT_CAPACITY: usize = 256;
|
|
|
|
impl EventBus {
|
|
/// 创建事件总线
|
|
pub fn new() -> Self {
|
|
let (sender, _) = broadcast::channel(DEFAULT_CAPACITY);
|
|
Self { sender }
|
|
}
|
|
|
|
/// 创建指定容量的事件总线
|
|
pub fn with_capacity(capacity: usize) -> Self {
|
|
let (sender, _) = broadcast::channel(capacity);
|
|
Self { sender }
|
|
}
|
|
|
|
/// 发送事件
|
|
pub async fn send(&self, event: WorkflowEvent) {
|
|
// broadcast::send 是同步的，忽略接收者已关闭的错误
|
|
let _ = self.sender.send(event);
|
|
}
|
|
|
|
/// 订阅事件
|
|
pub fn subscribe(&self) -> EventSubscriber {
|
|
self.sender.subscribe()
|
|
}
|
|
|
|
/// 发送人工审批请求
|
|
pub fn emit_human_approval_request(&self, event: WorkflowEvent) -> Result<usize, SendError<WorkflowEvent>> {
|
|
self.sender.send(event)
|
|
}
|
|
}
|
|
|
|
impl Default for EventBus {
|
|
fn default() -> Self {
|
|
Self::new()
|
|
}
|
|
}
|
|
|
|
impl Clone for EventBus {
|
|
fn clone(&self) -> Self {
|
|
Self {
|
|
sender: self.sender.clone(),
|
|
}
|
|
}
|
|
}
|