futu_backend/
heartbeat.rs

1// 后端心跳保活
2//
3// C++ 分两个 cmd_id（`channel_impl.h:62-63`）：
4// - `kCmdHeartbeatPlatform = 6003` —— Platform 通道心跳
5// - `kCmdHeartbeatBroker   = 1003` —— Broker 通道心跳
6//
7// v1.4.7 之前全部硬编码 1003 发到 Platform 通道，服务端返回
8// `code=-102 CONN can not find command service`。
9//
10// 心跳必须加密: body = encrypt(sec_data(4B BE) + HeartBeatReq proto)
11
12use std::sync::Arc;
13use std::time::Duration;
14
15use crate::conn::BackendConn;
16use crate::proto_internal::ft_conn_heart_beat::HeartBeatReq;
17
18/// Platform 通道心跳命令 ID（C++ `kCmdHeartbeatPlatform`）
19pub const CMD_HEARTBEAT_PLATFORM: u16 = 6003;
20/// Broker 通道心跳命令 ID（C++ `kCmdHeartbeatBroker`）
21pub const CMD_HEARTBEAT_BROKER: u16 = 1003;
22
23/// 启动 **Platform 通道**心跳任务（cmd=6003）
24pub fn start_heartbeat(conn: Arc<BackendConn>, interval: Duration) -> tokio::task::JoinHandle<()> {
25    start_heartbeat_with_cmd(conn, interval, CMD_HEARTBEAT_PLATFORM, "platform")
26}
27
28/// 启动 **Broker 通道**心跳任务（cmd=1003）
29pub fn start_broker_heartbeat(
30    conn: Arc<BackendConn>,
31    interval: Duration,
32) -> tokio::task::JoinHandle<()> {
33    start_heartbeat_with_cmd(conn, interval, CMD_HEARTBEAT_BROKER, "broker")
34}
35
36/// 通用心跳任务。连续 3 次失败后退出任务（触发重连逻辑）。
37fn start_heartbeat_with_cmd(
38    conn: Arc<BackendConn>,
39    interval: Duration,
40    cmd_id: u16,
41    channel_name: &'static str,
42) -> tokio::task::JoinHandle<()> {
43    tokio::spawn(async move {
44        let mut ticker = tokio::time::interval(interval);
45        ticker.tick().await; // 跳过首次
46        let mut fail_count = 0u32;
47        const MAX_FAILURES: u32 = 3;
48
49        loop {
50            ticker.tick().await;
51            if !conn.is_connected() {
52                tracing::error!(
53                    channel = channel_name,
54                    "heartbeat observed disconnected backend conn; exiting to trigger reconnect"
55                );
56                break;
57            }
58
59            let req = HeartBeatReq {
60                pre_time_delay: Some(0),
61            };
62            let body = prost::Message::encode_to_vec(&req);
63
64            match conn.request(cmd_id, body).await {
65                Ok(resp) => {
66                    tracing::trace!(
67                        channel = channel_name,
68                        cmd_id,
69                        body_len = resp.body.len(),
70                        "backend heartbeat ok"
71                    );
72                    fail_count = 0; // 重置失败计数
73                }
74                Err(e) => {
75                    fail_count += 1;
76                    tracing::warn!(
77                        channel = channel_name,
78                        cmd_id,
79                        error = %e,
80                        fail_count,
81                        max = MAX_FAILURES,
82                        "backend heartbeat failed"
83                    );
84                    if fail_count >= MAX_FAILURES {
85                        tracing::error!(
86                            channel = channel_name,
87                            "heartbeat failed {MAX_FAILURES} times, connection likely lost"
88                        );
89                        break;
90                    }
91                }
92            }
93        }
94    })
95}
futu_backend/heartbeat.rs

futu_backend/
heartbeat.rs
