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::protect::ProtectionManager;
12
13/// 业务处理器 trait
14#[async_trait]
15pub trait RequestHandler: Send + Sync + 'static {
16    /// 处理请求，返回响应 body（protobuf 编码后的字节）
17    /// 返回 None 表示不产生响应（例如异步响应场景）
18    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>>;
19}
20
21/// 函数形式的处理器包装
22///
23/// 用来把闭包 `Fn(&IncomingRequest) -> Future<Option<Vec<u8>>>` 变成
24/// [`RequestHandler`] trait 对象，避免每个 handler 都写一个 struct + impl。
25pub struct FnHandler<F>(pub F);
26
27#[async_trait]
28impl<F, Fut> RequestHandler for FnHandler<F>
29where
30    F: Fn(u64, bytes::Bytes) -> Fut + Send + Sync + 'static,
31    Fut: Future<Output = Option<Vec<u8>>> + Send + 'static,
32{
33    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
34        (self.0)(conn_id, request.body.clone()).await
35    }
36}
37
38/// 请求路由器
39pub struct RequestRouter {
40    handlers: RwLock<HashMap<u32, Arc<dyn RequestHandler>>>,
41    protection: ProtectionManager,
42}
43
44impl RequestRouter {
45    /// 创建空路由器。使用 [`Self::register`] 挂 handler。
46    pub fn new() -> Self {
47        Self {
48            handlers: RwLock::new(HashMap::new()),
49            protection: ProtectionManager::new(),
50        }
51    }
52
53    /// 注册业务处理器
54    pub fn register(&self, proto_id: u32, handler: Arc<dyn RequestHandler>) {
55        self.handlers.write().insert(proto_id, handler);
56    }
57
58    /// 分发请求到对应处理器
59    pub async fn dispatch(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
60        if self.protection.check_request_freq_limit(conn_id, request) {
61            tracing::warn!(
62                proto_id = request.proto_id,
63                conn_id = conn_id,
64                "request frequency limit exceeded"
65            );
66            return Some(make_error_response(-1, "request frequency limit exceeded"));
67        }
68
69        let handler = {
70            let handlers = self.handlers.read();
71            handlers.get(&request.proto_id).cloned()
72        };
73
74        match handler {
75            Some(h) => h.handle(conn_id, request).await,
76            None => {
77                tracing::warn!(
78                    proto_id = request.proto_id,
79                    conn_id = conn_id,
80                    "no handler registered"
81                );
82                // 返回通用错误响应
83                Some(make_error_response(-1, "unknown protocol"))
84            }
85        }
86    }
87}
88
89impl Default for RequestRouter {
90    fn default() -> Self {
91        Self::new()
92    }
93}
94
95/// 构造通用错误响应（使用 InitConnect::Response 格式，与 C++ 兼容）
96fn make_error_response(ret_type: i32, msg: &str) -> Vec<u8> {
97    let resp = futu_proto::init_connect::Response {
98        ret_type,
99        ret_msg: Some(msg.to_string()),
100        err_code: None,
101        s2c: None,
102    };
103    prost::Message::encode_to_vec(&resp)
104}
105
106#[cfg(test)]
107mod tests;
futu_server/router.rs

futu_server/
router.rs
