futu_server/push/
kline_delivery.rs

1use super::*;
2
3#[derive(Clone, Debug, PartialEq, Eq, Hash)]
4pub(super) struct KlineCursorKey {
5    pub(super) conn_id: u64,
6    pub(super) connection_generation: u64,
7    pub(super) security_key: String,
8    pub(super) sub_type: i32,
9    pub(super) rehab_type: i32,
10}
11
12#[derive(Clone, Debug, PartialEq, Eq)]
13pub(super) struct KlinePushCursor {
14    time_key: String,
15    fingerprint: u64,
16}
17
18#[derive(Clone, Debug, PartialEq, Eq)]
19pub(super) struct RegisteredKlinePushCursor {
20    intent_epoch: u64,
21    point: KlinePushCursor,
22}
23
24fn cursor_from_body(proto_id: u32, body: &[u8]) -> Option<KlinePushCursor> {
25    use prost::Message;
26
27    if proto_id != futu_core::proto_id::QOT_UPDATE_KL {
28        return None;
29    }
30    let response = futu_proto::qot_update_kl::Response::decode(body).ok()?;
31    let point = response.s2c?.kl_list.into_iter().last()?;
32    Some(KlinePushCursor {
33        time_key: point.time.clone(),
34        fingerprint: push_body_hash(&point.encode_to_vec()),
35    })
36}
37
38fn cursor_allows(previous: Option<&KlinePushCursor>, candidate: &KlinePushCursor) -> bool {
39    !previous.is_some_and(|previous| {
40        previous.time_key > candidate.time_key
41            || (previous.time_key == candidate.time_key
42                && previous.fingerprint == candidate.fingerprint)
43    })
44}
45
46fn registered_cursor_allows(
47    previous: Option<&RegisteredKlinePushCursor>,
48    intent_epoch: u64,
49    candidate: &KlinePushCursor,
50) -> bool {
51    previous.is_none_or(|previous| {
52        if previous.point.time_key > candidate.time_key {
53            return false;
54        }
55        previous.intent_epoch != intent_epoch || cursor_allows(Some(&previous.point), candidate)
56    })
57}
58
59impl PushDispatcher {
60    /// Send one quote first-push to the target physical connection.
61    pub async fn push_qot_to_conn(&self, conn_id: u64, proto_id: u32, body: Vec<u8>) {
62        self.push_qot_to_conn_internal(conn_id, None, None, None, proto_id, body)
63            .await;
64    }
65
66    /// Direct first-push with the ordinary quote route identity required by
67    /// the shared KLine cursor.
68    pub async fn push_qot_to_conn_with_route(
69        &self,
70        conn_id: u64,
71        security_key: &str,
72        sub_type: i32,
73        rehab_type: i32,
74        proto_id: u32,
75        body: Vec<u8>,
76    ) {
77        self.push_qot_to_conn_internal(
78            conn_id,
79            Some((security_key, sub_type, rehab_type)),
80            None,
81            None,
82            proto_id,
83            body,
84        )
85        .await;
86    }
87
88    /// Direct ordinary KLine first-push with a second terminal lease check at
89    /// the physical client sink, after any queueing delay.
90    #[allow(clippy::too_many_arguments)]
91    pub async fn push_qot_to_conn_with_route_lease(
92        &self,
93        conn_id: u64,
94        expected_connection_generation: u64,
95        registration_lease: crate::subscription::QotPushRegistrationLease,
96        security_key: &str,
97        sub_type: i32,
98        rehab_type: i32,
99        proto_id: u32,
100        body: Vec<u8>,
101    ) {
102        self.push_qot_to_conn_internal(
103            conn_id,
104            Some((security_key, sub_type, rehab_type)),
105            Some(expected_connection_generation),
106            Some(&registration_lease),
107            proto_id,
108            body,
109        )
110        .await;
111    }
112
113    async fn push_qot_to_conn_internal(
114        &self,
115        conn_id: u64,
116        route: Option<(&str, i32, i32)>,
117        expected_connection_generation: Option<u64>,
118        registration_lease: Option<&crate::subscription::QotPushRegistrationLease>,
119        proto_id: u32,
120        body: Vec<u8>,
121    ) {
122        if !self.delivery_ready() {
123            return;
124        }
125        let first_ticker_cursor = event_contract_ticker_cursor_from_body(proto_id, &body);
126        let kline_cursor = cursor_from_body(proto_id, &body);
127        let routed_kline = route.is_some() && kline_cursor.is_some();
128        let deliver = |intent_epoch: u64| {
129            let push = self.connections.get(&conn_id).and_then(|conn| {
130                if expected_connection_generation
131                    .is_some_and(|expected| conn.session_generation != expected)
132                    || !should_push_to(&conn, Scope::QotRead, "quote_first")
133                {
134                    return None;
135                }
136                let frame =
137                    conn.make_frame(proto_id, self.next_push_serial_no(), Bytes::from(body));
138                Some((conn.session_generation, conn.tx.clone(), frame))
139            });
140            if let Some((connection_generation, tx, frame)) = push {
141                let sent = if let (Some((security_key, sub_type, rehab_type)), Some(candidate)) =
142                    (route, kline_cursor)
143                {
144                    let key = KlineCursorKey {
145                        conn_id,
146                        connection_generation,
147                        security_key: security_key.to_owned(),
148                        sub_type,
149                        rehab_type,
150                    };
151                    let mut cursors = self.kline_cursors.lock();
152                    if !registered_cursor_allows(cursors.get(&key), intent_epoch, &candidate) {
153                        return;
154                    }
155                    let sent =
156                        self.try_send_qot_client_frame(tx, frame, sub_type, "push_qot_to_conn");
157                    if sent {
158                        cursors.insert(
159                            key,
160                            RegisteredKlinePushCursor {
161                                intent_epoch,
162                                point: candidate,
163                            },
164                        );
165                    }
166                    sent
167                } else {
168                    self.try_send_qot_client_frame(tx, frame, 0, "push_qot_to_conn")
169                };
170                if sent && let Some((sec_key, sequence)) = first_ticker_cursor {
171                    self.event_contract_cursors
172                        .lock()
173                        .ticker_sequences
174                        .entry((conn_id, sec_key))
175                        .and_modify(|current| *current = (*current).max(sequence))
176                        .or_insert(sequence);
177                }
178            }
179        };
180        if let Some(lease) = registration_lease {
181            let Some((security_key, sub_type, rehab_type)) = route else {
182                return;
183            };
184            if !lease.matches_delivery(
185                conn_id,
186                expected_connection_generation.unwrap_or_default(),
187                security_key,
188                sub_type,
189                rehab_type,
190            ) {
191                return;
192            }
193            let intent_epoch = lease.intent_epoch();
194            let _ = self
195                .subscriptions
196                .with_current_qot_push_registration_lease(lease, || deliver(intent_epoch));
197        } else if routed_kline {
198            let Some((security_key, sub_type, rehab_type)) = route else {
199                return;
200            };
201            let intent_epoch = self
202                .subscriptions
203                .current_qot_push_intent_epoch_by_cache_key(
204                    conn_id,
205                    security_key,
206                    sub_type,
207                    rehab_type,
208                );
209            let _ = self
210                .subscriptions
211                .with_current_qot_push_intent_by_cache_key(
212                    conn_id,
213                    security_key,
214                    sub_type,
215                    rehab_type,
216                    intent_epoch,
217                    || deliver(intent_epoch.unwrap_or(0)),
218                );
219        } else {
220            deliver(0);
221        }
222    }
223
224    /// Fan out one ordinary quote event without KLine section metadata.
225    pub async fn push_qot(
226        &self,
227        security_key: &str,
228        sub_type: i32,
229        rehab_type: i32,
230        proto_id: u32,
231        body: Vec<u8>,
232    ) {
233        self.push_qot_with_kline_section(security_key, sub_type, rehab_type, proto_id, body, None)
234            .await;
235    }
236
237    /// Fan out one canonical quote event. `kline_trade_section` is internal
238    /// route metadata used only for C++-compatible native connection filtering.
239    pub async fn push_qot_with_kline_section(
240        &self,
241        security_key: &str,
242        sub_type: i32,
243        rehab_type: i32,
244        proto_id: u32,
245        body: Vec<u8>,
246        kline_trade_section: Option<futu_domain_qot_klrt::KlineTradeSection>,
247    ) {
248        if !self.delivery_ready() {
249            return;
250        }
251        if matches!(
252            proto_id,
253            futu_core::proto_id::QOT_UPDATE_EVENT_CONTRACT_ORDER_BOOK
254                | futu_core::proto_id::QOT_UPDATE_EVENT_CONTRACT_KLINE
255                | futu_core::proto_id::QOT_UPDATE_EVENT_CONTRACT_TICKER
256        ) {
257            self.push_event_contract_qot(security_key, sub_type, rehab_type, proto_id, &body);
258            return;
259        }
260        let kline_cursor = cursor_from_body(proto_id, &body);
261        if !self.external_sinks.is_empty() {
262            let should_send_external = if let Some(candidate) = kline_cursor.as_ref() {
263                let key = (security_key.to_owned(), sub_type, rehab_type);
264                let mut cursors = self.external_kline_cursors.lock();
265                if !cursor_allows(cursors.get(&key), candidate) {
266                    false
267                } else {
268                    cursors.insert(key, candidate.clone());
269                    true
270                }
271            } else {
272                true
273            };
274            if should_send_external {
275                for sink in &self.external_sinks {
276                    sink.on_quote_push(security_key, sub_type, rehab_type, proto_id, &body);
277                }
278            }
279        }
280        let body = Bytes::from(body);
281        let subscribers = self
282            .subscriptions
283            .get_qot_push_subscriber_intents_by_cache_key(security_key, sub_type, rehab_type);
284        let body_sha1 = FutuFrame::body_sha1(&body);
285        for (conn_id, intent_epoch) in subscribers {
286            if kline_trade_section.is_some_and(|section| {
287                !futu_domain_qot_klrt::kline_trade_section_allows_conn(
288                    section,
289                    self.subscriptions.get_conn_session_by_cache_key(
290                        conn_id,
291                        security_key,
292                        sub_type,
293                    ),
294                )
295            }) {
296                continue;
297            }
298            let Some((connection_generation, tx, frame)) =
299                self.connections.get(&conn_id).and_then(|conn| {
300                    if !should_push_to(&conn, Scope::QotRead, "quote") {
301                        return None;
302                    }
303                    let serial_no = self.next_push_serial_no();
304                    let frame =
305                        conn.make_frame_with_sha1(proto_id, serial_no, body.clone(), body_sha1);
306                    Some((conn.session_generation, conn.tx.clone(), frame))
307                })
308            else {
309                continue;
310            };
311
312            if let Some(candidate) = kline_cursor.clone() {
313                let key = KlineCursorKey {
314                    conn_id,
315                    connection_generation,
316                    security_key: security_key.to_owned(),
317                    sub_type,
318                    rehab_type,
319                };
320                let _ = self
321                    .subscriptions
322                    .with_current_qot_push_intent_by_cache_key(
323                        conn_id,
324                        security_key,
325                        sub_type,
326                        rehab_type,
327                        Some(intent_epoch),
328                        || {
329                            let mut cursors = self.kline_cursors.lock();
330                            if !registered_cursor_allows(
331                                cursors.get(&key),
332                                intent_epoch,
333                                &candidate,
334                            ) {
335                                return;
336                            }
337                            if self.try_send_qot_client_frame(tx, frame, sub_type, "push_qot") {
338                                cursors.insert(
339                                    key,
340                                    RegisteredKlinePushCursor {
341                                        intent_epoch,
342                                        point: candidate,
343                                    },
344                                );
345                            }
346                        },
347                    );
348            } else {
349                let _ = self.try_send_qot_client_frame(tx, frame, sub_type, "push_qot");
350            }
351        }
352    }
353}
futu_server/push/kline_delivery.rs

futu_server/push/
kline_delivery.rs
