flare_core/server/builder/
simple.rs

1//! 简单模式服务端构建器
2//!
3//! 提供最小实现，使用闭包定义消息处理逻辑，适合快速原型开发和学习。
4//!
5//! ## 特点
6//! - ✅ **使用闭包处理消息和连接事件**：无需实现 trait，直接使用闭包
7//! - ✅ **最小依赖**：只提供基本的消息处理功能
8//! - ✅ **零配置**：使用默认配置即可运行
9//! - ✅ **轻量级**：不包含中间件、管道等高级功能
10//! - ✅ **快速上手**：几行代码即可启动服务器
11//!
12//! ## 适用场景
13//! - 快速原型开发
14//! - 学习和测试
15//! - 小型应用
16//! - 需要完全控制消息处理流程的场景
17//!
18//! ## 架构说明
19//!
20//! 简单模式基于 `HybridServer`，共享所有核心能力（多协议支持、序列化协商、心跳检测等），
21//! 但消息处理逻辑通过闭包定义，不依赖高级抽象。
22
23use crate::common::error::Result;
24use crate::common::protocol::{Frame, PayloadCommand};
25use crate::server::HybridServer;
26use crate::server::builder::{BaseServerBuilderConfig, ServerWrapper};
27use crate::server::events::handler::ServerEventHandler;
28use crate::server::handle::ServerHandle;
29use async_trait::async_trait;
30use std::sync::Arc;
31use tokio::sync::Mutex;
32
33/// 消息处理上下文
34///
35/// 提供给消息处理函数的上下文，包含连接信息和服务器操作处理器
36pub struct MessageContext {
37    /// 连接 ID
38    pub connection_id: String,
39    /// 服务器操作处理器（轻量级，用于发送消息和连接管理）
40    handle: Arc<dyn ServerHandle>,
41}
42
43impl MessageContext {
44    /// 创建新的消息上下文
45    fn new(connection_id: String, handle: Arc<dyn ServerHandle>) -> Self {
46        Self {
47            connection_id,
48            handle,
49        }
50    }
51
52    /// 向指定连接发送消息
53    pub async fn send_to(&self, connection_id: &str, frame: &Frame) -> Result<()> {
54        self.handle.send_to(connection_id, frame).await
55    }
56
57    /// 向指定用户的所有连接发送消息
58    pub async fn send_to_user(&self, user_id: &str, frame: &Frame) -> Result<()> {
59        self.handle.send_to_user(user_id, frame).await
60    }
61
62    /// 广播消息到所有连接
63    pub async fn broadcast(&self, frame: &Frame) -> Result<()> {
64        self.handle.broadcast(frame).await
65    }
66
67    /// 广播消息到所有连接，排除指定连接
68    pub async fn broadcast_except(&self, frame: &Frame, exclude_connection_id: &str) -> Result<()> {
69        self.handle
70            .broadcast_except(frame, exclude_connection_id)
71            .await
72    }
73
74    /// 断开指定连接
75    pub async fn disconnect(&self, connection_id: &str) -> Result<()> {
76        self.handle.disconnect(connection_id).await
77    }
78
79    /// 获取连接数量
80    pub fn connection_count(&self) -> usize {
81        self.handle.connection_count()
82    }
83
84    /// 获取用户数量
85    pub fn user_count(&self) -> usize {
86        self.handle.user_count()
87    }
88}
89
90/// 消息处理函数类型
91pub type MessageHandlerFn = Box<
92    dyn for<'a> Fn(
93            &'a Frame,
94            &'a MessageContext,
95        ) -> std::pin::Pin<
96            Box<dyn std::future::Future<Output = Result<Option<Frame>>> + Send + 'a>,
97        > + Send
98        + Sync,
99>;
100
101/// 连接事件处理函数类型
102pub type OnConnectFn = Box<
103    dyn for<'a> Fn(
104            &'a str,
105            &'a MessageContext,
106        )
107            -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<()>> + Send + 'a>>
108        + Send
109        + Sync,
110>;
111
112/// 断开连接事件处理函数类型
113pub type OnDisconnectFn = Box<
114    dyn for<'a> Fn(
115            &'a str,
116            &'a MessageContext,
117        )
118            -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<()>> + Send + 'a>>
119        + Send
120        + Sync,
121>;
122
123/// 简化的连接处理器（仅用于存储闭包和 handle）
124struct SimpleConnectionHandler {
125    message_handler: Option<MessageHandlerFn>,
126    on_connect: Option<OnConnectFn>,
127    on_disconnect: Option<OnDisconnectFn>,
128    handle: Arc<Mutex<Option<Arc<dyn ServerHandle>>>>,
129}
130
131impl SimpleConnectionHandler {
132    async fn set_handle(&self, handle: Arc<dyn ServerHandle>) {
133        *self.handle.lock().await = Some(handle);
134    }
135}
136
137/// 将 SimpleConnectionHandler 适配为 ServerEventHandler
138struct SimpleEventHandlerAdapter {
139    handler: Arc<SimpleConnectionHandler>,
140}
141
142#[async_trait]
143impl ServerEventHandler for SimpleEventHandlerAdapter {
144    async fn handle_message(
145        &self,
146        command: &PayloadCommand,
147        connection_id: &str,
148    ) -> Result<Option<Frame>> {
149        let handle = {
150            let handle_guard = self.handler.handle.lock().await;
151            handle_guard.clone()
152        };
153
154        let context = if let Some(ref handle) = handle {
155            MessageContext::new(connection_id.to_string(), Arc::clone(handle))
156        } else {
157            return Err(crate::common::error::FlareError::general_error(
158                "Server handle is not available",
159            ));
160        };
161
162        let frame = Frame {
163            message_id: command.message_id.clone(),
164            command: Some(crate::common::protocol::flare::core::commands::Command {
165                r#type: Some(
166                    crate::common::protocol::flare::core::commands::command::Type::Payload(
167                        command.clone(),
168                    ),
169                ),
170            }),
171            metadata: std::collections::HashMap::new(),
172            reliability: crate::common::protocol::Reliability::AtLeastOnce as i32,
173            timestamp: 0,
174        };
175
176        if let Some(ref message_handler) = self.handler.message_handler {
177            message_handler(&frame, &context).await
178        } else {
179            Ok(None)
180        }
181    }
182
183    async fn on_connect(&self, connection_id: &str) -> Result<()> {
184        let handle = {
185            let handle_guard = self.handler.handle.lock().await;
186            handle_guard.clone()
187        };
188
189        let context = if let Some(ref handle) = handle {
190            MessageContext::new(connection_id.to_string(), Arc::clone(handle))
191        } else {
192            return Err(crate::common::error::FlareError::general_error(
193                "Server handle is not available",
194            ));
195        };
196
197        if let Some(ref on_connect) = self.handler.on_connect {
198            on_connect(connection_id, &context).await
199        } else {
200            Ok(())
201        }
202    }
203
204    async fn on_disconnect(&self, connection_id: &str, _reason: Option<&str>) -> Result<()> {
205        let handle = {
206            let handle_guard = self.handler.handle.lock().await;
207            handle_guard.clone()
208        };
209
210        let context = if let Some(ref handle) = handle {
211            MessageContext::new(connection_id.to_string(), Arc::clone(handle))
212        } else {
213            return Err(crate::common::error::FlareError::general_error(
214                "Server handle is not available",
215            ));
216        };
217
218        if let Some(ref on_disconnect) = self.handler.on_disconnect {
219            on_disconnect(connection_id, &context).await
220        } else {
221            Ok(())
222        }
223    }
224}
225
226/// 简化的服务器实例
227///
228/// 提供简化的接口，自动处理服务器引用
229pub struct SimpleServer {
230    wrapper: ServerWrapper,
231    handler: Arc<SimpleConnectionHandler>,
232    handle: Arc<dyn ServerHandle>,
233}
234
235impl SimpleServer {
236    /// 启动服务器
237    pub async fn start(&mut self) -> Result<()> {
238        // 设置 ServerHandle
239        self.handler.set_handle(Arc::clone(&self.handle)).await;
240        self.wrapper.start().await
241    }
242
243    /// 停止服务器
244    pub async fn stop(&mut self) -> Result<()> {
245        self.wrapper.stop().await
246    }
247
248    /// 检查服务器是否运行
249    pub fn is_running(&self) -> bool {
250        self.wrapper.is_running()
251    }
252
253    /// 获取连接数量
254    pub fn connection_count(&self) -> usize {
255        self.wrapper.connection_count()
256    }
257
258    /// 获取用户数量
259    pub fn user_count(&self) -> usize {
260        self.wrapper.user_count()
261    }
262
263    /// 获取 ServerHandle（用于消息发送和连接管理）
264    pub fn handle(&self) -> Arc<dyn ServerHandle> {
265        Arc::clone(&self.handle)
266    }
267
268    /// 向指定连接发送消息
269    pub async fn send_to(&self, connection_id: &str, frame: &Frame) -> Result<()> {
270        self.wrapper.send_to(connection_id, frame).await
271    }
272
273    /// 向指定用户的所有连接发送消息
274    pub async fn send_to_user(&self, user_id: &str, frame: &Frame) -> Result<()> {
275        self.wrapper.send_to_user(user_id, frame).await
276    }
277
278    /// 广播消息到所有连接
279    pub async fn broadcast(&self, frame: &Frame) -> Result<()> {
280        self.wrapper.broadcast(frame).await
281    }
282
283    /// 广播消息到所有连接，排除指定连接
284    pub async fn broadcast_except(&self, frame: &Frame, exclude_connection_id: &str) -> Result<()> {
285        self.wrapper
286            .broadcast_except(frame, exclude_connection_id)
287            .await
288    }
289
290    /// 断开指定连接
291    pub async fn disconnect(&self, connection_id: &str) -> Result<()> {
292        self.wrapper.disconnect(connection_id).await
293    }
294
295    /// 获取协议列表
296    pub fn protocols(&self) -> Vec<crate::common::config_types::TransportProtocol> {
297        self.wrapper.protocols()
298    }
299}
300
301/// 简单模式服务端构建器
302///
303/// 提供最小实现，使用闭包定义消息处理逻辑，适合快速原型开发和学习。
304///
305/// ## 设计原则
306///
307/// - **公共逻辑统一处理**：基于 `HybridServer`，共享所有核心能力（多协议、序列化协商、心跳等）
308/// - **最小抽象**：消息处理通过闭包定义，不依赖 trait 实现
309/// - **零配置**：使用默认配置即可运行
310///
311/// ## 使用方式
312///
313/// 使用闭包定义消息处理和连接事件处理逻辑，无需实现 trait。
314pub struct ServerBuilder {
315    base: BaseServerBuilderConfig,
316    message_handler: Option<MessageHandlerFn>,
317    on_connect: Option<OnConnectFn>,
318    on_disconnect: Option<OnDisconnectFn>,
319}
320
321impl ServerBuilder {
322    /// 创建新的服务端构建器
323    ///
324    /// # 参数
325    /// - `bind_address`: 监听地址，例如 "0.0.0.0:8080"
326    pub fn new(bind_address: impl Into<String>) -> Self {
327        Self {
328            base: BaseServerBuilderConfig::new(bind_address),
329            message_handler: None,
330            on_connect: None,
331            on_disconnect: None,
332        }
333    }
334
335    /// 设置认证器（如果启用认证，必须提供）
336    ///
337    /// 如果设置了认证器，还需要在配置中启用认证：
338    /// ```rust,ignore
339    /// .enable_auth()
340    /// .with_authenticator(authenticator)
341    /// ```
342    pub fn with_authenticator(
343        mut self,
344        authenticator: Arc<dyn crate::server::auth::Authenticator>,
345    ) -> Self {
346        self.base = self.base.with_authenticator(authenticator);
347        self
348    }
349
350    /// 启用认证
351    pub fn enable_auth(mut self) -> Self {
352        self.base = self.base.enable_auth();
353        self
354    }
355
356    /// 设置认证超时时间
357    pub fn with_auth_timeout(mut self, timeout: std::time::Duration) -> Self {
358        self.base = self.base.with_auth_timeout(timeout);
359        self
360    }
361
362    /// 设置消息处理函数
363    ///
364    /// # 参数
365    /// - `handler`: 消息处理函数，接收 `Frame` 和 `MessageContext`，返回 `Option<Frame>`（可选回复）
366    pub fn on_message<F>(mut self, handler: F) -> Self
367    where
368        F: for<'a> Fn(
369                &'a Frame,
370                &'a MessageContext,
371            ) -> std::pin::Pin<
372                Box<dyn std::future::Future<Output = Result<Option<Frame>>> + Send + 'a>,
373            > + Send
374            + Sync
375            + 'static,
376    {
377        self.message_handler = Some(Box::new(move |frame, ctx| handler(frame, ctx)));
378        self
379    }
380
381    /// 设置连接建立事件处理函数
382    ///
383    /// # 参数
384    /// - `handler`: 连接建立处理函数，接收 connection_id 和 MessageContext
385    pub fn on_connect<F>(mut self, handler: F) -> Self
386    where
387        F: for<'a> Fn(
388                &'a str,
389                &'a MessageContext,
390            ) -> std::pin::Pin<
391                Box<dyn std::future::Future<Output = Result<()>> + Send + 'a>,
392            > + Send
393            + Sync
394            + 'static,
395    {
396        self.on_connect = Some(Box::new(move |conn_id, ctx| handler(conn_id, ctx)));
397        self
398    }
399
400    /// 设置连接断开事件处理函数
401    ///
402    /// # 参数
403    /// - `handler`: 连接断开处理函数，接收 connection_id 和 MessageContext
404    pub fn on_disconnect<F>(mut self, handler: F) -> Self
405    where
406        F: for<'a> Fn(
407                &'a str,
408                &'a MessageContext,
409            ) -> std::pin::Pin<
410                Box<dyn std::future::Future<Output = Result<()>> + Send + 'a>,
411            > + Send
412            + Sync
413            + 'static,
414    {
415        self.on_disconnect = Some(Box::new(move |conn_id, ctx| handler(conn_id, ctx)));
416        self
417    }
418
419    /// 设置传输协议
420    pub fn with_protocol(
421        mut self,
422        protocol: crate::common::config_types::TransportProtocol,
423    ) -> Self {
424        self.base = self.base.with_protocol(protocol);
425        self
426    }
427
428    /// 启用多协议监听
429    pub fn with_protocols(
430        mut self,
431        protocols: Vec<crate::common::config_types::TransportProtocol>,
432    ) -> Self {
433        self.base = self.base.with_protocols(protocols);
434        self
435    }
436
437    /// 设置最大连接数
438    pub fn with_max_connections(mut self, max: usize) -> Self {
439        self.base = self.base.with_max_connections(max);
440        self
441    }
442
443    /// 设置握手超时时间
444    pub fn with_handshake_timeout(mut self, timeout: std::time::Duration) -> Self {
445        self.base = self.base.with_handshake_timeout(timeout);
446        self
447    }
448
449    /// 设置最大并发握手数
450    pub fn with_max_handshake_concurrency(mut self, max: usize) -> Self {
451        self.base = self.base.with_max_handshake_concurrency(max);
452        self
453    }
454
455    /// 设置单次连接写入超时时间
456    pub fn with_write_timeout(mut self, timeout: std::time::Duration) -> Self {
457        self.base = self.base.with_write_timeout(timeout);
458        self
459    }
460
461    /// 设置 fanout 发送最大并发度
462    pub fn with_fanout_concurrency(mut self, max: usize) -> Self {
463        self.base = self.base.with_fanout_concurrency(max);
464        self
465    }
466
467    /// 设置心跳配置
468    pub fn with_heartbeat(
469        mut self,
470        heartbeat: crate::common::config_types::HeartbeatConfig,
471    ) -> Self {
472        self.base = self.base.with_heartbeat(heartbeat);
473        self
474    }
475
476    /// 设置 TLS 配置
477    pub fn with_tls(mut self, tls: crate::common::config_types::TlsConfig) -> Self {
478        self.base = self.base.with_tls(tls);
479        self
480    }
481
482    /// 设置默认序列化格式（用于协商，默认 Protobuf）
483    pub fn with_default_format(
484        mut self,
485        format: crate::common::protocol::SerializationFormat,
486    ) -> Self {
487        self.base = self.base.with_default_format(format);
488        self
489    }
490
491    /// 设置默认压缩算法（用于协商，默认 None）
492    pub fn with_default_compression(
493        mut self,
494        compression: crate::common::compression::CompressionAlgorithm,
495    ) -> Self {
496        self.base = self.base.with_default_compression(compression);
497        self
498    }
499
500    /// 构建服务端
501    ///
502    /// # 返回
503    /// 返回配置好的 SimpleServer 实例
504    pub fn build(self) -> Result<SimpleServer> {
505        let handler = Arc::new(SimpleConnectionHandler {
506            message_handler: self.message_handler,
507            on_connect: self.on_connect,
508            on_disconnect: self.on_disconnect,
509            handle: Arc::new(Mutex::new(None)),
510        });
511
512        let event_handler: Arc<dyn crate::server::events::handler::ServerEventHandler> =
513            Arc::new(SimpleEventHandlerAdapter {
514                handler: handler.clone(),
515            });
516
517        let server = HybridServer::with_connection_manager(
518            self.base.config,
519            None,
520            None,
521            Some(event_handler),
522            self.base.authenticator,
523        )?;
524
525        // 使用 ServerWrapper 统一管理
526        let wrapper = ServerWrapper::new(server);
527
528        // 创建 ServerHandle（通过 ServerWrapper）
529        let handle = wrapper.get_server_handle().ok_or_else(|| {
530            crate::common::error::FlareError::general_error("Failed to create ServerHandle")
531        })?;
532
533        Ok(SimpleServer {
534            wrapper,
535            handler,
536            handle,
537        })
538    }
539}
flare_core/server/builder/simple.rs

flare_core/server/builder/
simple.rs
