futu_core/conn_ip/
address_plan.rs

1use std::collections::BTreeMap;
2use std::time::Duration;
3
4/// C++ `GetConcurrencyAddress` returns at most four simultaneous candidates.
5/// Ref: `channel_address_manager.cpp:552-636`.
6pub const CHANNEL_CONCURRENT_ADDRESS_LIMIT: usize = 4;
7/// Ref: `connector.cpp:37` (`kTimeoutConnectConcurrencyAddrMs`).
8pub const CPP_CONCURRENCY_STAGE_DELAY: Duration = Duration::from_secs(3);
9/// Ref: `connector.cpp:38` (`kTimeoutConnectGuaranteedAddrMs`).
10pub const CPP_GUARANTEED_STAGE_DELAY: Duration = Duration::from_secs(1);
11/// C++ built-in Platform catalog primary port.
12///
13/// Ref: `channel_address_manager.cpp:335,675-731`. This is only the offline
14/// hardcoded fallback; a dynamic CMD1321 catalog replaces it.
15pub const CPP_PLATFORM_HARDCODED_PRIMARY_PORT: u16 = 443;
16/// C++ built-in Platform catalog same-IP backup port.
17///
18/// Ref: `channel_address_manager.cpp:335,675-731`. Dynamic CMD1321 rows carry
19/// their own optional backup port and are not forced to this value.
20pub const CPP_PLATFORM_HARDCODED_BACKUP_PORT: u16 = 9595;
21
22#[derive(Debug, Clone, PartialEq, Eq)]
23pub struct ChannelAddressCandidate {
24    pub endpoint: String,
25    pub region: u32,
26}
27
28#[derive(Debug, Clone, PartialEq, Eq)]
29pub struct ChannelAddressPlanFacts {
30    pub server_candidates: Vec<ChannelAddressCandidate>,
31    pub concurrency_fallback: Vec<String>,
32    pub guaranteed_candidates: Vec<String>,
33    pub anti_ddos: Option<String>,
34    pub emergency_fallback: Option<String>,
35    pub forced: Option<String>,
36    pub rotation: usize,
37    pub backup_endpoints: BTreeMap<String, String>,
38}
39
40#[derive(Debug, Clone, PartialEq, Eq)]
41pub struct ChannelAddressPlan {
42    pub forced: Option<String>,
43    pub concurrency: Vec<String>,
44    pub guaranteed: Option<String>,
45    pub anti_ddos: Option<String>,
46    pub emergency: Option<String>,
47    pub backup_endpoints: BTreeMap<String, String>,
48}
49
50impl ChannelAddressPlan {
51    #[must_use]
52    pub fn backup_endpoint_for(&self, primary: &str) -> Option<&str> {
53        self.backup_endpoints.get(primary).map(String::as_str)
54    }
55}
56
57#[derive(Debug, Clone, PartialEq, Eq)]
58pub struct ChannelAddressAttempt {
59    pub endpoint: String,
60    pub start_after: Duration,
61}
62
63/// Plan the common C++ Connector address stages for Platform and Broker.
64///
65/// The emergency fallback is Rust's compatibility escape hatch for a missing
66/// dynamic catalog. It is only exposed when every C++ strategy stage is empty,
67/// so it cannot revive the retired previous-address priority path.
68///
69/// Ref: `channel_address_manager.cpp:523-636` and
70/// `FTNet/channel/impl/connector.cpp:224-266`.
71#[must_use]
72pub fn plan_channel_addresses_like_cpp(facts: ChannelAddressPlanFacts) -> ChannelAddressPlan {
73    if let Some(forced) = facts.forced {
74        return ChannelAddressPlan {
75            forced: Some(forced),
76            concurrency: Vec::new(),
77            guaranteed: None,
78            anti_ddos: None,
79            emergency: None,
80            backup_endpoints: BTreeMap::new(),
81        };
82    }
83
84    let concurrency = if facts.server_candidates.is_empty() {
85        rotated_limit(facts.concurrency_fallback, facts.rotation)
86    } else {
87        server_concurrency(facts.server_candidates, facts.rotation)
88    };
89    let guaranteed = rotated_one(facts.guaranteed_candidates, facts.rotation);
90    let anti_ddos = facts.anti_ddos;
91    let emergency = (concurrency.is_empty() && guaranteed.is_none() && anti_ddos.is_none())
92        .then_some(facts.emergency_fallback)
93        .flatten();
94    let mut backup_endpoints = facts.backup_endpoints;
95    backup_endpoints.retain(|primary, _| {
96        concurrency.iter().any(|endpoint| endpoint == primary)
97            || guaranteed.as_ref() == Some(primary)
98            || anti_ddos.as_ref() == Some(primary)
99            || emergency.as_ref() == Some(primary)
100    });
101
102    ChannelAddressPlan {
103        forced: None,
104        concurrency,
105        guaranteed,
106        anti_ddos,
107        emergency,
108        backup_endpoints,
109    }
110}
111
112/// Expand a strategy plan into C++'s overlapping Connector timer schedule.
113///
114/// Concurrency starts immediately, guaranteed joins after 3 seconds, and
115/// anti-DDoS joins one second after guaranteed. Earlier attempts remain live;
116/// the first successful TCP connection wins.
117///
118/// Ref: `connector.cpp:224-271,375-397`.
119#[must_use]
120pub fn channel_address_attempt_schedule_like_cpp(
121    plan: &ChannelAddressPlan,
122) -> Vec<ChannelAddressAttempt> {
123    if let Some(forced) = plan.forced.as_ref() {
124        return vec![attempt(forced, Duration::from_secs(0))];
125    }
126
127    let mut attempts: Vec<_> = plan
128        .concurrency
129        .iter()
130        .map(|endpoint| attempt(endpoint, Duration::from_secs(0)))
131        .collect();
132    let mut next_stage_at = if attempts.is_empty() {
133        Duration::from_secs(0)
134    } else {
135        CPP_CONCURRENCY_STAGE_DELAY
136    };
137    if let Some(guaranteed) = plan.guaranteed.as_ref() {
138        attempts.push(attempt(guaranteed, next_stage_at));
139        next_stage_at += CPP_GUARANTEED_STAGE_DELAY;
140    }
141    if let Some(anti_ddos) = plan.anti_ddos.as_ref() {
142        attempts.push(attempt(anti_ddos, next_stage_at));
143    }
144    if attempts.is_empty()
145        && let Some(emergency) = plan.emergency.as_ref()
146    {
147        attempts.push(attempt(emergency, Duration::from_secs(0)));
148    }
149    attempts
150}
151
152fn attempt(endpoint: &str, start_after: Duration) -> ChannelAddressAttempt {
153    ChannelAddressAttempt {
154        endpoint: endpoint.to_string(),
155        start_after,
156    }
157}
158
159fn server_concurrency(candidates: Vec<ChannelAddressCandidate>, rotation: usize) -> Vec<String> {
160    let mut by_region: BTreeMap<u32, Vec<String>> = BTreeMap::new();
161    for candidate in candidates {
162        by_region
163            .entry(candidate.region)
164            .or_default()
165            .push(candidate.endpoint);
166    }
167
168    let mut selected: Vec<_> = by_region
169        .into_values()
170        .map(|addresses| addresses[rotation % addresses.len()].clone())
171        .collect();
172    if !selected.is_empty() {
173        let offset = rotation % selected.len();
174        selected.rotate_left(offset);
175    }
176    selected.truncate(CHANNEL_CONCURRENT_ADDRESS_LIMIT);
177    selected
178}
179
180fn rotated_limit(mut addresses: Vec<String>, rotation: usize) -> Vec<String> {
181    if !addresses.is_empty() {
182        let offset = rotation % addresses.len();
183        addresses.rotate_left(offset);
184    }
185    addresses.truncate(CHANNEL_CONCURRENT_ADDRESS_LIMIT);
186    addresses
187}
188
189fn rotated_one(addresses: Vec<String>, rotation: usize) -> Option<String> {
190    (!addresses.is_empty()).then(|| addresses[rotation % addresses.len()].clone())
191}
futu_core/conn_ip/address_plan.rs

futu_core/conn_ip/
address_plan.rs
