sa_token_core/online/
mod.rs

1// Author: 金书记 | Author: Jin Shuji
2//! Online users and realtime push.
3//! 在线用户与实时推送。
4
5use std::collections::HashMap;
6use std::sync::Arc;
7use std::time::Duration;
8
9use async_trait::async_trait;
10use chrono::{DateTime, Utc};
11use tokio::sync::RwLock;
12
13use crate::dao::SaTokenDao;
14use crate::error::{SaTokenError, SaTokenResult};
15use crate::keys::LOGIN_TYPE_DEFAULT;
16
17mod push;
18mod store;
19
20pub use push::dispatch_to_pushers;
21pub use store::{DistributedOnlineStore, LocalOnlineStore, OnlineStore, StoredOnlineUser};
22
23/// One live connection belonging to a login id.
24/// 某个登录账号下的一条在线连接。
25#[derive(Debug, Clone)]
26pub struct OnlineUser {
27    /// Account system; empty/`login` means default.
28    /// 账号体系；空或 `login` 表示默认。
29    pub login_type: String,
30    /// Login id | 登录 ID
31    pub login_id: String,
32    /// Token value | Token 值
33    pub token: String,
34    /// Device / terminal label | 设备/终端标识
35    pub device: String,
36    /// First connect time | 首次连接时间
37    pub connect_time: DateTime<Utc>,
38    /// Last activity time | 最近活跃时间
39    pub last_activity: DateTime<Utc>,
40    /// Extra key-value metadata | 扩展元数据
41    pub metadata: HashMap<String, String>,
42}
43
44impl OnlineUser {
45    /// Build a presence record for the default account system.
46    /// 为默认账号体系构造一条 presence 记录。
47    pub fn new(
48        login_id: impl Into<String>,
49        token: impl Into<String>,
50        device: impl Into<String>,
51    ) -> Self {
52        let now = Utc::now();
53        Self {
54            login_type: LOGIN_TYPE_DEFAULT.to_string(),
55            login_id: login_id.into(),
56            token: token.into(),
57            device: device.into(),
58            connect_time: now,
59            last_activity: now,
60            metadata: HashMap::new(),
61        }
62    }
63}
64
65/// Push payload | 推送载荷
66#[derive(Debug, Clone)]
67pub struct PushMessage {
68    /// Message id | 消息 ID
69    pub message_id: String,
70    /// Message body | 消息正文
71    pub content: String,
72    /// Message kind | 消息类型
73    pub message_type: MessageType,
74    /// Event timestamp | 事件时间戳
75    pub timestamp: DateTime<Utc>,
76    /// Extra key-value metadata | 扩展元数据
77    pub metadata: HashMap<String, String>,
78}
79
80/// Message kind | 消息种类
81#[derive(Debug, Clone, PartialEq)]
82pub enum MessageType {
83    /// Plain text | 纯文本
84    Text,
85    /// Binary payload | 二进制载荷
86    Binary,
87    /// Kick-out signal | 踢下线信号
88    KickOut,
89    /// Notification | 通知
90    Notification,
91    /// Custom message type | 自定义消息类型
92    Custom(String),
93}
94
95/// Deliver a message to one user.
96/// 向单个用户投递消息。
97#[async_trait]
98pub trait MessagePusher: Send + Sync {
99    /// Push a message to the user | 向用户推送消息
100    async fn push(&self, login_id: &str, message: PushMessage) -> Result<(), SaTokenError>;
101}
102
103/// Online user manager | 在线用户管理器
104pub struct OnlineManager {
105    store: Arc<dyn OnlineStore>,
106    pushers: Arc<RwLock<Vec<Arc<dyn MessagePusher>>>>,
107}
108
109impl std::fmt::Debug for OnlineManager {
110    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
111        f.write_str("OnlineManager { .. }")
112    }
113}
114
115impl OnlineManager {
116    /// Process-local (backward compatible).
117    /// 进程内实现（保持旧 `new()` 语义）。
118    pub fn new() -> Self {
119        Self::local()
120    }
121
122    /// Process-local store | 进程内存储
123    pub fn local() -> Self {
124        Self {
125            store: Arc::new(LocalOnlineStore::new()),
126            pushers: Arc::new(RwLock::new(Vec::new())),
127        }
128    }
129
130    /// Shared store; `entry_ttl` of 1 day bounds leaked WS rows.
131    /// 共享存储；默认 1 天 TTL 限制异常断开造成的泄漏。
132    pub fn distributed(dao: Arc<SaTokenDao>) -> Self {
133        Self {
134            store: Arc::new(DistributedOnlineStore::new(
135                dao,
136                Some(Duration::from_secs(86400)),
137            )),
138            pushers: Arc::new(RwLock::new(Vec::new())),
139        }
140    }
141
142    /// Build with a custom [`OnlineStore`] | 使用自定义在线存储构建
143    pub fn with_store(store: Arc<dyn OnlineStore>) -> Self {
144        Self {
145            store,
146            pushers: Arc::new(RwLock::new(Vec::new())),
147        }
148    }
149
150    /// Register a realtime pusher | 注册实时推送器
151    pub async fn register_pusher(&self, pusher: Arc<dyn MessagePusher>) {
152        self.pushers.write().await.push(pusher);
153    }
154
155    /// Mark a user connection online | 标记用户连接在线
156    pub async fn mark_online(&self, user: OnlineUser) -> SaTokenResult<()> {
157        self.store.mark_online(user).await
158    }
159
160    /// Default account-system wrapper (old two-arg API).
161    /// 默认账号体系包装（旧两参数 API）。
162    pub async fn mark_offline(&self, login_id: &str, token: &str) -> SaTokenResult<()> {
163        self.store
164            .mark_offline(LOGIN_TYPE_DEFAULT, login_id, token)
165            .await
166    }
167
168    /// Mark offline for a login type | 按登录类型标记离线
169    pub async fn mark_offline_with_type(
170        &self,
171        login_type: &str,
172        login_id: &str,
173        token: &str,
174    ) -> SaTokenResult<()> {
175        self.store.mark_offline(login_type, login_id, token).await
176    }
177
178    /// Mark all connections offline | 标记该账号全部离线
179    pub async fn mark_offline_all(&self, login_id: &str) -> SaTokenResult<()> {
180        self.store
181            .mark_offline_all(LOGIN_TYPE_DEFAULT, login_id)
182            .await
183    }
184
185    /// Mark all offline for a login type | 按登录类型全部离线
186    pub async fn mark_offline_all_with_type(
187        &self,
188        login_type: &str,
189        login_id: &str,
190    ) -> SaTokenResult<()> {
191        self.store.mark_offline_all(login_type, login_id).await
192    }
193
194    /// Whether the login id is online | 登录 ID 是否在线
195    pub async fn is_online(&self, login_id: &str) -> SaTokenResult<bool> {
196        self.store.is_online(LOGIN_TYPE_DEFAULT, login_id).await
197    }
198
199    /// `get_online_count` — get online count | `get_online_count`
200    pub async fn get_online_count(&self) -> SaTokenResult<usize> {
201        self.store.get_online_count().await
202    }
203
204    /// List online users for a login id | 列出某登录 ID 的在线用户
205    pub async fn get_online_users(&self) -> SaTokenResult<Vec<String>> {
206        self.store.get_online_users().await
207    }
208
209    /// `get_user_sessions` — get user sessions | `get_user_sessions`
210    pub async fn get_user_sessions(&self, login_id: &str) -> SaTokenResult<Vec<OnlineUser>> {
211        self.store
212            .get_user_sessions(LOGIN_TYPE_DEFAULT, login_id)
213            .await
214    }
215
216    /// Refresh last-activity timestamp | 刷新最近活跃时间
217    pub async fn update_activity(&self, login_id: &str, token: &str) -> SaTokenResult<()> {
218        self.store
219            .update_activity(LOGIN_TYPE_DEFAULT, login_id, token)
220            .await
221    }
222
223    /// `update_activity_with_type` — update activity with type | `update_activity_with_type`
224    pub async fn update_activity_with_type(
225        &self,
226        login_type: &str,
227        login_id: &str,
228        token: &str,
229    ) -> SaTokenResult<()> {
230        self.store
231            .update_activity(login_type, login_id, token)
232            .await
233    }
234
235    async fn cloned_pushers(&self) -> Vec<Arc<dyn MessagePusher>> {
236        self.pushers.read().await.clone()
237    }
238
239    /// `push_to_user` — push to user | `push_to_user`
240    pub async fn push_to_user(&self, login_id: &str, content: String) -> SaTokenResult<()> {
241        let message = PushMessage {
242            message_id: uuid::Uuid::new_v4().to_string(),
243            content,
244            message_type: MessageType::Text,
245            timestamp: Utc::now(),
246            metadata: HashMap::new(),
247        };
248        let pushers = self.cloned_pushers().await;
249        dispatch_to_pushers(&pushers, login_id, message).await
250    }
251
252    /// `push_to_users` — push to users | `push_to_users`
253    pub async fn push_to_users(
254        &self,
255        login_ids: Vec<String>,
256        content: String,
257    ) -> SaTokenResult<()> {
258        for login_id in login_ids {
259            self.push_to_user(&login_id, content.clone()).await?;
260        }
261        Ok(())
262    }
263
264    /// Broadcast to all online users | 向全部在线用户广播
265    pub async fn broadcast(&self, content: String) -> SaTokenResult<()> {
266        let login_ids = self.get_online_users().await?;
267        self.push_to_users(login_ids, content).await
268    }
269
270    /// `push_message_to_user` — push message to user | `push_message_to_user`
271    pub async fn push_message_to_user(
272        &self,
273        login_id: &str,
274        message: PushMessage,
275    ) -> SaTokenResult<()> {
276        let pushers = self.cloned_pushers().await;
277        dispatch_to_pushers(&pushers, login_id, message).await
278    }
279
280    /// `kick_out_notify` — kick out notify | `kick_out_notify`
281    pub async fn kick_out_notify(&self, login_id: &str, reason: String) -> SaTokenResult<()> {
282        let message = PushMessage {
283            message_id: uuid::Uuid::new_v4().to_string(),
284            content: reason,
285            message_type: MessageType::KickOut,
286            timestamp: Utc::now(),
287            metadata: HashMap::new(),
288        };
289        self.push_message_to_user(login_id, message).await?;
290        self.mark_offline_all(login_id).await
291    }
292}
293
294impl Default for OnlineManager {
295    fn default() -> Self {
296        Self::new()
297    }
298}
299
300/// In-memory pusher for development.
301/// 开发用内存推送器。
302pub struct InMemoryPusher {
303    messages: Arc<RwLock<HashMap<String, Vec<PushMessage>>>>,
304}
305
306impl std::fmt::Debug for InMemoryPusher {
307    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
308        f.write_str("InMemoryPusher { .. }")
309    }
310}
311
312impl InMemoryPusher {
313    /// Create a new instance | 创建新实例
314    pub fn new() -> Self {
315        Self {
316            messages: Arc::new(RwLock::new(HashMap::new())),
317        }
318    }
319
320    /// `get_messages` — get messages | `get_messages`
321    pub async fn get_messages(&self, login_id: &str) -> Vec<PushMessage> {
322        self.messages
323            .read()
324            .await
325            .get(login_id)
326            .cloned()
327            .unwrap_or_default()
328    }
329
330    /// Clear buffered messages for a login id (dev / tests).
331    /// 清空某登录账号的缓冲消息（开发 / 测试）。
332    pub async fn clear_messages(&self, login_id: &str) {
333        self.messages.write().await.remove(login_id);
334    }
335}
336
337impl Default for InMemoryPusher {
338    fn default() -> Self {
339        Self::new()
340    }
341}
342
343#[async_trait]
344impl MessagePusher for InMemoryPusher {
345    async fn push(&self, login_id: &str, message: PushMessage) -> Result<(), SaTokenError> {
346        self.messages
347            .write()
348            .await
349            .entry(login_id.to_string())
350            .or_default()
351            .push(message);
352        Ok(())
353    }
354}
sa_token_core/online/mod.rs

sa_token_core/online/
mod.rs
