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
18fn cursor_from_body(proto_id: u32, body: &[u8]) -> Option<KlinePushCursor> {
19    use prost::Message;
20
21    if proto_id != futu_core::proto_id::QOT_UPDATE_KL {
22        return None;
23    }
24    let response = futu_proto::qot_update_kl::Response::decode(body).ok()?;
25    let point = response.s2c?.kl_list.into_iter().last()?;
26    Some(KlinePushCursor {
27        time_key: point.time.clone(),
28        fingerprint: push_body_hash(&point.encode_to_vec()),
29    })
30}
31
32fn cursor_allows(previous: Option<&KlinePushCursor>, candidate: &KlinePushCursor) -> bool {
33    !previous.is_some_and(|previous| {
34        previous.time_key > candidate.time_key
35            || (previous.time_key == candidate.time_key
36                && previous.fingerprint == candidate.fingerprint)
37    })
38}
39
40impl PushDispatcher {
41    /// Send one quote first-push to the target physical connection.
42    pub async fn push_qot_to_conn(&self, conn_id: u64, proto_id: u32, body: Vec<u8>) {
43        self.push_qot_to_conn_internal(conn_id, None, proto_id, body)
44            .await;
45    }
46
47    /// Direct first-push with the ordinary quote route identity required by
48    /// the shared KLine cursor.
49    pub async fn push_qot_to_conn_with_route(
50        &self,
51        conn_id: u64,
52        security_key: &str,
53        sub_type: i32,
54        rehab_type: i32,
55        proto_id: u32,
56        body: Vec<u8>,
57    ) {
58        self.push_qot_to_conn_internal(
59            conn_id,
60            Some((security_key, sub_type, rehab_type)),
61            proto_id,
62            body,
63        )
64        .await;
65    }
66
67    async fn push_qot_to_conn_internal(
68        &self,
69        conn_id: u64,
70        route: Option<(&str, i32, i32)>,
71        proto_id: u32,
72        body: Vec<u8>,
73    ) {
74        if !self.delivery_ready() {
75            return;
76        }
77        let first_ticker_cursor = event_contract_ticker_cursor_from_body(proto_id, &body);
78        let kline_cursor = cursor_from_body(proto_id, &body);
79        let push = self.connections.get(&conn_id).and_then(|conn| {
80            if !should_push_to(&conn, Scope::QotRead, "quote_first") {
81                return None;
82            }
83            let frame = conn.make_frame(proto_id, self.next_push_serial_no(), Bytes::from(body));
84            Some((conn.session_generation, conn.tx.clone(), frame))
85        });
86        if let Some((connection_generation, tx, frame)) = push {
87            let sent = if let (Some((security_key, sub_type, rehab_type)), Some(candidate)) =
88                (route, kline_cursor)
89            {
90                let key = KlineCursorKey {
91                    conn_id,
92                    connection_generation,
93                    security_key: security_key.to_owned(),
94                    sub_type,
95                    rehab_type,
96                };
97                let mut cursors = self.kline_cursors.lock();
98                if !cursor_allows(cursors.get(&key), &candidate) {
99                    return;
100                }
101                let sent = self.try_send_qot_client_frame(tx, frame, sub_type, "push_qot_to_conn");
102                if sent {
103                    cursors.insert(key, candidate);
104                }
105                sent
106            } else {
107                self.try_send_qot_client_frame(tx, frame, 0, "push_qot_to_conn")
108            };
109            if sent && let Some((sec_key, sequence)) = first_ticker_cursor {
110                self.event_contract_cursors
111                    .lock()
112                    .ticker_sequences
113                    .entry((conn_id, sec_key))
114                    .and_modify(|current| *current = (*current).max(sequence))
115                    .or_insert(sequence);
116            }
117        }
118    }
119
120    /// Fan out one ordinary quote event without KLine section metadata.
121    pub async fn push_qot(
122        &self,
123        security_key: &str,
124        sub_type: i32,
125        rehab_type: i32,
126        proto_id: u32,
127        body: Vec<u8>,
128    ) {
129        self.push_qot_with_kline_section(security_key, sub_type, rehab_type, proto_id, body, None)
130            .await;
131    }
132
133    /// Fan out one canonical quote event. `kline_trade_section` is internal
134    /// route metadata used only for C++-compatible native connection filtering.
135    pub async fn push_qot_with_kline_section(
136        &self,
137        security_key: &str,
138        sub_type: i32,
139        rehab_type: i32,
140        proto_id: u32,
141        body: Vec<u8>,
142        kline_trade_section: Option<futu_domain_qot_klrt::KlineTradeSection>,
143    ) {
144        if !self.delivery_ready() {
145            return;
146        }
147        if matches!(
148            proto_id,
149            futu_core::proto_id::QOT_UPDATE_EVENT_CONTRACT_ORDER_BOOK
150                | futu_core::proto_id::QOT_UPDATE_EVENT_CONTRACT_KLINE
151                | futu_core::proto_id::QOT_UPDATE_EVENT_CONTRACT_TICKER
152        ) {
153            self.push_event_contract_qot(security_key, sub_type, rehab_type, proto_id, &body);
154            return;
155        }
156        let kline_cursor = cursor_from_body(proto_id, &body);
157        if !self.external_sinks.is_empty() {
158            let should_send_external = if let Some(candidate) = kline_cursor.as_ref() {
159                let key = (security_key.to_owned(), sub_type, rehab_type);
160                let mut cursors = self.external_kline_cursors.lock();
161                if !cursor_allows(cursors.get(&key), candidate) {
162                    false
163                } else {
164                    cursors.insert(key, candidate.clone());
165                    true
166                }
167            } else {
168                true
169            };
170            if should_send_external {
171                for sink in &self.external_sinks {
172                    sink.on_quote_push(security_key, sub_type, rehab_type, proto_id, &body);
173                }
174            }
175        }
176        let body = Bytes::from(body);
177        let subscribers = self.subscriptions.get_qot_push_subscribers_by_cache_key(
178            security_key,
179            sub_type,
180            rehab_type,
181        );
182        let body_sha1 = FutuFrame::body_sha1(&body);
183        for conn_id in subscribers {
184            if kline_trade_section.is_some_and(|section| {
185                !futu_domain_qot_klrt::kline_trade_section_allows_conn(
186                    section,
187                    self.subscriptions.get_conn_session_by_cache_key(
188                        conn_id,
189                        security_key,
190                        sub_type,
191                    ),
192                )
193            }) {
194                continue;
195            }
196            let Some((connection_generation, tx, frame)) =
197                self.connections.get(&conn_id).and_then(|conn| {
198                    if !should_push_to(&conn, Scope::QotRead, "quote") {
199                        return None;
200                    }
201                    let serial_no = self.next_push_serial_no();
202                    let frame =
203                        conn.make_frame_with_sha1(proto_id, serial_no, body.clone(), body_sha1);
204                    Some((conn.session_generation, conn.tx.clone(), frame))
205                })
206            else {
207                continue;
208            };
209
210            if let Some(candidate) = kline_cursor.clone() {
211                let key = KlineCursorKey {
212                    conn_id,
213                    connection_generation,
214                    security_key: security_key.to_owned(),
215                    sub_type,
216                    rehab_type,
217                };
218                let mut cursors = self.kline_cursors.lock();
219                if !cursor_allows(cursors.get(&key), &candidate) {
220                    continue;
221                }
222                if self.try_send_qot_client_frame(tx, frame, sub_type, "push_qot") {
223                    cursors.insert(key, candidate);
224                }
225            } else {
226                let _ = self.try_send_qot_client_frame(tx, frame, sub_type, "push_qot");
227            }
228        }
229    }
230}
futu_server/push/kline_delivery.rs

futu_server/push/
kline_delivery.rs
