sa_token_core/event/
mod.rs

1// Author: 金书记
2//
3//! Event Listener Module | 事件监听模块
4//!
5//! Provides event listening capabilities for sa-token, supporting monitoring of login, logout, kick-out, and other operations.
6//!
7//! 提供 sa-token 的事件监听功能，支持监听登录、登出、踢出等操作。
8//!
9//! ## EventBus Code Flow Logic | EventBus 代码流程逻辑
10//!
11//! ### Overall Architecture | 整体架构
12//!
13//! ```text
14//! ┌─────────────────────────────────────────────────────────────┐
15//! │                    SaTokenEventBus                          │
16//! │  ┌────────────────────────────────────────────────────┐    │
17//! │  │  listeners: Arc<RwLock<Vec<Arc<dyn SaTokenListener>>>>  │
18//! │  │  config: EventBusConfig                            │    │
19//! │  │  - Stores all registered listeners                 │    │
20//! │  │    存储所有注册的监听器                             │    │
21//! │  │  - Uses RwLock for thread safety                   │    │
22//! │  │    使用 RwLock 保证线程安全                        │    │
23//! │  │  - Arc wrapping allows multi-thread sharing        │    │
24//! │  │    Arc 包装允许多线程共享                          │    │
25//! │  └────────────────────────────────────────────────────┘    │
26//! └─────────────────────────────────────────────────────────────┘
27//! ```
28//!
29//! ### Core Processes | 核心流程
30//!
31//! #### 1. Listener Registration Process | 监听器注册流程
32//!
33//! ```text
34//! ┌──────────┐     ┌──────────────┐     ┌─────────────┐
35//! │User Code │────▶│ register()   │────▶│Acquire Write│
36//! │用户代码  │     │              │     │Lock 写锁获取│
37//! └──────────┘     │ - Receive    │     │             │
38//!                  │   listener   │     │ - Get lock  │
39//!                  │   接收监听器  │     │   获取写锁   │
40//!                  │ - Arc wrap   │     │ - Add to    │
41//!                  │   Arc包装    │     │   list      │
42//!                  └──────────────┘     │   添加到列表 │
43//!                                       │ - Release   │
44//!                                       │   释放写锁   │
45//!                                       └─────────────┘
46//!
47//! Steps | 步骤：
48//! 1. User creates custom listener instance
49//!    用户创建自定义监听器实例
50//! 2. Wrap listener with Arc::new()
51//!    使用 Arc::new() 包装监听器
52//! 3. Call event_bus.register(listener).await
53//!    调用 event_bus.register(listener).await
54//! 4. EventBus acquires write lock, adds listener to Vec
55//!    EventBus 获取写锁，将监听器添加到 Vec 中
56//! 5. Registration complete, waiting for event triggers
57//!    监听器注册完成，等待事件触发
58//! ```
59//!
60//! #### 2. Event Publishing Process (DispatchMode) | 事件发布流程 (分发模式)
61//!
62//! ```text
63//! SaTokenManager::login OK
64//!        │
65//!        ▼
66//!  event = SaTokenEvent::login(login_id, token)
67//!        │
68//!        ▼
69//!  event_bus.publish(event)  ← dispatch by config.dispatch_mode
70//!        │
71//!        ├─[Sequential (default)]──── for each listener: spawn + timeout + await
72//!        ├─[Concurrent]───────────── spawn all + timeout + join_all
73//!        └─[Detached]─────────────── tokio::spawn, return immediately
74//! ```
75//!
76//! ### Thread Safety Guarantees | 线程安全保证
77//!
78//! ```text
79//! Arc<RwLock<Vec<Arc<dyn SaTokenListener>>>>
80//!  │    │     │    │
81//!  │    │     │    └─ Listener trait object | 监听器 trait 对象
82//!  │    │     └────── Listener collection | 监听器集合
83//!  │    └──────────── Read-write lock protection | 读写锁保护
84//!  └───────────────── Cross-thread sharing | 跨线程共享
85//!
86//! - Arc: Allows EventBus to be shared across multiple Manager instances
87//!        允许 EventBus 被多个 Manager 实例共享
88//! - RwLock: Allows multiple readers to publish events concurrently, writer has exclusive registration
89//!           允许多个读者同时发布事件，写者独占注册
90//! - Inner Arc: Listeners can be shared across multiple EventBus instances
91//!              监听器可以被多个 EventBus 共享
92//! ```
93//!
94//! ## Usage Example | 使用示例
95//!
96//! ```rust,ignore
97//! use sa_token_core::event::{SaTokenEvent, SaTokenListener, SaTokenEventBus};
98//!
99//! // Custom listener | 自定义监听器
100//! struct MyListener;
101//!
102//! #[async_trait]
103//! impl SaTokenListener for MyListener {
104//!     async fn on_login(&self, login_id: &str, token: &str, login_type: &str) {
105//!         println!("User {} logged in, token: {}", login_id, token);
106//!         // 用户 {} 登录了，token: {}
107//!     }
108//!     
109//!     async fn on_logout(&self, login_id: &str, token: &str, login_type: &str) {
110//!         println!("User {} logged out", login_id);
111//!         // 用户 {} 登出了
112//!     }
113//! }
114//!
115//! // Register listener | 注册监听器
116//! let event_bus = SaTokenEventBus::new();
117//! event_bus.register(Arc::new(MyListener)).await;
118//! ```
119
120use async_trait::async_trait;
121use chrono::{DateTime, Utc};
122use serde::{Deserialize, Serialize};
123use std::sync::Arc;
124use std::sync::RwLock;
125use std::time::Duration;
126
127/// 事件类型 | Event Type
128#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
129pub enum SaTokenEventType {
130    /// 登录事件 | Login event
131    Login,
132    /// 登出事件 | Logout event
133    Logout,
134    /// 踢出下线事件 | Kick out event
135    KickOut,
136    /// Token 续期事件 | Token renewal event
137    RenewTimeout,
138    /// 被顶下线事件（被其他设备登录）| Replaced by another login
139    Replaced,
140    /// 被封禁事件 | Banned event
141    Banned,
142    /// 解封事件 | Unbanned event
143    Unbanned,
144    /// 开启二级认证 | Open safe authentication
145    OpenSafe,
146    /// 关闭二级认证 | Close safe authentication
147    CloseSafe,
148    /// 二级认证校验通过 | Safe verification passed
149    SafeVerify,
150    /// 权限/角色数据变更 | Permission or role data changed
151    ///
152    /// 由 [`crate::service::AuthzService`] 的写操作触发。
153    /// Emitted by write operations in `AuthzService`.
154    GrantChanged,
155}
156
157/// 事件分发模式 | Event dispatch mode
158///
159/// 控制监听器的执行方式：顺序、并行、后台。
160/// Controls how listeners are executed: sequential, concurrent, or background.
161#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
162pub enum DispatchMode {
163    /// 顺序执行，await 全部监听器（默认，兼容旧行为）
164    ///
165    /// Sequential execution, awaiting all listeners (default, compatible with old behavior).
166    #[default]
167    Sequential,
168    /// 并行执行，await 全部监听器
169    ///
170    /// Concurrent execution, awaiting all listeners in parallel.
171    Concurrent,
172    /// 后台执行，不阻塞 publish 调用方返回
173    ///
174    /// Detached execution, does not block the publisher.
175    Detached,
176}
177
178/// EventBus 运行时配置 | EventBus runtime configuration
179#[derive(Debug, Clone, Serialize, Deserialize)]
180pub struct EventBusConfig {
181    /// 分发模式 | Dispatch mode
182    pub dispatch_mode: DispatchMode,
183    /// 单个监听器最大执行时长；None 表示不限时
184    ///
185    /// Maximum execution time per listener; `None` means no timeout.
186    pub listener_timeout: Option<Duration>,
187}
188
189impl Default for EventBusConfig {
190    fn default() -> Self {
191        Self {
192            dispatch_mode: DispatchMode::Sequential,
193            listener_timeout: Some(Duration::from_secs(5)),
194        }
195    }
196}
197
198impl EventBusConfig {
199    /// 创建无超时限制的配置（用于向后兼容）
200    ///
201    /// Creates a config with no timeout (for backward compatibility).
202    pub fn no_timeout() -> Self {
203        Self {
204            listener_timeout: None,
205            ..Default::default()
206        }
207    }
208}
209
210/// 事件数据 | Event data
211#[derive(Debug, Clone, Serialize, Deserialize)]
212pub struct SaTokenEvent {
213    /// 事件类型 | Event type
214    pub event_type: SaTokenEventType,
215    /// 登录ID | Login ID
216    pub login_id: String,
217    /// Token 值 | Token value
218    pub token: String,
219    /// 登录类型（如 "default", "admin" 等）| Login type (e.g. "default", "admin")
220    pub login_type: String,
221    /// 事件发生时间 | Event timestamp
222    pub timestamp: DateTime<Utc>,
223    /// 额外数据（用于扩展）| Extra data (for extension)
224    pub extra: Option<serde_json::Value>,
225}
226
227impl SaTokenEvent {
228    /// 创建登录事件 | Create login event
229    pub fn login(login_id: impl Into<String>, token: impl Into<String>) -> Self {
230        Self {
231            event_type: SaTokenEventType::Login,
232            login_id: login_id.into(),
233            token: token.into(),
234            login_type: "default".to_string(),
235            timestamp: Utc::now(),
236            extra: None,
237        }
238    }
239
240    /// 创建登出事件 | Create logout event
241    pub fn logout(login_id: impl Into<String>, token: impl Into<String>) -> Self {
242        Self {
243            event_type: SaTokenEventType::Logout,
244            login_id: login_id.into(),
245            token: token.into(),
246            login_type: "default".to_string(),
247            timestamp: Utc::now(),
248            extra: None,
249        }
250    }
251
252    /// 创建踢出下线事件 | Create kick out event
253    pub fn kick_out(login_id: impl Into<String>, token: impl Into<String>) -> Self {
254        Self {
255            event_type: SaTokenEventType::KickOut,
256            login_id: login_id.into(),
257            token: token.into(),
258            login_type: "default".to_string(),
259            timestamp: Utc::now(),
260            extra: None,
261        }
262    }
263
264    /// 创建 Token 续期事件 | Create token renewal event
265    ///
266    /// # 参数 | Parameters
267    /// - `login_id`: 登录 ID
268    /// - `token`: Token 值
269    /// - `timeout_seconds`: 续期后的有效时长（秒）| Renewed validity period (seconds)
270    pub fn renew_timeout(
271        login_id: impl Into<String>,
272        token: impl Into<String>,
273        timeout_seconds: i64,
274    ) -> Self {
275        Self {
276            event_type: SaTokenEventType::RenewTimeout,
277            login_id: login_id.into(),
278            token: token.into(),
279            login_type: "default".to_string(),
280            timestamp: Utc::now(),
281            extra: Some(serde_json::json!({ "timeout_seconds": timeout_seconds })),
282        }
283    }
284
285    /// 创建被顶下线事件 | Create replaced event
286    pub fn replaced(login_id: impl Into<String>, token: impl Into<String>) -> Self {
287        Self {
288            event_type: SaTokenEventType::Replaced,
289            login_id: login_id.into(),
290            token: token.into(),
291            login_type: "default".to_string(),
292            timestamp: Utc::now(),
293            extra: None,
294        }
295    }
296
297    /// 创建被封禁事件 | Create banned event
298    ///
299    /// # 参数 | Parameters
300    /// - `login_id`: 登录 ID
301    /// - `service`: 封禁服务标识（如 "login", "comment"）| Service identifier
302    /// - `level`: 封禁等级 | Ban level
303    pub fn banned(login_id: impl Into<String>, service: impl Into<String>, level: i32) -> Self {
304        Self {
305            event_type: SaTokenEventType::Banned,
306            login_id: login_id.into(),
307            token: String::new(),
308            login_type: "default".to_string(),
309            timestamp: Utc::now(),
310            extra: Some(serde_json::json!({ "service": service.into(), "level": level })),
311        }
312    }
313
314    /// 创建解封事件 | Create unbanned event
315    ///
316    /// # 参数 | Parameters
317    /// - `login_id`: 登录 ID
318    /// - `service`: 解封服务标识 | Service identifier that was unbanned
319    pub fn unbanned(login_id: impl Into<String>, service: impl Into<String>) -> Self {
320        Self {
321            event_type: SaTokenEventType::Unbanned,
322            login_id: login_id.into(),
323            token: String::new(),
324            login_type: "default".to_string(),
325            timestamp: Utc::now(),
326            extra: Some(serde_json::json!({ "service": service.into() })),
327        }
328    }
329
330    /// 创建开启二级认证事件 | Create open safe event
331    ///
332    /// service 存入 extra 字段而非 login_type，避免语义混乱。
333    /// Service stored in `extra` instead of `login_type` to avoid semantic confusion.
334    pub fn open_safe(token: impl Into<String>, service: impl Into<String>) -> Self {
335        let svc = service.into();
336        Self {
337            event_type: SaTokenEventType::OpenSafe,
338            login_id: String::new(),
339            token: token.into(),
340            login_type: "default".to_string(),
341            timestamp: Utc::now(),
342            extra: Some(serde_json::json!({ "service": svc })),
343        }
344    }
345
346    /// 创建关闭二级认证事件 | Create close safe event
347    pub fn close_safe(token: impl Into<String>, service: impl Into<String>) -> Self {
348        let svc = service.into();
349        Self {
350            event_type: SaTokenEventType::CloseSafe,
351            login_id: String::new(),
352            token: token.into(),
353            login_type: "default".to_string(),
354            timestamp: Utc::now(),
355            extra: Some(serde_json::json!({ "service": svc })),
356        }
357    }
358
359    /// 创建二级认证校验通过事件 | Create safe verification passed event
360    pub fn safe_verify(token: impl Into<String>, service: impl Into<String>) -> Self {
361        let svc = service.into();
362        Self {
363            event_type: SaTokenEventType::SafeVerify,
364            login_id: String::new(),
365            token: token.into(),
366            login_type: "default".to_string(),
367            timestamp: Utc::now(),
368            extra: Some(serde_json::json!({ "service": svc })),
369        }
370    }
371
372    /// 创建权限/角色变更事件 | Create grant changed event
373    pub fn grant_changed(login_id: impl Into<String>, login_type: impl Into<String>) -> Self {
374        Self {
375            event_type: SaTokenEventType::GrantChanged,
376            login_id: login_id.into(),
377            token: String::new(),
378            login_type: login_type.into(),
379            timestamp: Utc::now(),
380            extra: None,
381        }
382    }
383
384    /// 设置登录类型 | Set login type
385    pub fn with_login_type(mut self, login_type: impl Into<String>) -> Self {
386        self.login_type = login_type.into();
387        self
388    }
389
390    /// 设置额外数据 | Set extra data
391    pub fn with_extra(mut self, extra: serde_json::Value) -> Self {
392        self.extra = Some(extra);
393        self
394    }
395}
396
397/// 事件监听器 trait | Event Listener Trait
398///
399/// 实现此 trait 来自定义事件处理逻辑
400/// Implement this trait to customize event handling logic
401///
402/// # 使用示例 | Usage Example
403///
404/// ```rust,ignore
405/// use async_trait::async_trait;
406/// use sa_token_core::SaTokenListener;
407///
408/// struct MyListener;
409///
410/// #[async_trait]
411/// impl SaTokenListener for MyListener {
412///     async fn on_login(&self, login_id: &str, token: &str, login_type: &str) {
413///         // 自定义登录处理 | Custom login handling
414///         println!("User {} logged in", login_id);
415///     }
416/// }
417/// ```
418#[async_trait]
419pub trait SaTokenListener: Send + Sync {
420    /// 登录事件 | Login Event
421    async fn on_login(&self, login_id: &str, token: &str, login_type: &str) {
422        let _ = (login_id, token, login_type);
423    }
424
425    /// 登出事件 | Logout Event
426    async fn on_logout(&self, login_id: &str, token: &str, login_type: &str) {
427        let _ = (login_id, token, login_type);
428    }
429
430    /// 踢出下线事件 | Kick Out Event
431    async fn on_kick_out(&self, login_id: &str, token: &str, login_type: &str) {
432        let _ = (login_id, token, login_type);
433    }
434
435    /// Token 续期事件 | Token Renewal Event
436    ///
437    /// # 参数 | Parameters
438    /// - `login_id`: 登录 ID | Login ID
439    /// - `token`: Token 值 | Token value
440    /// - `login_type`: 登录类型 | Login type
441    /// - `timeout_seconds`: 续期后的有效时长（秒）| Renewed validity period (seconds)
442    async fn on_renew_timeout(
443        &self,
444        login_id: &str,
445        token: &str,
446        login_type: &str,
447        timeout_seconds: i64,
448    ) {
449        let _ = (login_id, token, login_type, timeout_seconds);
450    }
451
452    /// 被顶下线事件 | Replaced Event
453    async fn on_replaced(&self, login_id: &str, token: &str, login_type: &str) {
454        let _ = (login_id, token, login_type);
455    }
456
457    /// 被封禁事件 | Banned Event
458    async fn on_banned(&self, login_id: &str, login_type: &str) {
459        let _ = (login_id, login_type);
460    }
461
462    /// 解封事件 | Unbanned Event
463    ///
464    /// # 参数 | Parameters
465    /// - `login_id`: 登录 ID | Login ID
466    /// - `service`: 解封的服务标识 | Service identifier that was unbanned
467    /// - `login_type`: 登录类型 | Login type
468    async fn on_unbanned(&self, login_id: &str, service: &str, login_type: &str) {
469        let _ = (login_id, service, login_type);
470    }
471
472    /// 开启二级认证 | Open Safe Authentication
473    async fn on_open_safe(&self, token: &str, service: &str) {
474        let _ = (token, service);
475    }
476
477    /// 关闭二级认证 | Close Safe Authentication
478    async fn on_close_safe(&self, token: &str, service: &str) {
479        let _ = (token, service);
480    }
481
482    /// 二级认证校验通过 | Safe Verification Passed
483    ///
484    /// # 参数 | Parameters
485    /// - `token`: Token 值 | Token value
486    /// - `service`: 业务标识 | Service identifier
487    async fn on_safe_verify(&self, token: &str, service: &str) {
488        let _ = (token, service);
489    }
490
491    /// 权限/角色变更事件 | Grant Changed Event
492    async fn on_grant_changed(&self, login_id: &str, login_type: &str) {
493        let _ = (login_id, login_type);
494    }
495
496    /// 通用事件处理（所有事件都会触发此方法）
497    /// Generic Event Handler (triggered by all events)
498    async fn on_event(&self, event: &SaTokenEvent) {
499        let _ = event;
500    }
501}
502
503/// Listener list snapshot type for publish (Arc clone of Arc<Vec>).
504/// publish 用的监听器快照类型（对 Arc<Vec> 做 Arc clone）。
505type ListenerList = Arc<Vec<Arc<dyn SaTokenListener>>>;
506
507/// 事件总线 - 管理所有监听器并分发事件
508///
509/// Event bus - manages all listeners and dispatches events.
510///
511/// 列表用内层 `Arc<Vec>` 做 publish 快照；配置保持值字段。
512/// Inner `Arc<Vec>` makes publish a pointer snapshot; config stays a value field.
513#[derive(Clone)]
514pub struct SaTokenEventBus {
515    listeners: Arc<RwLock<ListenerList>>,
516    config: EventBusConfig,
517}
518
519impl std::fmt::Debug for SaTokenEventBus {
520    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
521        f.write_str("SaTokenEventBus { .. }")
522    }
523}
524
525impl SaTokenEventBus {
526    /// 创建新的事件总线（默认配置）
527    ///
528    /// Creates a new event bus with default configuration.
529    pub fn new() -> Self {
530        Self::with_config(EventBusConfig::default())
531    }
532
533    /// 创建事件总线（自定义配置）
534    ///
535    /// Creates an event bus with custom configuration.
536    pub fn with_config(config: EventBusConfig) -> Self {
537        Self {
538            listeners: Arc::new(RwLock::new(Arc::new(Vec::new()))),
539            config,
540        }
541    }
542
543    /// 获取配置引用 | Get configuration reference
544    pub fn config(&self) -> &EventBusConfig {
545        &self.config
546    }
547
548    /// poison 时 into_inner 恢复，单个 listener unwind 不能卡住整条总线。
549    /// Recover from a poisoned lock so one unwind cannot jam the bus.
550    fn read_guard(&self) -> std::sync::RwLockReadGuard<'_, Arc<Vec<Arc<dyn SaTokenListener>>>> {
551        self.listeners.read().unwrap_or_else(|poisoned| {
552            tracing::warn!("EventBus RwLock poisoned, recovering");
553            poisoned.into_inner()
554        })
555    }
556
557    fn write_guard(&self) -> std::sync::RwLockWriteGuard<'_, Arc<Vec<Arc<dyn SaTokenListener>>>> {
558        self.listeners.write().unwrap_or_else(|poisoned| {
559            tracing::warn!("EventBus RwLock poisoned during write, recovering");
560            poisoned.into_inner()
561        })
562    }
563
564    /// 热路径：拷贝表指针后立即放锁，供后续 await 使用。
565    /// Hot path: clone the table pointer then drop the lock before any `.await`.
566    fn snapshot(&self) -> Arc<Vec<Arc<dyn SaTokenListener>>> {
567        Arc::clone(&*self.read_guard())
568    }
569
570    /// 注册监听器 | Registers a listener.
571    pub fn register(&self, listener: Arc<dyn SaTokenListener>) {
572        let mut guard = self.write_guard();
573        let mut next = Vec::with_capacity(guard.len() + 1);
574        next.extend(guard.iter().cloned());
575        next.push(listener);
576        *guard = Arc::new(next);
577    }
578
579    /// 异步注册监听器（为了保持 API 兼容性）
580    ///
581    /// Registers a listener asynchronously (for API compatibility).
582    pub async fn register_async(&self, listener: Arc<dyn SaTokenListener>) {
583        self.register(listener);
584    }
585
586    /// 移除所有监听器 | Clears all listeners.
587    pub fn clear(&self) {
588        *self.write_guard() = Arc::new(Vec::new());
589    }
590
591    /// 只读长度。Arc&lt;Vec&gt; Deref 到 Vec，不必为 count 克隆表。
592    /// Read `len` via Deref; never clone the vec just to count.
593    pub fn listener_count(&self) -> usize {
594        self.read_guard().len()
595    }
596
597    /// 发布事件（按 DispatchMode 分发）
598    ///
599    /// Publishes an event (dispatches according to DispatchMode).
600    pub async fn publish(&self, event: SaTokenEvent) {
601        match self.config.dispatch_mode {
602            DispatchMode::Sequential => {
603                self.dispatch_sequential(event).await;
604            }
605            DispatchMode::Concurrent => {
606                self.dispatch_concurrent(event).await;
607            }
608            DispatchMode::Detached => {
609                let bus = self.clone();
610                tokio::spawn(async move {
611                    bus.dispatch_sequential(event).await;
612                });
613            }
614        }
615    }
616
617    /// 顺序分发（超时 + panic 隔离）
618    ///
619    /// Sequential dispatch (timeout + panic isolation).
620    async fn dispatch_sequential(&self, event: SaTokenEvent) {
621        let listeners = self.snapshot();
622        let timeout = self.config.listener_timeout;
623        for listener in listeners.iter() {
624            Self::invoke_listener_safe(Arc::clone(listener), &event, timeout).await;
625        }
626    }
627
628    /// 并行分发（检查 JoinError）
629    ///
630    /// Concurrent dispatch (checks JoinError).
631    async fn dispatch_concurrent(&self, event: SaTokenEvent) {
632        let listeners = self.snapshot();
633        let timeout = self.config.listener_timeout;
634        let mut handles = Vec::with_capacity(listeners.len());
635
636        for listener in listeners.iter() {
637            let listener = Arc::clone(listener);
638            let ev = event.clone();
639            let handle = tokio::spawn(async move {
640                Self::invoke_listener_safe(listener, &ev, timeout).await;
641            });
642            handles.push(handle);
643        }
644
645        for (idx, handle) in handles.into_iter().enumerate() {
646            if let Err(e) = handle.await {
647                if e.is_panic() {
648                    tracing::warn!(
649                        listener_idx = idx,
650                        "listener task panicked in concurrent mode"
651                    );
652                } else {
653                    tracing::warn!(listener_idx = idx, "listener task cancelled");
654                }
655            }
656        }
657    }
658
659    /// 单监听器安全调用（spawn 隔离 panic + timeout 保护）
660    ///
661    /// Safe invocation of a single listener (spawn isolates panic + timeout).
662    async fn invoke_listener_safe(
663        listener: Arc<dyn SaTokenListener>,
664        event: &SaTokenEvent,
665        timeout: Option<Duration>,
666    ) {
667        let event_owned = event.clone();
668        let handle = tokio::spawn(async move {
669            let fut = Self::dispatch_to_listener(&listener, &event_owned);
670            match timeout {
671                Some(d) => match tokio::time::timeout(d, fut).await {
672                    Ok(()) => Ok(()),
673                    Err(_elapsed) => Err("timeout"),
674                },
675                None => {
676                    fut.await;
677                    Ok(())
678                }
679            }
680        });
681
682        match handle.await {
683            Ok(Ok(())) => {}
684            Ok(Err("timeout")) => {
685                tracing::warn!(
686                    event_type = ?event.event_type,
687                    "listener timed out during event dispatch"
688                );
689            }
690            Ok(Err(_)) => {}
691            Err(e) if e.is_panic() => {
692                tracing::warn!(
693                    event_type = ?event.event_type,
694                    "listener panicked during event dispatch"
695                );
696            }
697            Err(e) => {
698                tracing::warn!("listener task cancelled: {:?}", e);
699            }
700        }
701    }
702
703    /// 分发事件到单个监听器（on_event + typed 方法）
704    ///
705    /// Dispatches an event to a single listener (on_event + typed method).
706    async fn dispatch_to_listener(listener: &Arc<dyn SaTokenListener>, event: &SaTokenEvent) {
707        listener.on_event(event).await;
708
709        match event.event_type {
710            SaTokenEventType::Login => {
711                listener
712                    .on_login(&event.login_id, &event.token, &event.login_type)
713                    .await;
714            }
715            SaTokenEventType::Logout => {
716                listener
717                    .on_logout(&event.login_id, &event.token, &event.login_type)
718                    .await;
719            }
720            SaTokenEventType::KickOut => {
721                listener
722                    .on_kick_out(&event.login_id, &event.token, &event.login_type)
723                    .await;
724            }
725            SaTokenEventType::RenewTimeout => {
726                let timeout_seconds = event
727                    .extra
728                    .as_ref()
729                    .and_then(|v| v.get("timeout_seconds"))
730                    .and_then(|v| v.as_i64())
731                    .unwrap_or(0);
732                listener
733                    .on_renew_timeout(
734                        &event.login_id,
735                        &event.token,
736                        &event.login_type,
737                        timeout_seconds,
738                    )
739                    .await;
740            }
741            SaTokenEventType::Replaced => {
742                listener
743                    .on_replaced(&event.login_id, &event.token, &event.login_type)
744                    .await;
745            }
746            SaTokenEventType::Banned => {
747                listener.on_banned(&event.login_id, &event.login_type).await;
748            }
749            SaTokenEventType::Unbanned => {
750                let service = event
751                    .extra
752                    .as_ref()
753                    .and_then(|v| v.get("service"))
754                    .and_then(|v| v.as_str())
755                    .unwrap_or("");
756                listener
757                    .on_unbanned(&event.login_id, service, &event.login_type)
758                    .await;
759            }
760            SaTokenEventType::OpenSafe => {
761                let service = event
762                    .extra
763                    .as_ref()
764                    .and_then(|v| v.get("service"))
765                    .and_then(|v| v.as_str())
766                    .unwrap_or(&event.login_type);
767                listener.on_open_safe(&event.token, service).await;
768            }
769            SaTokenEventType::CloseSafe => {
770                let service = event
771                    .extra
772                    .as_ref()
773                    .and_then(|v| v.get("service"))
774                    .and_then(|v| v.as_str())
775                    .unwrap_or(&event.login_type);
776                listener.on_close_safe(&event.token, service).await;
777            }
778            SaTokenEventType::SafeVerify => {
779                let service = event
780                    .extra
781                    .as_ref()
782                    .and_then(|v| v.get("service"))
783                    .and_then(|v| v.as_str())
784                    .unwrap_or("");
785                listener.on_safe_verify(&event.token, service).await;
786            }
787            SaTokenEventType::GrantChanged => {
788                listener
789                    .on_grant_changed(&event.login_id, &event.login_type)
790                    .await;
791            }
792        }
793    }
794}
795
796impl Default for SaTokenEventBus {
797    fn default() -> Self {
798        Self::new()
799    }
800}
801
802/// 简单的日志监听器示例 | Simple logging listener example
803pub struct LoggingListener;
804
805#[async_trait]
806impl SaTokenListener for LoggingListener {
807    async fn on_login(&self, login_id: &str, token: &str, login_type: &str) {
808        tracing::info!(
809            login_id = %login_id,
810            token = %token,
811            login_type = %login_type,
812            "用户登录"
813        );
814    }
815
816    async fn on_logout(&self, login_id: &str, token: &str, login_type: &str) {
817        tracing::info!(
818            login_id = %login_id,
819            token = %token,
820            login_type = %login_type,
821            "用户登出"
822        );
823    }
824
825    async fn on_kick_out(&self, login_id: &str, token: &str, login_type: &str) {
826        tracing::warn!(
827            login_id = %login_id,
828            token = %token,
829            login_type = %login_type,
830            "用户被踢出下线"
831        );
832    }
833
834    async fn on_renew_timeout(
835        &self,
836        login_id: &str,
837        token: &str,
838        login_type: &str,
839        timeout_seconds: i64,
840    ) {
841        tracing::debug!(
842            login_id = %login_id,
843            token = %token,
844            login_type = %login_type,
845            timeout_seconds = timeout_seconds,
846            "Token 续期"
847        );
848    }
849
850    async fn on_replaced(&self, login_id: &str, token: &str, login_type: &str) {
851        tracing::warn!(
852            login_id = %login_id,
853            token = %token,
854            login_type = %login_type,
855            "用户被顶下线"
856        );
857    }
858
859    async fn on_banned(&self, login_id: &str, login_type: &str) {
860        tracing::warn!(
861            login_id = %login_id,
862            login_type = %login_type,
863            "用户被封禁"
864        );
865    }
866
867    async fn on_unbanned(&self, login_id: &str, service: &str, login_type: &str) {
868        tracing::info!(
869            login_id = %login_id,
870            service = %service,
871            login_type = %login_type,
872            "用户被解封"
873        );
874    }
875
876    async fn on_safe_verify(&self, token: &str, service: &str) {
877        tracing::debug!(
878            token = %token,
879            service = %service,
880            "二级认证校验通过"
881        );
882    }
883}
884
885impl std::fmt::Debug for LoggingListener {
886    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
887        f.write_str("LoggingListener { .. }")
888    }
889}
890
891#[cfg(test)]
892mod tests {
893    use super::*;
894
895    struct TestListener {
896        login_count: Arc<RwLock<i32>>,
897    }
898
899    impl TestListener {
900        fn new() -> Self {
901            Self {
902                login_count: Arc::new(RwLock::new(0)),
903            }
904        }
905    }
906
907    #[async_trait]
908    impl SaTokenListener for TestListener {
909        async fn on_login(&self, _login_id: &str, _token: &str, _login_type: &str) {
910            let mut count = self.login_count.write().unwrap();
911            *count += 1;
912        }
913    }
914
915    #[tokio::test]
916    async fn test_event_bus() {
917        let bus = SaTokenEventBus::with_config(EventBusConfig::no_timeout());
918        let listener = Arc::new(TestListener::new());
919        let login_count = Arc::clone(&listener.login_count);
920
921        bus.register(listener);
922
923        let event = SaTokenEvent::login("user_123", "token_abc");
924        bus.publish(event).await;
925
926        let count = login_count.read().unwrap();
927        assert_eq!(*count, 1);
928    }
929
930    #[test]
931    fn test_event_creation() {
932        let event = SaTokenEvent::login("user_123", "token_abc");
933        assert_eq!(event.event_type, SaTokenEventType::Login);
934        assert_eq!(event.login_id, "user_123");
935        assert_eq!(event.token, "token_abc");
936    }
937}
sa_token_core/event/mod.rs

sa_token_core/event/
mod.rs
