futu_gateway_core/bridge/
push_delivery.rs

1//! QOT push delivery helpers.
2
3use super::PushEvent;
4
5#[cfg(test)]
6mod tests;
7
8/// v1.4.106 codex 1140 F8: qot push 投递助手. channel full / closed →
9/// 记录 metric (qot_push_dropped_total + per-sub-type 桶) + warn log;
10/// 不再 silent 吞错. 替换原 `let _ = push_tx.try_send(event)` pattern.
11///
12/// `event` 必须是 `PushEvent::QuotePush { sub_type, .. }`。广播型 QOT push
13/// 使用 `try_send_qot_broadcast_with_metric`，统一归桶 0。
14pub fn try_send_qot_push_with_metric(
15    push_tx: &tokio::sync::mpsc::Sender<PushEvent>,
16    event: PushEvent,
17    metrics: &futu_server::metrics::GatewayMetrics,
18) {
19    // 取 sub_type 用于 per-sub-type 桶记录 (在 move event 之前提取).
20    let sub_type = match &event {
21        PushEvent::QuotePush { sub_type, .. } => *sub_type,
22        // 非 QuotePush 不属于按 SubType 拆分的行情路径, 桶 0 = 未知.
23        _ => 0,
24    };
25    try_send_with_qot_drop_metric(push_tx, event, metrics, sub_type, "quote push");
26}
27
28/// 广播型 QOT push 投递助手。
29///
30/// 典型来源是权限变化通知 (CMD6006 / CMD6651) 与到价提醒 (CMD5300 →
31/// Qot_UpdatePriceReminder)。这些 push 没有 SubType，metric 统一归桶 0，但
32/// channel full / closed 仍必须可观测，不能回到 `let _ = try_send(...)`。
33pub(super) fn try_send_qot_broadcast_with_metric(
34    push_tx: &tokio::sync::mpsc::Sender<PushEvent>,
35    proto_id: u32,
36    body: Vec<u8>,
37    metrics: &futu_server::metrics::GatewayMetrics,
38    source: &'static str,
39) {
40    try_send_with_qot_drop_metric(
41        push_tx,
42        PushEvent::BroadcastPush { proto_id, body },
43        metrics,
44        0,
45        source,
46    );
47}
48
49/// 首推缓存直发到单连接的投递助手。
50///
51/// `QuotePushToConn` 不走普通 `(sec_key, sub_type, rehab_type)` fanout，但它仍是
52/// QOT push 流量；队列满 / 关闭时应使用同一组可观测 metric。
53pub fn try_send_qot_to_conn_with_metric(
54    push_tx: &tokio::sync::mpsc::Sender<PushEvent>,
55    conn_id: u64,
56    proto_id: u32,
57    body: Vec<u8>,
58    metrics: &futu_server::metrics::GatewayMetrics,
59    sub_type: i32,
60    source: &'static str,
61) {
62    try_send_with_qot_drop_metric(
63        push_tx,
64        PushEvent::QuotePushToConn {
65            conn_id,
66            proto_id,
67            body,
68        },
69        metrics,
70        sub_type,
71        source,
72    );
73}
74
75fn try_send_with_qot_drop_metric(
76    push_tx: &tokio::sync::mpsc::Sender<PushEvent>,
77    event: PushEvent,
78    metrics: &futu_server::metrics::GatewayMetrics,
79    sub_type: i32,
80    source: &'static str,
81) {
82    if let Err(e) = push_tx.try_send(event) {
83        metrics.record_qot_push_dropped(sub_type);
84        match e {
85            tokio::sync::mpsc::error::TrySendError::Full(_) => {
86                tracing::warn!(
87                    sub_type,
88                    source,
89                    "qot push dropped: push_tx queue full (cache updated, \
90                     subscriber missed event). v1.4.106 audit 1140 F8."
91                );
92            }
93            tokio::sync::mpsc::error::TrySendError::Closed(_) => {
94                tracing::warn!(
95                    sub_type,
96                    source,
97                    "qot push dropped: push_tx channel closed (downstream gone). \
98                     v1.4.106 audit 1140 F8."
99                );
100            }
101        }
102    }
103}
futu_gateway_core/bridge/push_delivery.rs

futu_gateway_core/bridge/
push_delivery.rs
