flare_core/server/connection/
trait.rs

1//! 服务端连接管理器抽象
2//!
3//! 定义服务端连接管理的标准接口，支持用户自定义实现
4//! 默认实现使用 ConnectionManager
5
6use crate::common::MessageParser;
7use crate::common::error::Result;
8use crate::common::protocol::Frame;
9use crate::transport::connection::Connection;
10use async_trait::async_trait;
11use std::sync::Arc;
12use std::time::Duration;
13use tokio::sync::Mutex;
14
15/// 连接信息（Trait 版本，用于跨异步边界传递）
16#[derive(Clone)]
17pub struct ConnectionInfo {
18    /// 连接 ID（唯一标识符）
19    pub connection_id: String,
20    /// 用户 ID（如果已认证）
21    pub user_id: Option<String>,
22    /// 创建时间戳（Unix 时间戳，秒）
23    pub created_at: u64,
24    /// 最后活跃时间戳（Unix 时间戳，秒）
25    pub last_active: u64,
26    /// 连接元数据
27    pub metadata: std::collections::HashMap<String, String>,
28    /// 设备信息（如果已提供）
29    pub device_info: Option<crate::common::device::DeviceInfo>,
30    /// 序列化格式（由客户端协商决定）
31    pub serialization_format: crate::common::protocol::SerializationFormat,
32    /// 压缩算法（由客户端协商决定）
33    pub compression: crate::common::compression::CompressionAlgorithm,
34    /// 加密方式（由客户端协商决定）
35    pub encryption: crate::common::encryption::EncryptionAlgorithm,
36    /// 是否已验证（如果启用认证，只有已验证的连接才能收发消息）
37    pub authenticated: bool,
38    /// 认证时间戳（Unix 时间戳，秒，如果已验证）
39    pub authenticated_at: Option<u64>,
40    /// 协商是否已完成（CONNECT 和 CONNECT_ACK 完成）
41    pub negotiation_completed: bool,
42    /// 协商是否已确认（客户端收到 CONNECT_ACK 后发送确认，服务端收到后标记）
43    /// 确认后，消息必须严格按照协商好的方式处理，不再容错
44    pub negotiation_confirmed: bool,
45    /// 缓存的 MessageParser（协商完成后创建，避免每次消息处理都创建新的 parser）
46    pub cached_parser: Option<std::sync::Arc<crate::common::MessageParser>>,
47    /// 缓存的 MessagePipeline（协商完成后创建，如果配置了中间件或处理器）
48    /// 可选，默认不创建以保持性能
49    pub cached_pipeline: Option<std::sync::Arc<crate::common::message::pipeline::MessagePipeline>>,
50}
51
52/// 连接管理器抽象 trait
53///
54/// 实现此 trait 以提供自定义的连接管理逻辑
55/// 例如：基于 Redis 的分布式连接管理、基于数据库的持久化等
56#[async_trait]
57pub trait ConnectionManagerTrait: Send + Sync + std::any::Any {
58    /// 获取 Any 引用，用于类型向下转换
59    fn as_any(&self) -> &dyn std::any::Any;
60    /// 添加连接
61    async fn add_connection(
62        &self,
63        connection_id: String,
64        connection: Arc<Mutex<Box<dyn Connection>>>,
65        user_id: Option<String>,
66    ) -> Result<()>;
67
68    /// 移除连接
69    async fn remove_connection(&self, connection_id: &str) -> Result<()>;
70
71    /// 获取连接
72    async fn get_connection(
73        &self,
74        connection_id: &str,
75    ) -> Option<(Arc<Mutex<Box<dyn Connection>>>, ConnectionInfo)>;
76
77    /// 获取用户的所有连接 ID
78    async fn get_user_connections(&self, user_id: &str) -> Vec<String>;
79
80    /// 绑定用户到连接
81    async fn bind_user(&self, connection_id: &str, user_id: String) -> Result<()>;
82
83    /// 更新连接的最后活跃时间
84    async fn update_connection_active(&self, connection_id: &str) -> Result<()>;
85
86    /// 设置连接为已验证状态（认证通过后调用）
87    async fn set_connection_authenticated(
88        &self,
89        connection_id: &str,
90        user_id: Option<String>,
91    ) -> Result<()>;
92
93    /// 获取所有连接 ID
94    async fn list_connections(&self) -> Vec<String>;
95
96    /// 获取连接总数
97    async fn connection_count(&self) -> usize;
98
99    /// Returns a synchronous snapshot of the total connection count.
100    ///
101    /// Implementations must use cheap local state such as cached counters or atomics.
102    /// Distributed implementations may return their most recent local snapshot.
103    fn connection_count_snapshot(&self) -> usize;
104
105    /// Returns a synchronous user-count snapshot with the same constraints as
106    /// [`Self::connection_count_snapshot`].
107    fn user_count_snapshot(&self) -> usize;
108
109    /// 清理超时连接
110    async fn cleanup_timeout_connections(&self, timeout: Duration) -> Vec<String>;
111
112    // ========== 底层发送方法（字节数组）==========
113
114    /// 向指定连接发送数据（字节数组）
115    async fn send_to_connection(&self, connection_id: &str, data: &[u8]) -> Result<()>;
116
117    /// 向指定用户的所有连接发送数据（字节数组）
118    async fn send_to_user(&self, user_id: &str, data: &[u8]) -> Result<()>;
119
120    /// 广播消息到所有连接（字节数组）
121    async fn broadcast(&self, data: &[u8]) -> Result<()>;
122
123    /// 广播消息到所有连接，排除指定连接（字节数组）
124    async fn broadcast_except(&self, data: &[u8], exclude_connection_id: &str) -> Result<()>;
125
126    // ========== Frame 级别发送方法（需要 MessageParser）==========
127
128    /// 向指定连接发送 Frame（自动序列化）
129    ///
130    /// # 参数
131    /// - `connection_id`: 连接 ID
132    /// - `frame`: 要发送的 Frame
133    /// - `parser`: 消息解析器，用于序列化 Frame（如果为 None，则从连接的协商信息创建）
134    ///
135    /// # 返回
136    /// 发送成功返回 `Ok(())`，失败返回错误
137    ///
138    /// # 注意
139    /// 如果 parser 为 None，将从连接的 ConnectionInfo 中获取协商后的序列化格式和压缩算法创建 parser
140    async fn send_frame_to(
141        &self,
142        connection_id: &str,
143        frame: &Frame,
144        parser: Option<&MessageParser>,
145    ) -> Result<()> {
146        let connection_info = self.get_connection(connection_id).await;
147
148        if let Some((_, info)) = &connection_info {
149            if !info.authenticated {
150                let is_system_command = frame.command.as_ref().and_then(|cmd| {
151                    if let Some(crate::common::protocol::flare::core::commands::command::Type::System(sys_cmd)) = &cmd.r#type {
152                        Some(sys_cmd.r#type == crate::common::protocol::flare::core::commands::system_command::Type::ConnectAck as i32
153                            || sys_cmd.r#type == crate::common::protocol::flare::core::commands::system_command::Type::Ping as i32
154                            || sys_cmd.r#type == crate::common::protocol::flare::core::commands::system_command::Type::Pong as i32
155                            || sys_cmd.r#type == crate::common::protocol::flare::core::commands::system_command::Type::Error as i32
156                            || sys_cmd.r#type == crate::common::protocol::flare::core::commands::system_command::Type::Close as i32)
157                    } else {
158                        None
159                    }
160                }).unwrap_or(false);
161
162                if !is_system_command {
163                    return Err(crate::common::error::FlareError::authentication_failed(
164                        format!("连接 {} 未验证，无法发送消息", connection_id),
165                    ));
166                }
167            }
168        } else {
169            return Err(crate::common::error::FlareError::connection_failed(
170                format!("连接 {} 不存在", connection_id),
171            ));
172        }
173
174        let data = if let Some(p) = parser {
175            p.serialize(frame)?
176        } else if let Some((_, info)) = connection_info {
177            let connection_parser =
178                MessageParser::new(info.serialization_format, info.compression, info.encryption);
179            connection_parser.serialize(frame)?
180        } else {
181            use crate::common::message::parser::PRE_NEGOTIATION_PARSER;
182            PRE_NEGOTIATION_PARSER.serialize(frame)?
183        };
184
185        self.send_to_connection(connection_id, &data).await?;
186        self.update_connection_active(connection_id).await?;
187        Ok(())
188    }
189
190    /// 向指定用户的所有连接发送 Frame（自动序列化）
191    ///
192    /// # 参数
193    /// - `user_id`: 用户 ID
194    /// - `frame`: 要发送的 Frame
195    /// - `parser`: 消息解析器，用于序列化 Frame（如果为 None，则为每个连接使用其协商的格式）
196    ///
197    /// # 返回
198    /// 发送成功返回 `Ok(())`，失败返回错误
199    async fn send_frame_to_user(
200        &self,
201        user_id: &str,
202        frame: &Frame,
203        parser: Option<&MessageParser>,
204    ) -> Result<()> {
205        let connection_ids = self.get_user_connections(user_id).await;
206        for conn_id in connection_ids {
207            // 为每个连接使用其协商的格式（如果 parser 为 None）
208            let _ = self.send_frame_to(&conn_id, frame, parser).await;
209        }
210        Ok(())
211    }
212
213    /// 同一 Frame 发送到多个连接（群扇出主路径）。
214    ///
215    /// 默认逐连接（每连接自协商格式序列化）；连接管理器实现按
216    /// （序列化格式, 压缩）分组、每组序列化一次共享给组内无加密连接。
217    ///
218    /// # 返回
219    /// （成功连接数, 失败连接数）
220    async fn send_frame_to_connections(
221        &self,
222        connection_ids: &[String],
223        frame: &Frame,
224    ) -> (i32, i32) {
225        let mut success = 0i32;
226        let mut failure = 0i32;
227        for connection_id in connection_ids {
228            match self.send_frame_to(connection_id, frame, None).await {
229                Ok(()) => success += 1,
230                Err(_) => failure += 1,
231            }
232        }
233        (success, failure)
234    }
235
236    /// 广播 Frame 到所有连接（自动序列化）
237    ///
238    /// # 参数
239    /// - `frame`: 要广播的 Frame
240    /// - `parser`: 消息解析器，用于序列化 Frame（如果为 None，则为每个连接使用其协商的格式）
241    ///
242    /// # 返回
243    /// 广播成功返回 `Ok(())`，失败返回错误
244    async fn broadcast_frame(&self, frame: &Frame, parser: Option<&MessageParser>) -> Result<()> {
245        let connection_ids = self.list_connections().await;
246        for conn_id in connection_ids {
247            // 为每个连接使用其协商的格式（如果 parser 为 None）
248            let _ = self.send_frame_to(&conn_id, frame, parser).await;
249        }
250        Ok(())
251    }
252
253    /// 广播 Frame 到所有连接，排除指定连接（自动序列化）
254    ///
255    /// # 参数
256    /// - `frame`: 要广播的 Frame
257    /// - `exclude_connection_id`: 要排除的连接 ID
258    /// - `parser`: 消息解析器，用于序列化 Frame（如果为 None，则为每个连接使用其协商的格式）
259    ///
260    /// # 返回
261    /// 广播成功返回 `Ok(())`，失败返回错误
262    async fn broadcast_frame_except(
263        &self,
264        frame: &Frame,
265        exclude_connection_id: &str,
266        parser: Option<&MessageParser>,
267    ) -> Result<()> {
268        let connection_ids = self.list_connections().await;
269        for conn_id in connection_ids {
270            if conn_id != exclude_connection_id {
271                // 为每个连接使用其协商的格式（如果 parser 为 None）
272                let _ = self.send_frame_to(&conn_id, frame, parser).await;
273            }
274        }
275        Ok(())
276    }
277}
278
279/// 连接统计信息
280#[derive(Debug, Clone)]
281pub struct ConnectionStats {
282    /// 总连接数
283    pub total_connections: usize,
284    /// 总用户数
285    pub total_users: usize,
286}
flare_core/server/connection/trait.rs

flare_core/server/connection/
trait.rs
