flare_core/client/builder/
observer.rs

1//! 观察者模式客户端构建器
2//!
3//! 提供基本功能实现，使用 `ConnectionObserver` trait 处理连接事件和消息。
4//!
5//! ## 特点
6//! - ✅ **实现 `ConnectionObserver` trait**：自定义消息和事件处理
7//! - ✅ **事件处理**：支持自定义事件处理器
8//! - ✅ **消息路由**：支持消息路由功能
9//! - ✅ **灵活扩展**：可以添加自定义的观察器和处理器
10//!
11//! ## 适用场景
12//! - 需要自定义消息处理逻辑但不需要完整功能集
13//! - 需要事件驱动的架构
14//! - 需要消息路由功能
15//!
16//! ## 架构说明
17//!
18//! 观察者模式基于 `HybridClient`，使用 `ConnectionObserver` trait 处理所有连接事件。
19//! 观察者可以处理消息、连接建立、断开、错误等事件，提供灵活的事件驱动架构。
20
21use crate::client::HybridClient;
22use crate::client::builder::{BaseClientBuilderConfig, ClientWrapper};
23use crate::common::error::Result;
24use crate::common::protocol::Frame;
25use crate::transport::events::ConnectionObserver;
26use std::sync::Arc;
27
28/// 观察者模式客户端构建器
29///
30/// 提供基本功能实现，使用 `ConnectionObserver` trait 处理连接事件和消息。
31///
32/// ## 设计原则
33///
34/// - **公共逻辑统一处理**：基于 `HybridClient`，共享所有核心能力
35/// - **事件驱动**：使用 `ConnectionObserver` trait 处理所有连接事件
36/// - **灵活扩展**：支持自定义观察器和事件处理器
37///
38/// ## 使用方式
39///
40/// 用户只需要实现 `ConnectionObserver` trait，处理消息、连接建立、断开、错误等事件。
41pub struct ObserverClientBuilder {
42    base: BaseClientBuilderConfig,
43    observer: Option<Arc<dyn ConnectionObserver>>,
44    event_handler: Option<Arc<dyn crate::client::events::handler::ClientEventHandler>>,
45}
46
47impl ObserverClientBuilder {
48    /// 创建新的观察者模式构建器
49    pub fn new(server_url: impl Into<String>) -> Self {
50        Self {
51            base: BaseClientBuilderConfig::new(server_url),
52            observer: None,
53            event_handler: None,
54        }
55    }
56
57    /// 设置事件处理器（可选，用于自定义业务逻辑）
58    pub fn with_event_handler(
59        mut self,
60        event_handler: Arc<dyn crate::client::events::handler::ClientEventHandler>,
61    ) -> Self {
62        self.event_handler = Some(event_handler);
63        self
64    }
65
66    /// 设置观察者（必须）
67    pub fn with_observer(mut self, observer: Arc<dyn ConnectionObserver>) -> Self {
68        self.observer = Some(observer);
69        self
70    }
71
72    /// 设置传输协议
73    pub fn with_protocol(
74        mut self,
75        protocol: crate::common::config_types::TransportProtocol,
76    ) -> Self {
77        self.base = self.base.with_protocol(protocol);
78        self
79    }
80
81    /// 启用多协议竞速
82    ///
83    /// 协议列表的顺序就是优先级顺序，前面的协议优先级更高
84    pub fn with_protocol_race(
85        mut self,
86        protocols: Vec<crate::common::config_types::TransportProtocol>,
87    ) -> Self {
88        self.base = self.base.with_protocol_race(protocols);
89        self
90    }
91
92    /// 为特定协议设置服务器地址
93    pub fn with_protocol_url(
94        mut self,
95        protocol: crate::common::config_types::TransportProtocol,
96        url: String,
97    ) -> Self {
98        self.base = self.base.with_protocol_url(protocol, url);
99        self
100    }
101
102    /// 设置用户 ID
103    pub fn with_user_id(mut self, user_id: String) -> Self {
104        self.base = self.base.with_user_id(user_id);
105        self
106    }
107
108    /// 设置序列化格式
109    pub fn with_format(mut self, format: crate::common::protocol::SerializationFormat) -> Self {
110        self.base = self.base.with_format(format);
111        self
112    }
113
114    /// 设置压缩算法
115    pub fn with_compression(
116        mut self,
117        compression: crate::common::compression::CompressionAlgorithm,
118    ) -> Self {
119        self.base = self.base.with_compression(compression);
120        self
121    }
122
123    /// 设置心跳配置
124    pub fn with_heartbeat(
125        mut self,
126        heartbeat: crate::common::config_types::HeartbeatConfig,
127    ) -> Self {
128        self.base = self.base.with_heartbeat(heartbeat);
129        self
130    }
131
132    /// 设置 TLS 配置
133    pub fn with_tls(mut self, tls: crate::common::config_types::TlsConfig) -> Self {
134        self.base = self.base.with_tls(tls);
135        self
136    }
137
138    /// 设置连接超时
139    pub fn with_connect_timeout(mut self, timeout: std::time::Duration) -> Self {
140        self.base = self.base.with_connect_timeout(timeout);
141        self
142    }
143
144    /// 设置重连间隔
145    pub fn with_reconnect_interval(mut self, interval: std::time::Duration) -> Self {
146        self.base = self.base.with_reconnect_interval(interval);
147        self
148    }
149
150    /// 设置最大重连次数
151    pub fn with_max_reconnect_attempts(mut self, max: Option<u32>) -> Self {
152        self.base = self.base.with_max_reconnect_attempts(max);
153        self
154    }
155
156    /// 启用消息路由
157    ///
158    /// 启用后，可以通过 ClientCore 的 router 方法注册消息处理器
159    pub fn enable_router(mut self) -> Self {
160        self.base = self.base.enable_router();
161        self
162    }
163
164    /// 设置设备信息（用于协商和设备管理）
165    pub fn with_device_info(mut self, device_info: crate::common::device::DeviceInfo) -> Self {
166        self.base = self.base.with_device_info(device_info);
167        self
168    }
169
170    /// 设置 Token（用于认证，如果服务端启用认证，必须提供）
171    pub fn with_token(mut self, token: String) -> Self {
172        self.base = self.base.with_token(token);
173        self
174    }
175
176    /// 构建客户端（使用协议竞速）
177    pub async fn build_with_race(self) -> Result<ObserverClient> {
178        let observer = self.observer.ok_or_else(|| {
179            crate::common::error::FlareError::general_error("Observer is required")
180        })?;
181
182        let client = HybridClient::connect_with_race(self.base.config).await?;
183        let wrapper = ClientWrapper::new(client);
184
185        // 设置事件处理器（如果提供）
186        if let Some(event_handler) = self.event_handler {
187            wrapper.set_event_handler(Some(event_handler)).await;
188        }
189
190        wrapper
191            .add_observer(Arc::clone(&observer) as Arc<dyn ConnectionObserver>)
192            .await;
193
194        Ok(ObserverClient {
195            wrapper,
196            observer: Some(observer),
197        })
198    }
199
200    /// 构建客户端
201    pub fn build(self) -> Result<ObserverClient> {
202        let observer = self.observer.ok_or_else(|| {
203            crate::common::error::FlareError::general_error("Observer is required")
204        })?;
205
206        let mut client = HybridClient::new(self.base.config)?;
207
208        // 设置事件处理器（如果提供）
209        if let Some(event_handler) = self.event_handler {
210            client.core_mut().set_event_handler(Some(event_handler));
211        }
212
213        let wrapper = ClientWrapper::new(client);
214
215        Ok(ObserverClient {
216            wrapper,
217            observer: Some(observer),
218        })
219    }
220}
221
222/// 观察者模式客户端实例
223pub struct ObserverClient {
224    wrapper: ClientWrapper,
225    observer: Option<Arc<dyn ConnectionObserver>>,
226}
227
228impl ObserverClient {
229    /// 连接到服务器
230    pub async fn connect(&mut self) -> Result<()> {
231        // 先添加观察者（如果还未添加）
232        if let Some(observer) = &self.observer {
233            self.wrapper
234                .add_observer(Arc::clone(observer) as Arc<dyn ConnectionObserver>)
235                .await;
236        }
237
238        self.wrapper.connect().await
239    }
240
241    /// 断开连接
242    pub async fn disconnect(&mut self) -> Result<()> {
243        self.wrapper.disconnect().await
244    }
245
246    /// 发送消息
247    pub async fn send_frame(&mut self, frame: &Frame) -> Result<()> {
248        self.wrapper.send_frame(frame).await
249    }
250
251    /// 发送并等待响应（按 message_id 匹配）
252    pub async fn send_frame_and_wait(
253        &mut self,
254        frame: &Frame,
255        timeout: std::time::Duration,
256    ) -> Result<Frame> {
257        self.wrapper.send_frame_and_wait(frame, timeout).await
258    }
259
260    /// 检查连接状态
261    pub fn is_connected(&self) -> bool {
262        crate::client::runtime::run_client_async(self.wrapper.is_connected_async())
263    }
264
265    /// 获取连接 ID
266    pub fn connection_id(&self) -> Option<String> {
267        crate::client::runtime::run_client_async(self.wrapper.connection_id_async())
268    }
269
270    /// 获取活动协议
271    pub fn active_protocol(&self) -> crate::common::config_types::TransportProtocol {
272        self.wrapper.active_protocol()
273    }
274}
flare_core/client/builder/observer.rs

flare_core/client/builder/
observer.rs
