futu_server/
router.rs

1// 请求路由器：按 ProtoID 分发请求到注册的业务处理器
2
3use std::collections::HashMap;
4use std::future::Future;
5use std::sync::Arc;
6
7use async_trait::async_trait;
8use parking_lot::RwLock;
9
10use crate::conn::IncomingRequest;
11use crate::identity::StartupReadiness;
12use crate::protect::ProtectionManager;
13
14/// 业务处理器 trait
15#[async_trait]
16pub trait RequestHandler: Send + Sync + 'static {
17    /// 处理请求，返回响应 body（protobuf 编码后的字节）
18    /// 返回 None 表示不产生响应（例如异步响应场景）
19    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>>;
20}
21
22/// 函数形式的处理器包装
23///
24/// 用来把闭包 `Fn(&IncomingRequest) -> Future<Option<Vec<u8>>>` 变成
25/// [`RequestHandler`] trait 对象，避免每个 handler 都写一个 struct + impl。
26pub struct FnHandler<F>(pub F);
27
28#[async_trait]
29impl<F, Fut> RequestHandler for FnHandler<F>
30where
31    F: Fn(u64, bytes::Bytes) -> Fut + Send + Sync + 'static,
32    Fut: Future<Output = Option<Vec<u8>>> + Send + 'static,
33{
34    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
35        (self.0)(conn_id, request.body.clone()).await
36    }
37}
38
39/// 请求路由器
40pub struct RequestRouter {
41    handlers: RwLock<HashMap<u32, Arc<dyn RequestHandler>>>,
42    protection: ProtectionManager,
43    startup_readiness: RwLock<StartupReadiness>,
44    /// Reviewer M-9 (#55): the not-ready rejection warn is rate-limited so a
45    /// client polling during first login cannot flood the log; suppressed
46    /// rejections are counted and reported with the next emitted warn.
47    not_ready_warn_last: parking_lot::Mutex<Option<std::time::Instant>>,
48    not_ready_suppressed: std::sync::atomic::AtomicU64,
49}
50
51/// Minimum spacing between two "gateway authentication not ready" warns.
52const NOT_READY_WARN_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1);
53
54impl RequestRouter {
55    /// 创建空路由器。使用 [`Self::register`] 挂 handler。
56    pub fn new() -> Self {
57        Self::with_startup_readiness(StartupReadiness::default())
58    }
59
60    pub fn with_startup_readiness(startup_readiness: StartupReadiness) -> Self {
61        Self {
62            handlers: RwLock::new(HashMap::new()),
63            protection: ProtectionManager::new(),
64            startup_readiness: RwLock::new(startup_readiness),
65            not_ready_warn_last: parking_lot::Mutex::new(None),
66            not_ready_suppressed: std::sync::atomic::AtomicU64::new(0),
67        }
68    }
69
70    /// Returns `Some(suppressed_since_last_warn)` when a not-ready warn should
71    /// be emitted now, `None` when it must be suppressed (counted).
72    fn not_ready_warn_budget(&self) -> Option<u64> {
73        let now = std::time::Instant::now();
74        let mut last = self.not_ready_warn_last.lock();
75        let due = last.is_none_or(|at| now.duration_since(at) >= NOT_READY_WARN_INTERVAL);
76        if due {
77            *last = Some(now);
78            Some(
79                self.not_ready_suppressed
80                    .swap(0, std::sync::atomic::Ordering::AcqRel),
81            )
82        } else {
83            self.not_ready_suppressed
84                .fetch_add(1, std::sync::atomic::Ordering::AcqRel);
85            None
86        }
87    }
88
89    pub fn startup_readiness(&self) -> StartupReadiness {
90        self.startup_readiness.read().clone()
91    }
92
93    pub fn set_startup_readiness(&self, startup_readiness: StartupReadiness) {
94        *self.startup_readiness.write() = startup_readiness;
95    }
96
97    /// 注册业务处理器
98    pub fn register(&self, proto_id: u32, handler: Arc<dyn RequestHandler>) {
99        self.handlers.write().insert(proto_id, handler);
100    }
101
102    /// 分发请求到对应处理器
103    pub async fn dispatch(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
104        let startup = self.startup_readiness.read().clone();
105        if !startup.allows_proto(request.proto_id) {
106            // GitHub issue #55: 这条拒绝曾经完全无日志，SDK 自动重订阅被拒时只能
107            // 从 98/111 字符的 "未订阅" 文案反推。必须 loud。
108            if let Some(suppressed) = self.not_ready_warn_budget() {
109                tracing::warn!(
110                    conn_id,
111                    proto_id = request.proto_id,
112                    startup_state = ?startup.snapshot().state,
113                    suppressed_since_last_warn = suppressed,
114                    "request rejected: gateway authentication not ready"
115                );
116            }
117            return Some(make_error_response(-1, "gateway authentication not ready"));
118        }
119        if request.proto_id == futu_core::proto_id::VERIFICATION
120            && startup.snapshot().state != crate::identity::StartupState::Ready
121            && !request.caller_has_auth_setup_scope
122            && !(request.caller_is_loopback && request.caller_legacy_local_mode)
123        {
124            return Some(make_error_response(
125                -1,
126                "verification pre-login admission denied",
127            ));
128        }
129        if self.protection.check_request_freq_limit(conn_id, request) {
130            tracing::warn!(
131                proto_id = request.proto_id,
132                conn_id = conn_id,
133                "request frequency limit exceeded"
134            );
135            return Some(make_error_response(-1, "request frequency limit exceeded"));
136        }
137
138        let handler = {
139            let handlers = self.handlers.read();
140            handlers.get(&request.proto_id).cloned()
141        };
142
143        match handler {
144            Some(h) => {
145                futu_command_runtime::with_surface_command_admission(
146                    request.proto_id,
147                    h.handle(conn_id, request),
148                )
149                .await
150            }
151            None => {
152                tracing::warn!(
153                    proto_id = request.proto_id,
154                    conn_id = conn_id,
155                    "no handler registered"
156                );
157                // 返回通用错误响应
158                Some(make_error_response(-1, "unknown protocol"))
159            }
160        }
161    }
162}
163
164impl Default for RequestRouter {
165    fn default() -> Self {
166        Self::new()
167    }
168}
169
170/// 构造通用错误响应（使用 InitConnect::Response 格式，与 C++ 兼容）
171fn make_error_response(ret_type: i32, msg: &str) -> Vec<u8> {
172    let resp = futu_proto::init_connect::Response {
173        ret_type,
174        ret_msg: Some(msg.to_string()),
175        err_code: None,
176        s2c: None,
177    };
178    prost::Message::encode_to_vec(&resp)
179}
180
181#[cfg(test)]
182mod tests;
futu_server/router.rs

futu_server/
router.rs
