sz_rust_core/plugin/
event_bus.rs

1//! 事件总线 trait + EventHandler trait 抽象。
2//!
3//! 对应 design.md §2.2.2 接口 7-8。
4//! 所有 trait 必须 `Send + Sync + 'static`。
5
6use async_trait::async_trait;
7use serde::{Deserialize, Serialize};
8use std::sync::Arc;
9
10use super::schema::SysEvent;
11
12/// 事件 ID 类型。
13pub type EventId = i64;
14
15/// 订阅 ID 类型。
16pub type SubscriptionId = u64;
17
18/// 插件事件。
19#[derive(Debug, Clone, Serialize, Deserialize)]
20pub struct PluginEvent {
21    pub id: EventId,
22    pub tenant_id: i64,
23    pub event_type: String,
24    pub source_plugin: String,
25    pub payload: serde_json::Value,
26}
27
28impl From<SysEvent> for PluginEvent {
29    fn from(e: SysEvent) -> Self {
30        Self {
31            id: e.id,
32            tenant_id: e.tenant_id,
33            event_type: e.event_type,
34            source_plugin: e.source_plugin,
35            payload: e.payload,
36        }
37    }
38}
39
40/// 事件处理器 trait。
41#[async_trait]
42pub trait EventHandler: Send + Sync + 'static {
43    /// 处理事件，返回 `Ok(())` 表示处理成功。
44    async fn handle(&self, event: &PluginEvent) -> Result<(), String>;
45}
46
47/// 事件总线 trait。
48#[async_trait]
49pub trait EventBus: Send + Sync + 'static {
50    /// 发布事件，返回事件 ID。
51    async fn publish(&self, event: &PluginEvent) -> Result<EventId, String>;
52
53    /// 订阅事件，返回订阅 ID。
54    async fn subscribe(
55        &self,
56        event_type: &str,
57        handler: Arc<dyn EventHandler>,
58    ) -> Result<SubscriptionId, String>;
59
60    /// 取消订阅。
61    async fn unsubscribe(&self, sub_id: SubscriptionId) -> Result<(), String>;
62
63    /// 重放未投递事件（至少一次投递保障）。
64    async fn replay_pending(&self) -> Result<usize, String>;
65}
66
67/// 内存事件总线实现（用于测试和轻量场景）。
68pub struct InMemoryEventBus {
69    events: parking_lot::RwLock<Vec<PluginEvent>>,
70    next_id: parking_lot::Mutex<EventId>,
71    subscribers: parking_lot::RwLock<std::collections::HashMap<String, Vec<(SubscriptionId, Arc<dyn EventHandler>)>>>,
72    next_sub_id: parking_lot::Mutex<SubscriptionId>,
73}
74
75impl InMemoryEventBus {
76    pub fn new() -> Self {
77        Self {
78            events: parking_lot::RwLock::new(Vec::new()),
79            next_id: parking_lot::Mutex::new(1),
80            subscribers: parking_lot::RwLock::new(std::collections::HashMap::new()),
81            next_sub_id: parking_lot::Mutex::new(1),
82        }
83    }
84}
85
86impl Default for InMemoryEventBus {
87    fn default() -> Self {
88        Self::new()
89    }
90}
91
92#[async_trait]
93impl EventBus for InMemoryEventBus {
94    async fn publish(&self, event: &PluginEvent) -> Result<EventId, String> {
95        let id = {
96            let mut next = self.next_id.lock();
97            let id = *next;
98            *next += 1;
99            id
100        };
101        let mut event = event.clone();
102        event.id = id;
103        let event_type = event.event_type.clone();
104        self.events.write().push(event.clone());
105        let handlers: Vec<Arc<dyn EventHandler>> = {
106            let subs = self.subscribers.read();
107            subs.get(&event_type)
108                .map(|v| v.iter().map(|(_, h)| h.clone()).collect())
109                .unwrap_or_default()
110        };
111        for handler in handlers {
112            let _ = handler.handle(&event).await;
113        }
114        Ok(id)
115    }
116
117    async fn subscribe(
118        &self,
119        event_type: &str,
120        handler: Arc<dyn EventHandler>,
121    ) -> Result<SubscriptionId, String> {
122        let sub_id = {
123            let mut next = self.next_sub_id.lock();
124            let id = *next;
125            *next += 1;
126            id
127        };
128        let mut subs = self.subscribers.write();
129        subs.entry(event_type.to_string())
130            .or_default()
131            .push((sub_id, handler));
132        Ok(sub_id)
133    }
134
135    async fn unsubscribe(&self, sub_id: SubscriptionId) -> Result<(), String> {
136        let mut subs = self.subscribers.write();
137        for handlers in subs.values_mut() {
138            handlers.retain(|(id, _)| *id != sub_id);
139        }
140        Ok(())
141    }
142
143    async fn replay_pending(&self) -> Result<usize, String> {
144        Ok(0)
145    }
146}
147
148#[cfg(test)]
149mod tests {
150    use super::*;
151
152    struct CountHandler {
153        count: parking_lot::Mutex<usize>,
154    }
155
156    #[async_trait]
157    impl EventHandler for CountHandler {
158        async fn handle(&self, _event: &PluginEvent) -> Result<(), String> {
159            *self.count.lock() += 1;
160            Ok(())
161        }
162    }
163
164    #[tokio::test]
165    async fn test_publish_and_subscribe() {
166        let bus = InMemoryEventBus::new();
167        let handler = Arc::new(CountHandler { count: parking_lot::Mutex::new(0) });
168        let _ = bus.subscribe("test.event", handler.clone()).await;
169        let event = PluginEvent {
170            id: 0,
171            tenant_id: 1,
172            event_type: "test.event".to_string(),
173            source_plugin: "test".to_string(),
174            payload: serde_json::json!({}),
175        };
176        let id = bus.publish(&event).await.expect("发布失败");
177        assert!(id > 0);
178        assert_eq!(*handler.count.lock(), 1);
179    }
180}
sz_rust_core/plugin/event_bus.rs

sz_rust_core/plugin/
event_bus.rs
