flare_core/client/builder/
common.rs

1//! 客户端构建器通用组件
2
3use crate::client::Client;
4use crate::common::config_types::{HeartbeatAppState, HeartbeatConfig, TransportProtocol};
5use crate::common::error::Result;
6use crate::common::protocol::Frame;
7use std::sync::Arc;
8use std::time::Duration;
9use tokio::sync::Mutex;
10
11#[cfg(not(target_arch = "wasm32"))]
12use crate::client::HybridClient;
13
14#[cfg(target_arch = "wasm32")]
15use crate::client::WebSocketClient;
16
17/// 客户端包装器 — native 使用 HybridClient，WASM 使用 WebSocketClient。
18#[derive(Clone)]
19pub struct ClientWrapper {
20    #[cfg(not(target_arch = "wasm32"))]
21    client: Arc<Mutex<HybridClient>>,
22    #[cfg(target_arch = "wasm32")]
23    client: Arc<Mutex<WebSocketClient>>,
24    reconnect_gate: Arc<Mutex<()>>,
25}
26
27impl ClientWrapper {
28    #[cfg(not(target_arch = "wasm32"))]
29    pub fn new(client: HybridClient) -> Self {
30        Self {
31            client: Arc::new(Mutex::new(client)),
32            reconnect_gate: Arc::new(Mutex::new(())),
33        }
34    }
35
36    #[cfg(target_arch = "wasm32")]
37    pub fn new(client: WebSocketClient) -> Self {
38        Self {
39            client: Arc::new(Mutex::new(client)),
40            reconnect_gate: Arc::new(Mutex::new(())),
41        }
42    }
43
44    pub async fn connect(&self) -> Result<()> {
45        let mut client = self.client.lock().await;
46        client.connect().await
47    }
48
49    pub async fn disconnect(&self) -> Result<()> {
50        let mut client = self.client.lock().await;
51        client.disconnect().await
52    }
53
54    pub async fn send_frame_and_wait(&self, frame: &Frame, timeout: Duration) -> Result<Frame> {
55        self.ensure_ready_for_send().await?;
56        let mut client = self.client.lock().await;
57        client.send_frame_and_wait(frame, timeout).await
58    }
59
60    pub async fn send_frame(&self, frame: &Frame) -> Result<()> {
61        self.ensure_ready_for_send().await?;
62        let mut client = self.client.lock().await;
63        client.send_frame(frame).await
64    }
65
66    async fn ensure_ready_for_send(&self) -> Result<()> {
67        if self.is_send_ready().await {
68            return Ok(());
69        }
70
71        let _single_flight = self.reconnect_gate.lock().await;
72        if self.is_send_ready().await {
73            return Ok(());
74        }
75
76        {
77            let mut client = self.client.lock().await;
78            if !client.is_connected() {
79                client.connect().await?;
80            }
81        }
82
83        let core = {
84            let client = self.client.lock().await;
85            client.core().clone()
86        };
87        core.wait_for_negotiation(Duration::from_secs(10)).await
88    }
89
90    async fn is_send_ready(&self) -> bool {
91        let client = self.client.lock().await;
92        client.is_connected()
93            && client.core().can_send()
94            && client.core().is_negotiation_completed()
95    }
96
97    pub async fn add_observer(&self, observer: crate::transport::events::ArcObserver) {
98        let mut client = self.client.lock().await;
99        client.add_observer(observer);
100    }
101
102    #[cfg(not(target_arch = "wasm32"))]
103    pub async fn set_event_handler(
104        &self,
105        handler: Option<Arc<dyn crate::client::events::handler::ClientEventHandler>>,
106    ) {
107        let mut client = self.client.lock().await;
108        client.core_mut().set_event_handler(handler);
109    }
110
111    pub async fn is_connected_async(&self) -> bool {
112        let client = self.client.lock().await;
113        client.is_connected()
114    }
115
116    pub async fn connection_id_async(&self) -> Option<String> {
117        let client = self.client.lock().await;
118        client.connection_id()
119    }
120
121    pub fn active_protocol(&self) -> TransportProtocol {
122        #[cfg(not(target_arch = "wasm32"))]
123        {
124            crate::client::runtime::run_client_async(async {
125                let client = self.client.lock().await;
126                client.active_protocol()
127            })
128        }
129        #[cfg(target_arch = "wasm32")]
130        {
131            TransportProtocol::WebSocket
132        }
133    }
134
135    pub async fn with_core<F, R>(&self, f: F) -> R
136    where
137        F: FnOnce(&crate::client::transports::ClientCore) -> R,
138    {
139        let client = self.client.lock().await;
140        f(client.core())
141    }
142
143    pub async fn update_heartbeat_config(&self, config: HeartbeatConfig) {
144        self.with_core(|core| {
145            core.update_heartbeat_config(|current| {
146                *current = config;
147            });
148        })
149        .await;
150    }
151
152    pub async fn set_heartbeat_app_state(&self, state: HeartbeatAppState) {
153        self.with_core(|core| {
154            core.set_heartbeat_app_state(state);
155        })
156        .await;
157    }
158
159    pub async fn set_heartbeat_nat_timeout(&self, timeout: Option<Duration>) {
160        self.with_core(|core| {
161            core.set_heartbeat_nat_timeout(timeout);
162        })
163        .await;
164    }
165
166    pub async fn heartbeat_effective_interval(&self) -> Duration {
167        self.with_core(|core| core.heartbeat_effective_interval())
168            .await
169    }
170
171    /// 获取协商后的消息解析器快照（连接完成后用于解析 `ConnectionEvent::Message` 原始字节）
172    pub async fn parser_snapshot(&self) -> crate::common::MessageParser {
173        let client = self.client.lock().await;
174        client.core().parser.lock().await.clone()
175    }
176
177    /// 等待 CONNECT_ACK 协商完成（不长期占用客户端锁，便于 WASM 侧 drain 入站帧）
178    pub async fn wait_for_negotiation(&self, timeout: Duration) -> Result<()> {
179        let core = {
180            let client = self.client.lock().await;
181            client.core().clone()
182        };
183        core.wait_for_negotiation(timeout).await
184    }
185}
flare_core/client/builder/common.rs

flare_core/client/builder/
common.rs
