futu_server/subscription/
push_regs.rs

1use futu_core::qot_stock_key::QotSecurityKey;
2use futu_qot::subscription_plan::is_kl_sub_type;
3
4use super::SubscriptionManager;
5
6impl SubscriptionManager {
7    // ===== 行情 push 注册 (qot_push_regs, F3 独立) =====
8
9    /// **v1.4.106 codex 1131 F3 [P1]**: 注册接收 push (对齐 C++ `RegPush`).
10    /// 仅写本 map, **不动 `qot_subs`**. caller (RegQotPushHandler) 必须在
11    /// 调本方法前确认已订阅 (`is_globally_subscribed_broker`).
12    pub fn register_push_broker(
13        &self,
14        conn_id: u64,
15        sec_key: &QotSecurityKey,
16        sub_type: i32,
17        rehab_type: i32,
18    ) {
19        self.register_push_inner(conn_id, Self::broker_key(sec_key), sub_type, rehab_type)
20    }
21
22    fn register_push_inner(
23        &self,
24        conn_id: u64,
25        key: QotSecurityKey,
26        sub_type: i32,
27        rehab_type: i32,
28    ) {
29        let effective_rehab = if is_kl_sub_type(sub_type) {
30            rehab_type
31        } else {
32            0
33        };
34        let cache_key = key.cache_key();
35        let mut regs = self.qot_push_regs.write();
36        regs.by_tuple
37            .entry((key, sub_type, effective_rehab))
38            .or_default()
39            .insert(conn_id);
40        regs.qot_push_regs_by_cache_key
41            .entry(cache_key)
42            .or_default()
43            .entry((sub_type, effective_rehab))
44            .or_default()
45            .insert(conn_id);
46    }
47
48    /// **v1.4.106 codex 1131 F3 [P1]**: 取消 push 注册 — 不删 `qot_subs`.
49    pub fn unregister_push_broker(
50        &self,
51        conn_id: u64,
52        sec_key: &QotSecurityKey,
53        sub_type: i32,
54        rehab_type: i32,
55    ) {
56        self.unregister_push_inner(conn_id, Self::broker_key(sec_key), sub_type, rehab_type)
57    }
58
59    fn unregister_push_inner(
60        &self,
61        conn_id: u64,
62        key: QotSecurityKey,
63        sub_type: i32,
64        rehab_type: i32,
65    ) {
66        let effective_rehab = if is_kl_sub_type(sub_type) {
67            rehab_type
68        } else {
69            0
70        };
71        let cache_key = key.cache_key();
72        let map_key = (key, sub_type, effective_rehab);
73        let mut regs = self.qot_push_regs.write();
74        if let Some(set) = regs.by_tuple.get_mut(&map_key) {
75            set.remove(&conn_id);
76            if set.is_empty() {
77                regs.by_tuple.remove(&map_key);
78            }
79        }
80        if let Some(by_sub) = regs.qot_push_regs_by_cache_key.get_mut(&cache_key) {
81            if let Some(set) = by_sub.get_mut(&(sub_type, effective_rehab)) {
82                set.remove(&conn_id);
83                if set.is_empty() {
84                    by_sub.remove(&(sub_type, effective_rehab));
85                }
86            }
87            if by_sub.is_empty() {
88                regs.qot_push_regs_by_cache_key.remove(&cache_key);
89            }
90        }
91    }
92
93    /// **v1.4.106 codex 1131 F4 [P1]**: push delivery filter. 对齐 C++
94    /// `QotSubscribe::GetPushConn`.
95    pub fn get_qot_push_subscribers_broker(
96        &self,
97        sec_key: &QotSecurityKey,
98        sub_type: i32,
99        rehab_type: i32,
100    ) -> Vec<u64> {
101        self.get_qot_push_subscribers_inner(Self::broker_key(sec_key), sub_type, rehab_type)
102    }
103
104    /// Lookup by internal cache-key display
105    /// string carried by push events (`"market_code"` or `"market_code@b{id}"`).
106    ///
107    /// `PushEvent::QuotePush` does not carry `stock_id`, so dispatch cannot
108    /// reconstruct a full `QotSecurityKey`. Instead, match the already stored
109    /// subscription keys by their cache-key display string.
110    pub fn get_qot_push_subscribers_by_cache_key(
111        &self,
112        cache_key: &str,
113        sub_type: i32,
114        rehab_type: i32,
115    ) -> Vec<u64> {
116        let effective_rehab = if is_kl_sub_type(sub_type) {
117            rehab_type
118        } else {
119            0
120        };
121        let regs = self.qot_push_regs.read();
122        let mut out: Vec<u64> = regs
123            .qot_push_regs_by_cache_key
124            .get(cache_key)
125            .and_then(|by_sub| by_sub.get(&(sub_type, effective_rehab)))
126            .map_or_else(Vec::new, |subscribers| {
127                subscribers.iter().copied().collect()
128            });
129
130        out.sort_unstable();
131        out.dedup();
132        out
133    }
134
135    fn get_qot_push_subscribers_inner(
136        &self,
137        key: QotSecurityKey,
138        sub_type: i32,
139        rehab_type: i32,
140    ) -> Vec<u64> {
141        let effective_rehab = if is_kl_sub_type(sub_type) {
142            rehab_type
143        } else {
144            0
145        };
146        match self
147            .qot_push_regs
148            .read()
149            .by_tuple
150            .get(&(key, sub_type, effective_rehab))
151        {
152            Some(subscribers) => subscribers.iter().copied().collect(),
153            None => Vec::new(),
154        }
155    }
156
157    /// **v1.4.106 codex 1131 F3 [P1]**: 是否注册过 push (任意 rehab).
158    pub fn is_push_registered_any_rehab_broker(
159        &self,
160        conn_id: u64,
161        sec_key: &QotSecurityKey,
162        sub_type: i32,
163    ) -> bool {
164        let broker_key = Self::broker_key(sec_key);
165        let pr = self.qot_push_regs.read();
166        pr.by_tuple.iter().any(|((k, st, _rehab), set)| {
167            k == &broker_key && *st == sub_type && set.contains(&conn_id)
168        })
169    }
170}
futu_server/subscription/push_regs.rs

futu_server/subscription/
push_regs.rs
