futu_backend/channel_transport/
connect.rs

1use futu_core::conn_ip::{ChannelAddressPlan, channel_address_attempt_schedule_like_cpp};
2use futu_core::error::FutuError;
3use futu_core::log_redact::endpoint_log_fingerprint;
4use futures::stream::{FuturesUnordered, StreamExt};
5
6use crate::conn::{BackendConn, BackendProtocolIdentity, PushCallback};
7
8/// Execute the common C++ Connector TCP stages for Platform and Broker.
9///
10/// Ref: `FTNet/channel/impl/connector.cpp:224-266`.
11pub async fn connect_channel_address_plan(
12    plan: &ChannelAddressPlan,
13    push_cb: PushCallback,
14    protocol_identity: BackendProtocolIdentity,
15) -> Result<(BackendConn, String), FutuError> {
16    let schedule = channel_address_attempt_schedule_like_cpp(plan);
17    if schedule.is_empty() {
18        return Err(FutuError::Network(std::io::Error::new(
19            std::io::ErrorKind::InvalidInput,
20            "channel address plan is empty",
21        )));
22    }
23    let mut attempts: FuturesUnordered<_> = schedule
24        .into_iter()
25        .map(|attempt| {
26            let push_cb = push_cb.clone();
27            async move {
28                tokio::time::sleep(attempt.start_after).await;
29                let result =
30                    BackendConn::connect(&attempt.endpoint, push_cb, protocol_identity).await;
31                (attempt.endpoint, result)
32            }
33        })
34        .collect();
35    let mut last_error = None;
36    while let Some((endpoint, result)) = attempts.next().await {
37        match result {
38            Ok(conn) => return Ok((conn, endpoint)),
39            Err(error) => last_error = Some(error),
40        }
41    }
42
43    Err(last_error.unwrap_or_else(|| {
44        FutuError::Network(std::io::Error::other(
45            "Connector schedule completed without a TCP result",
46        ))
47    }))
48}
49
50pub fn channel_address_plan_fingerprints(plan: &ChannelAddressPlan) -> Vec<String> {
51    plan.forced
52        .iter()
53        .chain(plan.concurrency.iter())
54        .chain(plan.guaranteed.iter())
55        .chain(plan.anti_ddos.iter())
56        .chain(plan.emergency.iter())
57        .map(|address| endpoint_log_fingerprint(address))
58        .collect()
59}
futu_backend/channel_transport/connect.rs

futu_backend/channel_transport/
connect.rs
