flare_core/server/builder/
observer.rs

1//! 观察者模式服务端构建器
2//!
3//! 提供基本功能实现，使用 `ServerEventHandler` trait 处理消息和事件。
4//!
5//! ## 特点
6//! - ✅ **实现 `ServerEventHandler` trait（必需）**：提供细化的命令处理方法
7//! - ✅ **自动消息路由**：`ServerMessageWrapper` 自动将消息路由到对应的处理方法
8//! - ✅ **自动 ACK 处理**：框架自动处理 ACK 和错误响应
9//! - ✅ **设备管理**：支持设备冲突策略和多端管理
10//! - ✅ **认证机制**：支持 Token 认证
11//! - ✅ **连接管理**：支持共享连接管理器（多服务器实例）
12//!
13//! ## 适用场景
14//! - 需要自定义消息处理逻辑但不需要完整功能集
15//! - 需要设备管理和多端控制
16//! - 需要事件驱动的架构
17//! - 需要共享连接状态（多服务器实例）
18//!
19//! ## 架构说明
20//!
21//! 观察者模式基于 `HybridServer`，使用 `ServerMessageWrapper` 作为消息处理器。
22//! `ServerMessageWrapper` 自动将消息路由到 `ServerEventHandler` 的对应方法，
23//! 并处理 ACK 和错误响应。
24
25use crate::common::error::Result;
26use crate::common::protocol::Frame;
27use crate::server::HybridServer;
28use crate::server::builder::{BaseServerBuilderConfig, ServerWrapper};
29use crate::server::connection::ConnectionManager;
30use crate::server::handle::ServerHandle;
31use std::sync::Arc;
32use tracing::{error, info};
33
34/// 观察者模式服务端构建器
35///
36/// 提供基本功能实现，使用 `ServerEventHandler` trait 处理消息和事件。
37///
38/// ## 设计原则
39///
40/// - **公共逻辑统一处理**：基于 `HybridServer`，共享所有核心能力
41/// - **自动消息路由**：`ServerMessageWrapper` 自动将消息路由到 `ServerEventHandler` 的对应方法
42/// - **自动 ACK 处理**：如果 handler 返回 `None`，框架自动发送 ACK
43/// - **错误处理**：处理失败时自动发送错误 ACK，确保客户端能收到响应
44///
45/// ## 使用方式
46///
47/// 用户只需要实现 `ServerEventHandler` trait，框架会自动处理消息路由和 ACK。
48pub struct ObserverServerBuilder {
49    base: BaseServerBuilderConfig,
50    connection_manager: Option<Arc<ConnectionManager>>,
51    device_manager: Option<Arc<crate::server::device::DeviceManager>>,
52    event_handler: Arc<dyn crate::server::events::handler::ServerEventHandler>,
53}
54
55impl ObserverServerBuilder {
56    /// 创建新的观察者模式构建器
57    ///
58    /// # 参数
59    /// - `bind_address`: 绑定地址
60    /// - `event_handler`: 事件处理器（必须），用户只需要实现 `ServerEventHandler` 的 `handle_message` 方法即可
61    pub fn new(
62        bind_address: impl Into<String>,
63        event_handler: Arc<dyn crate::server::events::handler::ServerEventHandler>,
64    ) -> Self {
65        Self {
66            base: BaseServerBuilderConfig::new(bind_address),
67            connection_manager: None,
68            device_manager: None,
69            event_handler,
70        }
71    }
72
73    /// 设置认证器（如果启用认证，必须提供）
74    ///
75    /// 如果设置了认证器，还需要在配置中启用认证：
76    /// ```rust,ignore
77    /// .enable_auth()
78    /// .with_authenticator(authenticator)
79    /// ```
80    pub fn with_authenticator(
81        mut self,
82        authenticator: Arc<dyn crate::server::auth::Authenticator>,
83    ) -> Self {
84        self.base = self.base.with_authenticator(authenticator);
85        self
86    }
87
88    /// 启用认证
89    pub fn enable_auth(mut self) -> Self {
90        self.base = self.base.enable_auth();
91        self
92    }
93
94    /// 设置认证超时时间
95    pub fn with_auth_timeout(mut self, timeout: std::time::Duration) -> Self {
96        self.base = self.base.with_auth_timeout(timeout);
97        self
98    }
99
100    /// 设置设备管理器（用于设备冲突管理）
101    pub fn with_device_manager(
102        mut self,
103        device_manager: Arc<crate::server::device::DeviceManager>,
104    ) -> Self {
105        self.device_manager = Some(device_manager);
106        self
107    }
108
109    /// 设置连接管理器（可选，用于共享连接状态）
110    pub fn with_connection_manager(mut self, manager: Arc<ConnectionManager>) -> Self {
111        self.connection_manager = Some(manager);
112        self
113    }
114
115    /// 设置传输协议
116    pub fn with_protocol(
117        mut self,
118        protocol: crate::common::config_types::TransportProtocol,
119    ) -> Self {
120        self.base = self.base.with_protocol(protocol);
121        self
122    }
123
124    /// 启用多协议监听
125    pub fn with_protocols(
126        mut self,
127        protocols: Vec<crate::common::config_types::TransportProtocol>,
128    ) -> Self {
129        self.base = self.base.with_protocols(protocols);
130        self
131    }
132
133    /// 为特定协议设置监听地址
134    pub fn with_protocol_address(
135        mut self,
136        protocol: crate::common::config_types::TransportProtocol,
137        address: String,
138    ) -> Self {
139        self.base = self.base.with_protocol_address(protocol, address);
140        self
141    }
142
143    /// 设置最大连接数
144    pub fn with_max_connections(mut self, max: usize) -> Self {
145        self.base = self.base.with_max_connections(max);
146        self
147    }
148
149    /// 设置握手超时时间
150    pub fn with_handshake_timeout(mut self, timeout: std::time::Duration) -> Self {
151        self.base = self.base.with_handshake_timeout(timeout);
152        self
153    }
154
155    /// 设置最大并发握手数
156    pub fn with_max_handshake_concurrency(mut self, max: usize) -> Self {
157        self.base = self.base.with_max_handshake_concurrency(max);
158        self
159    }
160
161    /// 设置单次连接写入超时时间
162    pub fn with_write_timeout(mut self, timeout: std::time::Duration) -> Self {
163        self.base = self.base.with_write_timeout(timeout);
164        self
165    }
166
167    /// 设置 fanout 发送最大并发度
168    pub fn with_fanout_concurrency(mut self, max: usize) -> Self {
169        self.base = self.base.with_fanout_concurrency(max);
170        self
171    }
172
173    /// 设置心跳配置
174    pub fn with_heartbeat(
175        mut self,
176        heartbeat: crate::common::config_types::HeartbeatConfig,
177    ) -> Self {
178        self.base = self.base.with_heartbeat(heartbeat);
179        self
180    }
181
182    /// 设置 TLS 配置
183    pub fn with_tls(mut self, tls: crate::common::config_types::TlsConfig) -> Self {
184        self.base = self.base.with_tls(tls);
185        self
186    }
187
188    /// 设置默认序列化格式（用于协商，默认 Protobuf）
189    pub fn with_default_format(
190        mut self,
191        format: crate::common::protocol::SerializationFormat,
192    ) -> Self {
193        self.base = self.base.with_default_format(format);
194        self
195    }
196
197    /// 设置默认压缩算法（用于协商，默认 None）
198    pub fn with_default_compression(
199        mut self,
200        compression: crate::common::compression::CompressionAlgorithm,
201    ) -> Self {
202        self.base = self.base.with_default_compression(compression);
203        self
204    }
205
206    /// 构建服务端
207    ///
208    /// # 错误处理
209    /// - 如果配置无效（如启用了认证但未提供认证器），返回配置错误
210    /// - 如果服务器初始化失败，返回相应的错误
211    ///
212    /// # 返回
213    /// - `Ok(ObserverServer)` - 成功构建的服务端实例
214    /// - `Err(FlareError)` - 构建失败的错误信息
215    pub fn build(self) -> Result<ObserverServer> {
216        // 验证配置（使用公共验证逻辑）
217        crate::server::builder::common::validate_auth_config(
218            &self.base.config,
219            &self.base.authenticator,
220        )?;
221
222        // 创建消息解析器（使用公共创建逻辑）
223        // let parser = crate::server::builder::common::create_message_parser(&self.base.config);
224
225        info!(
226            "[ObserverServerBuilder] 开始构建服务端: bind_address={}, protocols={:?}",
227            self.base.config.bind_address,
228            self.base.config.get_protocols()
229        );
230
231        let server = HybridServer::with_connection_manager(
232            self.base.config,
233            self.connection_manager,
234            self.device_manager,
235            Some(self.event_handler),
236            self.base.authenticator,
237        )
238        .map_err(|e| {
239            error!("[ObserverServerBuilder] 构建服务端失败: {}", e);
240            e
241        })?;
242
243        info!("[ObserverServerBuilder] 服务端构建成功");
244        Ok(ObserverServer {
245            wrapper: ServerWrapper::new(server),
246        })
247    }
248}
249
250/// 观察者模式服务器实例
251pub struct ObserverServer {
252    wrapper: ServerWrapper,
253}
254
255impl ObserverServer {
256    /// 启动服务器
257    pub async fn start(&mut self) -> Result<()> {
258        self.wrapper.start().await
259    }
260
261    /// 停止服务器
262    pub async fn stop(&mut self) -> Result<()> {
263        self.wrapper.stop().await
264    }
265
266    /// 检查服务器是否运行
267    pub fn is_running(&self) -> bool {
268        self.wrapper.is_running()
269    }
270
271    /// 获取连接数量
272    pub fn connection_count(&self) -> usize {
273        self.wrapper.connection_count()
274    }
275
276    /// 获取用户数量
277    pub fn user_count(&self) -> usize {
278        self.wrapper.user_count()
279    }
280
281    /// 向指定连接发送消息
282    pub async fn send_to(&self, connection_id: &str, frame: &Frame) -> Result<()> {
283        self.wrapper.send_to(connection_id, frame).await
284    }
285
286    /// 向指定用户的所有连接发送消息
287    pub async fn send_to_user(&self, user_id: &str, frame: &Frame) -> Result<()> {
288        self.wrapper.send_to_user(user_id, frame).await
289    }
290
291    /// 广播消息到所有连接
292    pub async fn broadcast(&self, frame: &Frame) -> Result<()> {
293        self.wrapper.broadcast(frame).await
294    }
295
296    /// 广播消息到所有连接，排除指定连接
297    pub async fn broadcast_except(&self, frame: &Frame, exclude_connection_id: &str) -> Result<()> {
298        self.wrapper
299            .broadcast_except(frame, exclude_connection_id)
300            .await
301    }
302
303    /// 断开指定连接
304    pub async fn disconnect(&self, connection_id: &str) -> Result<()> {
305        self.wrapper.disconnect(connection_id).await
306    }
307
308    /// 获取协议列表
309    pub fn protocols(&self) -> Vec<crate::common::config_types::TransportProtocol> {
310        self.wrapper.protocols()
311    }
312
313    /// 获取连接管理器（用于创建 DefaultServerHandle）
314    ///
315    /// # 返回
316    /// 返回 ConnectionManagerTrait
317    pub fn get_server_handle_components(
318        &self,
319    ) -> Option<Arc<dyn crate::server::connection::ConnectionManagerTrait>> {
320        self.wrapper.get_server_handle_components()
321    }
322
323    /// 获取 ServerHandle（用于消息发送和连接管理）
324    pub fn get_server_handle(&self) -> Option<Arc<dyn ServerHandle>> {
325        self.wrapper.get_server_handle()
326    }
327}
flare_core/server/builder/observer.rs

flare_core/server/builder/
observer.rs
