futu_backend/channel_transport/
login.rs

1use async_trait::async_trait;
2use futu_core::conn_ip::ChannelAddressPlan;
3use futu_core::error::{FutuError, TransportFailureReason};
4use futu_core::log_redact::endpoint_log_fingerprint;
5
6use super::connect::connect_channel_address_plan;
7use crate::conn::{BackendConn, BackendProtocolIdentity, PushCallback};
8use crate::login::{
9    self, LoginResult, TcpLoginAttemptOutcome, TcpLoginSessionPhase, TcpLoginTarget,
10};
11
12#[async_trait]
13pub trait ChannelLoginDriver: Sync {
14    async fn attempt(
15        &self,
16        conn: &BackendConn,
17        target: TcpLoginTarget<'_>,
18    ) -> Result<TcpLoginAttemptOutcome, FutuError>;
19}
20
21pub struct ChannelConnectLoginSuccess {
22    pub conn: BackendConn,
23    pub login_result: LoginResult,
24    pub final_addr: String,
25}
26
27#[derive(Debug)]
28pub enum ChannelConnectLoginFailure {
29    Connect(FutuError),
30    RedirectConnect(FutuError),
31    RetryableLoginTransport(FutuError),
32    Login(FutuError),
33}
34
35impl ChannelConnectLoginFailure {
36    pub fn into_error(self) -> FutuError {
37        match self {
38            Self::Connect(error)
39            | Self::RedirectConnect(error)
40            | Self::RetryableLoginTransport(error)
41            | Self::Login(error) => error,
42        }
43    }
44}
45
46/// Execute the shared Platform/Broker redirect and backup-port login lifecycle.
47///
48/// Ref: `FTlogin/login/logger.cpp:392-438,850-900` and
49/// `FTNet/channel/impl/connector.cpp:108-172`.
50pub async fn connect_and_login_channel(
51    address_plan: &ChannelAddressPlan,
52    push_cb: PushCallback,
53    protocol_identity: BackendProtocolIdentity,
54    phase: TcpLoginSessionPhase,
55    channel_label: &'static str,
56    driver: &impl ChannelLoginDriver,
57) -> Result<ChannelConnectLoginSuccess, ChannelConnectLoginFailure> {
58    let (mut conn, mut current_addr) =
59        connect_channel_address_plan(address_plan, push_cb.clone(), protocol_identity)
60            .await
61            .map_err(ChannelConnectLoginFailure::Connect)?;
62    let mut redirect_ttl = None;
63    let mut redirect_count = 0;
64    let mut backup_port_attempted = false;
65
66    loop {
67        let (host_ip, host_port) = split_host_port(&current_addr);
68        let target = match redirect_ttl.take() {
69            Some(server_ttl) => TcpLoginTarget::redirected(phase, server_ttl, &host_ip, host_port),
70            None => TcpLoginTarget::fresh(phase, &host_ip, host_port),
71        };
72        match driver.attempt(&conn, target).await {
73            Ok(TcpLoginAttemptOutcome::Success(login_result)) => {
74                return Ok(ChannelConnectLoginSuccess {
75                    conn,
76                    login_result,
77                    final_addr: current_addr,
78                });
79            }
80            Ok(TcpLoginAttemptOutcome::Redirect(redirect)) => {
81                if redirect_count >= login::TCP_LOGIN_REDIRECT_SAFETY_LIMIT {
82                    return Err(ChannelConnectLoginFailure::Login(FutuError::Codec(
83                        format!(
84                            "{channel_label} login: too many redirects (>{})",
85                            login::TCP_LOGIN_REDIRECT_SAFETY_LIMIT,
86                        ),
87                    )));
88                }
89                let new_addr = redirect.endpoint();
90                tracing::info!(
91                    channel = channel_label,
92                    from_fp = %endpoint_log_fingerprint(&current_addr),
93                    to_fp = %endpoint_log_fingerprint(&new_addr),
94                    server_ttl = redirect.ttl,
95                    condition_flag = redirect.condition_flag,
96                    redirect_count,
97                    "channel login redirect"
98                );
99                conn = BackendConn::connect(&new_addr, push_cb.clone(), protocol_identity)
100                    .await
101                    .map_err(ChannelConnectLoginFailure::RedirectConnect)?;
102                current_addr = new_addr;
103                redirect_ttl = Some(redirect.ttl);
104                redirect_count += 1;
105                backup_port_attempted = false;
106            }
107            Err(error) => {
108                let Some(transport_reason) = retryable_login_transport_reason(&error) else {
109                    return Err(ChannelConnectLoginFailure::Login(error));
110                };
111                if backup_port_attempted {
112                    return Err(ChannelConnectLoginFailure::RetryableLoginTransport(error));
113                }
114                let Some(backup_addr) = address_plan
115                    .backup_endpoint_for(&current_addr)
116                    .map(str::to_owned)
117                else {
118                    return Err(ChannelConnectLoginFailure::RetryableLoginTransport(error));
119                };
120                tracing::warn!(
121                    channel = channel_label,
122                    transport_reason,
123                    primary_addr_fp = %endpoint_log_fingerprint(&current_addr),
124                    backup_addr_fp = %endpoint_log_fingerprint(&backup_addr),
125                    "channel login transport failed; retrying the C++ catalog backup port"
126                );
127                conn = BackendConn::connect(&backup_addr, push_cb.clone(), protocol_identity)
128                    .await
129                    .map_err(ChannelConnectLoginFailure::RedirectConnect)?;
130                current_addr = backup_addr;
131                redirect_ttl = None;
132                backup_port_attempted = true;
133            }
134        }
135    }
136}
137
138fn retryable_login_transport_reason(error: &FutuError) -> Option<&'static str> {
139    match error {
140        FutuError::Timeout => Some("timeout"),
141        FutuError::TransportFailure { reason, .. }
142            if matches!(
143                reason,
144                TransportFailureReason::PeerEof | TransportFailureReason::SendError
145            ) =>
146        {
147            Some(reason.as_str())
148        }
149        _ => None,
150    }
151}
152
153/// Preserve the historical fallback used when a compatibility endpoint lacks
154/// a numeric port. Production plans carry an explicit dynamic port; C++'s
155/// hardcoded/backup catalog uses 9595 (`channel_address_manager.cpp:335,675-731`).
156pub(super) fn split_host_port(addr: &str) -> (String, u32) {
157    match addr.rsplit_once(':') {
158        Some((host, port_text)) if !host.is_empty() => {
159            let port = port_text.parse::<u32>().unwrap_or(9595);
160            (host.to_string(), port)
161        }
162        _ => (addr.to_string(), 9595),
163    }
164}
futu_backend/channel_transport/login.rs

futu_backend/channel_transport/
login.rs
