futu_server/subscription/
push_regs.rs

1use futu_core::qot_stock_key::QotSecurityKey;
2use futu_domain_qot_subscription::is_kl_sub_type;
3
4use super::{QotPushRegistrationLease, 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 intent_epoch = self.next_qot_push_intent_epoch();
36        let mut regs = self.qot_push_regs.write();
37        regs.by_tuple
38            .entry((key.clone(), sub_type, effective_rehab))
39            .or_default()
40            .insert(conn_id);
41        regs.qot_push_regs_by_cache_key
42            .entry(cache_key)
43            .or_default()
44            .entry((sub_type, effective_rehab))
45            .or_default()
46            .insert(conn_id, intent_epoch);
47        regs.intent_epochs.insert(
48            (key.clone(), sub_type, effective_rehab, conn_id),
49            intent_epoch,
50        );
51        drop(regs);
52    }
53
54    /// **v1.4.106 codex 1131 F3 [P1]**: 取消 push 注册 — 不删 `qot_subs`.
55    pub fn unregister_push_broker(
56        &self,
57        conn_id: u64,
58        sec_key: &QotSecurityKey,
59        sub_type: i32,
60        rehab_type: i32,
61    ) {
62        self.unregister_push_inner(conn_id, Self::broker_key(sec_key), sub_type, rehab_type)
63    }
64
65    fn unregister_push_inner(
66        &self,
67        conn_id: u64,
68        key: QotSecurityKey,
69        sub_type: i32,
70        rehab_type: i32,
71    ) {
72        let effective_rehab = if is_kl_sub_type(sub_type) {
73            rehab_type
74        } else {
75            0
76        };
77        let cache_key = key.cache_key();
78        let map_key = (key.clone(), sub_type, effective_rehab);
79        let mut regs = self.qot_push_regs.write();
80        regs.intent_epochs
81            .remove(&(key, sub_type, effective_rehab, conn_id));
82        if let Some(set) = regs.by_tuple.get_mut(&map_key) {
83            set.remove(&conn_id);
84            if set.is_empty() {
85                regs.by_tuple.remove(&map_key);
86            }
87        }
88        if let Some(by_sub) = regs.qot_push_regs_by_cache_key.get_mut(&cache_key) {
89            if let Some(set) = by_sub.get_mut(&(sub_type, effective_rehab)) {
90                set.remove(&conn_id);
91                if set.is_empty() {
92                    by_sub.remove(&(sub_type, effective_rehab));
93                }
94            }
95            if by_sub.is_empty() {
96                regs.qot_push_regs_by_cache_key.remove(&cache_key);
97            }
98        }
99    }
100
101    /// **v1.4.106 codex 1131 F4 [P1]**: push delivery filter. 对齐 C++
102    /// `QotSubscribe::GetPushConn`.
103    pub fn get_qot_push_subscribers_broker(
104        &self,
105        sec_key: &QotSecurityKey,
106        sub_type: i32,
107        rehab_type: i32,
108    ) -> Vec<u64> {
109        self.get_qot_push_subscribers_inner(Self::broker_key(sec_key), sub_type, rehab_type)
110    }
111
112    /// Lookup by internal cache-key display
113    /// string carried by push events (`"market_code"` or `"market_code@b{id}"`).
114    ///
115    /// `PushEvent::QuotePush` does not carry `stock_id`, so dispatch cannot
116    /// reconstruct a full `QotSecurityKey`. Instead, match the already stored
117    /// subscription keys by their cache-key display string.
118    pub fn get_qot_push_subscribers_by_cache_key(
119        &self,
120        cache_key: &str,
121        sub_type: i32,
122        rehab_type: i32,
123    ) -> Vec<u64> {
124        self.get_qot_push_subscriber_intents_by_cache_key(cache_key, sub_type, rehab_type)
125            .into_iter()
126            .map(|(conn_id, _)| conn_id)
127            .collect()
128    }
129
130    /// Return the physical sink and exact registration intent that admitted a
131    /// cache-key routed push. Ordinary KLine delivery binds its success cursor
132    /// to this epoch so unregister -> re-register can first-push the current
133    /// point once without weakening duplicate suppression inside either intent.
134    pub(crate) fn get_qot_push_subscriber_intents_by_cache_key(
135        &self,
136        cache_key: &str,
137        sub_type: i32,
138        rehab_type: i32,
139    ) -> Vec<(u64, u64)> {
140        let effective_rehab = if is_kl_sub_type(sub_type) {
141            rehab_type
142        } else {
143            0
144        };
145        let regs = self.qot_push_regs.read();
146        let mut out: Vec<(u64, u64)> = regs
147            .qot_push_regs_by_cache_key
148            .get(cache_key)
149            .and_then(|by_sub| by_sub.get(&(sub_type, effective_rehab)))
150            .map_or_else(Vec::new, |subscribers| {
151                subscribers
152                    .iter()
153                    .map(|(conn_id, intent_epoch)| (*conn_id, *intent_epoch))
154                    .collect()
155            });
156
157        out.sort_unstable_by_key(|(conn_id, _)| *conn_id);
158        out
159    }
160
161    pub(crate) fn current_qot_push_intent_epoch_by_cache_key(
162        &self,
163        conn_id: u64,
164        cache_key: &str,
165        sub_type: i32,
166        rehab_type: i32,
167    ) -> Option<u64> {
168        let effective_rehab = if is_kl_sub_type(sub_type) {
169            rehab_type
170        } else {
171            0
172        };
173        self.qot_push_regs
174            .read()
175            .qot_push_regs_by_cache_key
176            .get(cache_key)?
177            .get(&(sub_type, effective_rehab))?
178            .get(&conn_id)
179            .copied()
180    }
181
182    /// Run one non-await publication only while the cache-key mirror still
183    /// names the exact registration intent observed by the caller. The read
184    /// fence prevents unregister/re-register from crossing the terminal
185    /// cursor-check, enqueue, and success-commit sequence.
186    pub(crate) fn with_current_qot_push_intent_by_cache_key<R>(
187        &self,
188        conn_id: u64,
189        cache_key: &str,
190        sub_type: i32,
191        rehab_type: i32,
192        expected_intent_epoch: Option<u64>,
193        publish: impl FnOnce() -> R,
194    ) -> Option<R> {
195        let effective_rehab = if is_kl_sub_type(sub_type) {
196            rehab_type
197        } else {
198            0
199        };
200        let regs = self.qot_push_regs.read();
201        let current_epoch = regs
202            .qot_push_regs_by_cache_key
203            .get(cache_key)
204            .and_then(|by_sub| by_sub.get(&(sub_type, effective_rehab)))
205            .and_then(|subscribers| subscribers.get(&conn_id))
206            .copied();
207        if current_epoch != expected_intent_epoch {
208            return None;
209        }
210        Some(publish())
211    }
212
213    fn get_qot_push_subscribers_inner(
214        &self,
215        key: QotSecurityKey,
216        sub_type: i32,
217        rehab_type: i32,
218    ) -> Vec<u64> {
219        let effective_rehab = if is_kl_sub_type(sub_type) {
220            rehab_type
221        } else {
222            0
223        };
224        match self
225            .qot_push_regs
226            .read()
227            .by_tuple
228            .get(&(key, sub_type, effective_rehab))
229        {
230            Some(subscribers) => subscribers.iter().copied().collect(),
231            None => Vec::new(),
232        }
233    }
234
235    /// **v1.4.106 codex 1131 F3 [P1]**: 是否注册过 push (任意 rehab).
236    pub fn is_push_registered_any_rehab_broker(
237        &self,
238        conn_id: u64,
239        sec_key: &QotSecurityKey,
240        sub_type: i32,
241    ) -> bool {
242        let broker_key = Self::broker_key(sec_key);
243        let pr = self.qot_push_regs.read();
244        pr.by_tuple.iter().any(|((k, st, _rehab), set)| {
245            k == &broker_key && *st == sub_type && set.contains(&conn_id)
246        })
247    }
248
249    /// Capture the current exact registration intent after a multi-effect
250    /// local commit. Returns `None` when another request already removed it.
251    pub fn current_qot_push_registration_lease_broker(
252        &self,
253        conn_id: u64,
254        sec_key: &QotSecurityKey,
255        sub_type: i32,
256        rehab_type: i32,
257    ) -> Option<QotPushRegistrationLease> {
258        let effective_rehab = if is_kl_sub_type(sub_type) {
259            rehab_type
260        } else {
261            0
262        };
263        let key = Self::broker_key(sec_key);
264        let regs = self.qot_push_regs.read();
265        let tuple = (key.clone(), sub_type, effective_rehab);
266        if !regs
267            .by_tuple
268            .get(&tuple)
269            .is_some_and(|connections| connections.contains(&conn_id))
270        {
271            return None;
272        }
273        let intent_epoch = regs
274            .intent_epochs
275            .get(&(key.clone(), sub_type, effective_rehab, conn_id))
276            .copied()?;
277        drop(regs);
278        Some(QotPushRegistrationLease {
279            key: key.clone(),
280            sub_type,
281            rehab_type: effective_rehab,
282            conn_id,
283            intent_epoch,
284            connection_generation: self
285                .connection_generations
286                .get(&conn_id)
287                .map(|generation| *generation)
288                .unwrap_or(0),
289            conn_session: self.get_conn_session_broker(conn_id, &key, sub_type),
290        })
291    }
292
293    /// Run one non-await publication while the registration, session, and
294    /// physical connection generation are all fenced against mutation.
295    pub fn with_current_qot_push_registration_lease<R>(
296        &self,
297        lease: &QotPushRegistrationLease,
298        publish: impl FnOnce() -> R,
299    ) -> Option<R> {
300        let regs = self.qot_push_regs.read();
301        let tuple = (lease.key.clone(), lease.sub_type, lease.rehab_type);
302        if !regs
303            .by_tuple
304            .get(&tuple)
305            .is_some_and(|connections| connections.contains(&lease.conn_id))
306            || regs
307                .intent_epochs
308                .get(&(
309                    lease.key.clone(),
310                    lease.sub_type,
311                    lease.rehab_type,
312                    lease.conn_id,
313                ))
314                .is_none_or(|epoch| *epoch != lease.intent_epoch)
315        {
316            return None;
317        }
318
319        let sessions = self.qot_sub_sessions.read();
320        let current_session = sessions
321            .by_key
322            .get(&(lease.key.clone(), lease.sub_type))
323            .and_then(|by_conn| by_conn.get(&lease.conn_id))
324            .copied()
325            .unwrap_or(1);
326        if current_session != lease.conn_session {
327            return None;
328        }
329
330        let current_generation = self
331            .connection_generations
332            .get(&lease.conn_id)
333            .map(|generation| *generation)
334            .unwrap_or(0);
335        if current_generation != lease.connection_generation {
336            return None;
337        }
338        Some(publish())
339    }
340}
futu_server/subscription/push_regs.rs

futu_server/subscription/
push_regs.rs
