futu_grpc/server/
push.rs

1use tokio::sync::broadcast;
2
3use futu_server::push::ExternalPushSink;
4
5use crate::proto::PushEvent;
6
7/// gRPC 推送广播器
8///
9/// 实现 `ExternalPushSink` trait，接收 PushDispatcher 的推送事件，
10/// 通过 broadcast channel 分发给所有 SubscribePush 流式连接。
11#[derive(Clone)]
12pub struct GrpcPushBroadcaster {
13    tx: broadcast::Sender<PushEvent>,
14}
15
16impl GrpcPushBroadcaster {
17    pub fn new(capacity: usize) -> Self {
18        let (tx, _) = broadcast::channel(capacity);
19        Self { tx }
20    }
21
22    /// 创建接收端
23    pub fn subscribe(&self) -> broadcast::Receiver<PushEvent> {
24        self.tx.subscribe()
25    }
26
27    fn has_receivers(&self) -> bool {
28        self.tx.receiver_count() > 0
29    }
30
31    fn send(&self, event: PushEvent) {
32        if !self.has_receivers() {
33            return;
34        }
35        let proto_id = event.proto_id;
36        let event_type = event.event_type.clone();
37        if self.tx.send(event).is_err() {
38            tracing::debug!(
39                proto_id,
40                event_type,
41                receiver_count = self.tx.receiver_count(),
42                "grpc push broadcast send skipped"
43            );
44        }
45    }
46}
47
48impl ExternalPushSink for GrpcPushBroadcaster {
49    /// **v1.4.106 codex 1131 F4 [P1]**: `rehab_type` 透传到 gRPC PushEvent.
50    /// gRPC 当前 broadcast 模型 (所有 qot:read subscriber 收到所有 quote
51    /// event), per-conn (sec_key, sub_type, rehab_type) 三元 filter 是 raw TCP
52    /// 专属. rehab_type 通过 PushEvent 让客户端可见.
53    fn on_quote_push(
54        &self,
55        sec_key: &str,
56        sub_type: i32,
57        rehab_type: i32,
58        proto_id: u32,
59        body: &[u8],
60    ) {
61        if !self.has_receivers() {
62            return;
63        }
64        self.send(PushEvent {
65            proto_id,
66            sec_key: sec_key.to_string(),
67            sub_type,
68            rehab_type,
69            body: body.to_vec(),
70            event_type: "quote".to_string(),
71            acc_id: 0,
72            trd_market: String::new(), // 行情推送无 trd_market
73        });
74    }
75
76    fn on_broadcast_push(&self, proto_id: u32, body: &[u8]) {
77        if !self.has_receivers() {
78            return;
79        }
80        self.send(PushEvent {
81            proto_id,
82            sec_key: String::new(),
83            sub_type: 0,
84            rehab_type: 0,
85            body: body.to_vec(),
86            event_type: "notify".to_string(),
87            acc_id: 0,
88            trd_market: String::new(),
89        });
90    }
91
92    fn on_private_user_push(&self, proto_id: u32, body: &[u8]) {
93        if !self.has_receivers() {
94            return;
95        }
96        self.send(PushEvent {
97            proto_id,
98            sec_key: String::new(),
99            sub_type: 0,
100            rehab_type: 0,
101            body: body.to_vec(),
102            event_type: "private_user".to_string(),
103            acc_id: 0,
104            trd_market: String::new(),
105        });
106    }
107
108    /// v1.4.105 D3 (Phase 4) T-B3: trade push trd_market 由 PushDispatcher 一
109    /// 次 decode 后透传, 不再各 sink 独立 decode body. 空 / unknown -> 空
110    /// 字符串 (PushEvent proto3 默认值, 老 client 解析兼容).
111    fn on_trade_push(&self, acc_id: u64, proto_id: u32, body: &[u8], trd_market: Option<&str>) {
112        if !self.has_receivers() {
113            return;
114        }
115        self.send(PushEvent {
116            proto_id,
117            sec_key: String::new(),
118            sub_type: 0,
119            rehab_type: 0,
120            body: body.to_vec(),
121            event_type: "trade".to_string(),
122            acc_id,
123            trd_market: trd_market.unwrap_or("").to_string(),
124        });
125    }
126}
futu_grpc/server/push.rs

futu_grpc/server/
push.rs
