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    /// 连接总数的同步快照（用于同步 stats 路径，禁止阻塞/网络）。
100    ///
101    /// 实现必须以缓存/原子计数等廉价方式返回，分布式实现可返回最近一次快照。
102    fn connection_count_snapshot(&self) -> usize;
103
104    /// 绑定用户数的同步快照（约束同 [`connection_count_snapshot`]）。
105    fn user_count_snapshot(&self) -> usize;
106
107    /// 清理超时连接
108    async fn cleanup_timeout_connections(&self, timeout: Duration) -> Vec<String>;
109
110    // ========== 底层发送方法（字节数组）==========
111
112    /// 向指定连接发送数据（字节数组）
113    async fn send_to_connection(&self, connection_id: &str, data: &[u8]) -> Result<()>;
114
115    /// 向指定用户的所有连接发送数据（字节数组）
116    async fn send_to_user(&self, user_id: &str, data: &[u8]) -> Result<()>;
117
118    /// 广播消息到所有连接（字节数组）
119    async fn broadcast(&self, data: &[u8]) -> Result<()>;
120
121    /// 广播消息到所有连接，排除指定连接（字节数组）
122    async fn broadcast_except(&self, data: &[u8], exclude_connection_id: &str) -> Result<()>;
123
124    // ========== Frame 级别发送方法（需要 MessageParser）==========
125
126    /// 向指定连接发送 Frame（自动序列化）
127    ///
128    /// # 参数
129    /// - `connection_id`: 连接 ID
130    /// - `frame`: 要发送的 Frame
131    /// - `parser`: 消息解析器，用于序列化 Frame（如果为 None，则从连接的协商信息创建）
132    ///
133    /// # 返回
134    /// 发送成功返回 `Ok(())`，失败返回错误
135    ///
136    /// # 注意
137    /// 如果 parser 为 None，将从连接的 ConnectionInfo 中获取协商后的序列化格式和压缩算法创建 parser
138    async fn send_frame_to(
139        &self,
140        connection_id: &str,
141        frame: &Frame,
142        parser: Option<&MessageParser>,
143    ) -> Result<()> {
144        let connection_info = self.get_connection(connection_id).await;
145
146        if let Some((_, info)) = &connection_info {
147            if !info.authenticated {
148                let is_system_command = frame.command.as_ref().and_then(|cmd| {
149                    if let Some(crate::common::protocol::flare::core::commands::command::Type::System(sys_cmd)) = &cmd.r#type {
150                        Some(sys_cmd.r#type == crate::common::protocol::flare::core::commands::system_command::Type::ConnectAck as i32
151                            || sys_cmd.r#type == crate::common::protocol::flare::core::commands::system_command::Type::Ping as i32
152                            || sys_cmd.r#type == crate::common::protocol::flare::core::commands::system_command::Type::Pong as i32
153                            || sys_cmd.r#type == crate::common::protocol::flare::core::commands::system_command::Type::Error as i32
154                            || sys_cmd.r#type == crate::common::protocol::flare::core::commands::system_command::Type::Close as i32)
155                    } else {
156                        None
157                    }
158                }).unwrap_or(false);
159
160                if !is_system_command {
161                    return Err(crate::common::error::FlareError::authentication_failed(
162                        format!("连接 {} 未验证，无法发送消息", connection_id),
163                    ));
164                }
165            }
166        } else {
167            return Err(crate::common::error::FlareError::connection_failed(
168                format!("连接 {} 不存在", connection_id),
169            ));
170        }
171
172        let data = if let Some(p) = parser {
173            p.serialize(frame)?
174        } else if let Some((_, info)) = connection_info {
175            let connection_parser =
176                MessageParser::new(info.serialization_format, info.compression, info.encryption);
177            connection_parser.serialize(frame)?
178        } else {
179            use crate::common::message::parser::PRE_NEGOTIATION_PARSER;
180            PRE_NEGOTIATION_PARSER.serialize(frame)?
181        };
182
183        self.send_to_connection(connection_id, &data).await?;
184        self.update_connection_active(connection_id).await?;
185        Ok(())
186    }
187
188    /// 向指定用户的所有连接发送 Frame（自动序列化）
189    ///
190    /// # 参数
191    /// - `user_id`: 用户 ID
192    /// - `frame`: 要发送的 Frame
193    /// - `parser`: 消息解析器，用于序列化 Frame（如果为 None，则为每个连接使用其协商的格式）
194    ///
195    /// # 返回
196    /// 发送成功返回 `Ok(())`，失败返回错误
197    async fn send_frame_to_user(
198        &self,
199        user_id: &str,
200        frame: &Frame,
201        parser: Option<&MessageParser>,
202    ) -> Result<()> {
203        let connection_ids = self.get_user_connections(user_id).await;
204        for conn_id in connection_ids {
205            // 为每个连接使用其协商的格式（如果 parser 为 None）
206            let _ = self.send_frame_to(&conn_id, frame, parser).await;
207        }
208        Ok(())
209    }
210
211    /// 广播 Frame 到所有连接（自动序列化）
212    ///
213    /// # 参数
214    /// - `frame`: 要广播的 Frame
215    /// - `parser`: 消息解析器，用于序列化 Frame（如果为 None，则为每个连接使用其协商的格式）
216    ///
217    /// # 返回
218    /// 广播成功返回 `Ok(())`，失败返回错误
219    async fn broadcast_frame(&self, frame: &Frame, parser: Option<&MessageParser>) -> Result<()> {
220        let connection_ids = self.list_connections().await;
221        for conn_id in connection_ids {
222            // 为每个连接使用其协商的格式（如果 parser 为 None）
223            let _ = self.send_frame_to(&conn_id, frame, parser).await;
224        }
225        Ok(())
226    }
227
228    /// 广播 Frame 到所有连接，排除指定连接（自动序列化）
229    ///
230    /// # 参数
231    /// - `frame`: 要广播的 Frame
232    /// - `exclude_connection_id`: 要排除的连接 ID
233    /// - `parser`: 消息解析器，用于序列化 Frame（如果为 None，则为每个连接使用其协商的格式）
234    ///
235    /// # 返回
236    /// 广播成功返回 `Ok(())`，失败返回错误
237    async fn broadcast_frame_except(
238        &self,
239        frame: &Frame,
240        exclude_connection_id: &str,
241        parser: Option<&MessageParser>,
242    ) -> Result<()> {
243        let connection_ids = self.list_connections().await;
244        for conn_id in connection_ids {
245            if conn_id != exclude_connection_id {
246                // 为每个连接使用其协商的格式（如果 parser 为 None）
247                let _ = self.send_frame_to(&conn_id, frame, parser).await;
248            }
249        }
250        Ok(())
251    }
252}
253
254/// 连接统计信息
255#[derive(Debug, Clone)]
256pub struct ConnectionStats {
257    /// 总连接数
258    pub total_connections: usize,
259    /// 总用户数
260    pub total_users: usize,
261}
flare_core/server/connection/trait.rs

flare_core/server/connection/
trait.rs
