futu_server/

telnet.rs

// Telnet 管理接口 — 运维调试用
//
// 支持的命令:
//   help          — 显示帮助
//   ping          — 连通测试
//   version       — 版本信息
//   show_subinfo  — 显示订阅状态
//   set_loglevel  — 设置日志级别
//   exit          — 关闭网关
//
// C++ 对齐: APIServer_Telnet.cpp (最大 5 个并发连接)

use std::sync::Arc;

use dashmap::DashMap;
use tokio::io::{AsyncBufReadExt, AsyncWrite, AsyncWriteExt, BufReader};
use tokio::net::TcpListener;
use tokio::sync::watch;

use crate::conn::ClientConn;
use crate::metrics::GatewayMetrics;
use crate::subscription::SubscriptionManager;

/// v1.4.97 P1-D-F: telnet `relogin` 命令的回调.
///
/// 对齐 C++ `GTWCmd_ReLogin` (FTGateway/FTGTW_Define_Key.h:5 +
/// GTWCmdAndPushReply.cpp:1780-1799) 命令名 — 不创新 `force_reconnect_backend`
/// (per CLAUDE.md 坑 #51 "对齐 C++ = 减法").
///
/// `main.rs` wires 一个 closure 清 `bridge.caches.login_cache` (LoginCache::clear),
/// 让下个 30s P1-D health tick 看到 `qot_logined=false` → 自动 trigger
/// AuthRefresher relogin (走现有 v1.4.92 P1-D ladder 路径).
///
/// `Arc<dyn Fn>` 而非 trait object: 不需要 mutable state, 多 telnet client
/// 并发调时是 idempotent (clear 已 cleared 的 cache 仍 OK).
pub type ReloginFn = Arc<dyn Fn() + Send + Sync>;

/// Telnet 管理服务
pub struct TelnetServer {
    listen_addr: String,
    connections: Arc<DashMap<u64, ClientConn>>,
    subscriptions: Option<Arc<SubscriptionManager>>,
    metrics: Option<Arc<GatewayMetrics>>,
    shutdown_tx: watch::Sender<bool>,
    /// v1.4.97 P1-D-F: optional relogin callback (None = command unavailable
    /// with helpful error msg).
    relogin_fn: Option<ReloginFn>,
}

const MAX_TELNET_CONNECTIONS: usize = 5;
const VERSION: &str = env!("CARGO_PKG_VERSION");

async fn write_telnet<W>(
    writer: &mut W,
    peer_addr: std::net::SocketAddr,
    context: &'static str,
    bytes: &[u8],
) -> bool
where
    W: AsyncWrite + Unpin,
{
    match writer.write_all(bytes).await {
        Ok(()) => true,
        Err(err) => {
            tracing::debug!(
                peer = %peer_addr,
                context,
                error = %err,
                "telnet write failed; client likely disconnected"
            );
            false
        }
    }
}

impl TelnetServer {
    pub fn new(
        listen_addr: String,
        connections: Arc<DashMap<u64, ClientConn>>,
        subscriptions: Option<Arc<SubscriptionManager>>,
        metrics: Option<Arc<GatewayMetrics>>,
        shutdown_tx: watch::Sender<bool>,
    ) -> Self {
        Self {
            listen_addr,
            connections,
            subscriptions,
            metrics,
            shutdown_tx,
            relogin_fn: None,
        }
    }

    /// v1.4.97 P1-D-F: builder-style setter for relogin callback.
    pub fn with_relogin_fn(mut self, relogin_fn: ReloginFn) -> Self {
        self.relogin_fn = Some(relogin_fn);
        self
    }

    pub async fn run(&self) -> anyhow::Result<()> {
        let (_shutdown_tx, shutdown_rx) = watch::channel(false);
        self.run_until_shutdown(shutdown_rx).await
    }

    pub async fn run_until_shutdown(
        &self,
        mut shutdown_rx: watch::Receiver<bool>,
    ) -> anyhow::Result<()> {
        let listener = TcpListener::bind(&self.listen_addr)
            .await
            .map_err(|error| {
                anyhow::anyhow!(
                    "{}",
                    crate::bind_hint::bind_error_message(
                        "Telnet",
                        "--telnet-port",
                        &self.listen_addr,
                        error
                    )
                )
            })?;
        tracing::info!(addr = %self.listen_addr, "Telnet server listening");

        let active = Arc::new(std::sync::atomic::AtomicUsize::new(0));

        loop {
            let (stream, peer_addr) = tokio::select! {
                _ = crate::listener::shutdown_requested(&mut shutdown_rx) => {
                    tracing::info!("Telnet server accept loop stopped by shutdown signal");
                    break;
                }
                accepted = listener.accept() => accepted?,
            };
            let count = active.load(std::sync::atomic::Ordering::Relaxed);
            if count >= MAX_TELNET_CONNECTIONS {
                tracing::warn!(peer = %peer_addr, "telnet max connections reached");
                drop(stream);
                continue;
            }

            active.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
            let active_clone = Arc::clone(&active);
            let connections = Arc::clone(&self.connections);
            let subscriptions = self.subscriptions.clone();
            let metrics = self.metrics.clone();
            let shutdown_tx = self.shutdown_tx.clone();
            // v1.4.97 P1-D-F: clone relogin callback into spawned task.
            let relogin_fn = self.relogin_fn.clone();

            tokio::spawn(async move {
                tracing::info!(peer = %peer_addr, "telnet client connected");
                let (reader, mut writer) = stream.into_split();
                let mut reader = BufReader::new(reader);

                if !write_telnet(
                    &mut writer,
                    peer_addr,
                    "greeting",
                    b"FutuOpenD Rust Gateway Telnet Console\r\nType 'help' for commands.\r\n> ",
                )
                .await
                {
                    active_clone.fetch_sub(1, std::sync::atomic::Ordering::Relaxed);
                    tracing::info!(peer = %peer_addr, "telnet client disconnected");
                    return;
                }

                let mut line = String::new();
                loop {
                    line.clear();
                    match reader.read_line(&mut line).await {
                        Ok(0) => break, // EOF
                        Ok(_) => {}
                        Err(_) => break,
                    }

                    let cmd = line.trim().to_lowercase();
                    let response = match cmd.as_str() {
                        "help" => "Available commands:\r\n\
                             help           — show this help\r\n\
                             ping           — connectivity test\r\n\
                             version        — show version\r\n\
                             show_subinfo   — show subscription info\r\n\
                             show_conn      — show active connections\r\n\
                             show_metrics   — show gateway metrics\r\n\
                             set_loglevel   — change log level (debug/info/warn/error)\r\n\
                             relogin        — clear login_cache; next P1-D tick triggers relogin (v1.4.97)\r\n\
                             exit           — shutdown gateway\r\n"
                            .to_string(),
                        "ping" => "pong\r\n".to_string(),
                        "version" => format!("futu-opend-rs v{VERSION}\r\n"),
                        "show_subinfo" => {
                            if let Some(ref subs) = subscriptions {
                                let total_quota = subs.get_total_used_quota();
                                format!("Total subscription quota used: {total_quota}/4000\r\n")
                            } else {
                                "Subscription manager not available\r\n".to_string()
                            }
                        }
                        "show_conn" => {
                            let count = connections.len();
                            let mut info = format!("Active connections: {count}\r\n");
                            for entry in connections.iter() {
                                let c = entry.value();
                                info += &format!("  conn_id={} state={:?}\r\n", c.conn_id, c.state);
                            }
                            info
                        }
                        "show_metrics" => {
                            if let Some(ref m) = metrics {
                                m.report()
                            } else {
                                "Metrics not available\r\n".to_string()
                            }
                        }
                        s if s.starts_with("set_loglevel") => {
                            let parts: Vec<&str> = s.split_whitespace().collect();
                            if parts.len() < 2 {
                                "Usage: set_loglevel <debug|info|warn|error>\r\n".to_string()
                            } else {
                                // tracing 动态级别调整需要 reload handle，
                                // 简化实现：只记录意图
                                let level = parts[1];
                                tracing::info!(level, "log level change requested via telnet");
                                format!(
                                    "Log level change to '{level}' noted (restart for full effect)\r\n"
                                )
                            }
                        }
                        "relogin" => {
                            // v1.4.97 P1-D-F: align with C++ GTWCmd_ReLogin
                            // (FTGateway/FTGTW_Define_Key.h:5).
                            //
                            // Clear login_cache → next 30s health-loop tick
                            // sees qot_logined=false → triggers AuthRefresher
                            // relogin via existing v1.4.92 P1-D ladder path.
                            // Tester: wait up to ~30s, see daemon log
                            // "P1-D loop alive" + "relogin succeeded".
                            if let Some(ref f) = relogin_fn {
                                tracing::info!(peer = %peer_addr,
                                    "v1.4.97 P1-D-F: relogin requested via telnet");
                                f();
                                "Relogin triggered. Watch daemon log for \
                                 \"P1-D ladder fired\" + \"relogin succeeded\" \
                                 in next 30s tick.\r\n"
                                    .to_string()
                            } else {
                                "Relogin callback not wired (daemon was started \
                                 without bridge handle). Relogin requires daemon \
                                 restart in this configuration.\r\n"
                                    .to_string()
                            }
                        }
                        "exit" => {
                            write_telnet(
                                &mut writer,
                                peer_addr,
                                "shutdown_notice",
                                b"Shutting down...\r\n",
                            )
                            .await;
                            if shutdown_tx.send(true).is_err() {
                                tracing::warn!(
                                    peer = %peer_addr,
                                    "telnet exit requested but shutdown receiver dropped"
                                );
                            }
                            break;
                        }
                        "" => String::new(),
                        _ => format!(
                            "Unknown command: '{cmd}'. Type 'help' for available commands.\r\n"
                        ),
                    };

                    if !response.is_empty()
                        && !write_telnet(&mut writer, peer_addr, "response", response.as_bytes())
                            .await
                    {
                        break;
                    }
                    if !write_telnet(&mut writer, peer_addr, "prompt", b"> ").await {
                        break;
                    }
                }

                active_clone.fetch_sub(1, std::sync::atomic::Ordering::Relaxed);
                tracing::info!(peer = %peer_addr, "telnet client disconnected");
            });
        }

        Ok(())
    }
}

#[cfg(test)]
mod tests;