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        let mut client = self.client.lock().await;
56        client.send_frame_and_wait(frame, timeout).await
57    }
58
59    pub async fn send_frame(&self, frame: &Frame) -> Result<()> {
60        self.ensure_ready_for_send().await?;
61        let mut client = self.client.lock().await;
62        client.send_frame(frame).await
63    }
64
65    async fn ensure_ready_for_send(&self) -> Result<()> {
66        if self.is_send_ready().await {
67            return Ok(());
68        }
69
70        let _single_flight = self.reconnect_gate.lock().await;
71        if self.is_send_ready().await {
72            return Ok(());
73        }
74
75        {
76            let mut client = self.client.lock().await;
77            if !client.is_connected() {
78                client.connect().await?;
79            }
80        }
81
82        let core = {
83            let client = self.client.lock().await;
84            client.core().clone()
85        };
86        core.wait_for_negotiation(Duration::from_secs(10)).await
87    }
88
89    async fn is_send_ready(&self) -> bool {
90        let client = self.client.lock().await;
91        client.is_connected()
92            && client.core().can_send()
93            && client.core().is_negotiation_completed()
94    }
95
96    pub async fn add_observer(&self, observer: crate::transport::events::ArcObserver) {
97        let mut client = self.client.lock().await;
98        client.add_observer(observer);
99    }
100
101    #[cfg(not(target_arch = "wasm32"))]
102    pub async fn set_event_handler(
103        &self,
104        handler: Option<Arc<dyn crate::client::events::handler::ClientEventHandler>>,
105    ) {
106        let mut client = self.client.lock().await;
107        client.core_mut().set_event_handler(handler);
108    }
109
110    pub async fn is_connected_async(&self) -> bool {
111        let client = self.client.lock().await;
112        client.is_connected()
113    }
114
115    pub async fn connection_id_async(&self) -> Option<String> {
116        let client = self.client.lock().await;
117        client.connection_id()
118    }
119
120    pub fn active_protocol(&self) -> TransportProtocol {
121        #[cfg(not(target_arch = "wasm32"))]
122        {
123            crate::client::runtime::run_client_async(async {
124                let client = self.client.lock().await;
125                client.active_protocol()
126            })
127        }
128        #[cfg(target_arch = "wasm32")]
129        {
130            TransportProtocol::WebSocket
131        }
132    }
133
134    pub async fn with_core<F, R>(&self, f: F) -> R
135    where
136        F: FnOnce(&crate::client::transports::ClientCore) -> R,
137    {
138        let client = self.client.lock().await;
139        f(client.core())
140    }
141
142    pub async fn update_heartbeat_config(&self, config: HeartbeatConfig) {
143        self.with_core(|core| {
144            core.update_heartbeat_config(|current| {
145                *current = config;
146            });
147        })
148        .await;
149    }
150
151    pub async fn set_heartbeat_app_state(&self, state: HeartbeatAppState) {
152        self.with_core(|core| {
153            core.set_heartbeat_app_state(state);
154        })
155        .await;
156    }
157
158    pub async fn set_heartbeat_nat_timeout(&self, timeout: Option<Duration>) {
159        self.with_core(|core| {
160            core.set_heartbeat_nat_timeout(timeout);
161        })
162        .await;
163    }
164
165    pub async fn heartbeat_effective_interval(&self) -> Duration {
166        self.with_core(|core| core.heartbeat_effective_interval())
167            .await
168    }
169
170    /// 获取协商后的消息解析器快照（连接完成后用于解析 `ConnectionEvent::Message` 原始字节）
171    pub async fn parser_snapshot(&self) -> crate::common::MessageParser {
172        let client = self.client.lock().await;
173        client.core().parser.lock().await.clone()
174    }
175
176    /// 等待 CONNECT_ACK 协商完成（不长期占用客户端锁，便于 WASM 侧 drain 入站帧）
177    pub async fn wait_for_negotiation(&self, timeout: Duration) -> Result<()> {
178        let core = {
179            let client = self.client.lock().await;
180            client.core().clone()
181        };
182        core.wait_for_negotiation(timeout).await
183    }
184}
flare_core/client/builder/common.rs

flare_core/client/builder/
common.rs
