futu_server/subscription/

views.rs

use std::collections::{HashMap, HashSet};
use std::hash::Hash;
use std::time::Duration;

use futu_core::qot_stock_key::QotSecurityKey;

use super::SubscriptionManager;

impl SubscriptionManager {
    // ===== Quota =====

    /// per-conn used quota — count 该 conn 的 (security_key, sub_type) 对数.
    pub fn get_conn_used_quota(&self, conn_id: u64) -> u32 {
        self.qot_subs
            .read()
            .iter()
            .filter(|(_, set)| set.contains(&conn_id))
            .count() as u32
    }

    /// per-conn used quota with a caller-provided key classifier.
    ///
    /// C++ 10.7 separates normal subscription quota from option-chain quota:
    /// `QotSubscribe::GetUsedQuota` skips keys resolvable by
    /// `ResolveOptionChainKey`; `GetUsedOptionQuota` handles those separately.
    /// Rust keeps the option classifier outside `SubscriptionManager` because
    /// it depends on `StaticDataCache` metadata.
    pub fn get_conn_used_quota_by<F>(&self, conn_id: u64, include: F) -> u32
    where
        F: Fn(&QotSecurityKey) -> bool,
    {
        self.qot_subs
            .read()
            .iter()
            .filter(|((key, _), set)| set.contains(&conn_id) && include(key))
            .count() as u32
    }

    /// **全局 used quota** — 对齐 C++ `m_nAllUsedQuota` (累加全局唯一
    /// SubKey count). 不重复计 conn — 多 conn 订同 (stock, sub_type) 全局算 1.
    pub fn get_total_used_quota(&self) -> u32 {
        self.qot_subs.read().len() as u32
    }

    /// Global used quota with a caller-provided key classifier.
    pub fn get_total_used_quota_by<F>(&self, include: F) -> u32
    where
        F: Fn(&QotSecurityKey) -> bool,
    {
        self.qot_subs
            .read()
            .iter()
            .filter(|((key, _), set)| !set.is_empty() && include(key))
            .count() as u32
    }

    /// Global option-chain quota, deduped by `(option_chain_key, sub_type)`.
    ///
    /// The supplied closure mirrors C++ `ResolveOptionChainKey`: `None` means
    /// the key is a normal security; `Some(chain)` means it participates in the
    /// option quota bill set.
    pub fn get_total_used_option_quota_by<F, K>(&self, option_bill_key: F) -> u32
    where
        F: Fn(&QotSecurityKey) -> Option<K>,
        K: Eq + Hash,
    {
        let mut bills = HashSet::new();
        for ((key, sub_type), set) in self.qot_subs.read().iter() {
            if set.is_empty() {
                continue;
            }
            if let Some(chain) = option_bill_key(key) {
                bills.insert((chain, *sub_type));
            }
        }
        bills.len() as u32
    }

    /// Per-connection option-chain quota, deduped by `(option_chain_key, sub_type)`.
    pub fn get_conn_used_option_quota_by<F, K>(&self, conn_id: u64, option_bill_key: F) -> u32
    where
        F: Fn(&QotSecurityKey) -> Option<K>,
        K: Eq + Hash,
    {
        let mut bills = HashSet::new();
        for ((key, sub_type), set) in self.qot_subs.read().iter() {
            if !set.contains(&conn_id) {
                continue;
            }
            if let Some(chain) = option_bill_key(key) {
                bills.insert((chain, *sub_type));
            }
        }
        bills.len() as u32
    }

    // ===== Conn-level views =====

    pub fn get_conn_qot_subs(&self, conn_id: u64) -> HashMap<i32, Vec<String>> {
        let qot = self.qot_subs.read();
        let mut result: HashMap<i32, Vec<String>> = HashMap::new();
        for ((key, sub_type), conn_ids) in qot.iter() {
            if conn_ids.contains(&conn_id) {
                result.entry(*sub_type).or_default().push(key.cache_key());
            }
        }
        result
    }

    pub fn get_all_qot_conn_ids(&self) -> HashSet<u64> {
        let qot = self.qot_subs.read();
        let mut ids = HashSet::new();
        for conn_ids in qot.values() {
            ids.extend(conn_ids);
        }
        ids
    }

    /// v1.4.106 codex 0932 F5 [P2]: 获取所有连接 ID（有交易账户订阅的）.
    pub fn get_all_trd_conn_ids(&self) -> HashSet<u64> {
        let trd = self.trd_acc_subs.read();
        let mut ids = HashSet::new();
        for conn_ids in trd.values() {
            ids.extend(conn_ids);
        }
        ids
    }

    /// v1.4.106 codex 1131 F2 [P1]: 计算全局 desired set.
    /// 返 (sec_key_display, sub_type), sec_key_display 是 cache_key 形态
    /// (`"market_code"` or `"market_code@b{id}"`).
    pub fn compute_global_desired_set(&self) -> Vec<(String, i32)> {
        let qot = self.qot_subs.read();
        let mut out = Vec::with_capacity(qot.len());
        for (k, sub_type) in qot.keys() {
            out.push((k.cache_key(), *sub_type));
        }
        out
    }

    /// **v1.4.106 codex 0631 F2 [P2]**: shared global desired-set keys helper.
    ///
    /// 返当前全集 `Vec<(sec_key, sub_type)>` (与 `compute_global_desired_set`
    /// 等价, 命名对齐 codex 0631 audit 习惯). 共享给 SubHandler / unsub /
    /// unsub_all / resubscribe_quotes 三路径, 不再每条 sub 单独发 delta —
    /// 对齐 CMD 6211 set-state 协议 (per-market full set replaces).
    ///
    /// caller 应做的事:
    /// 1. 调本 fn 拿当前全集
    /// 2. 应用 delta (sub: 加; unsub: 减; unsub_all: 移除本 conn 独占的)
    /// 3. resolve sec_key → stock_id (cache); cache miss → loud warn (本 fn 不
    ///    替 caller 决定 — 由 caller 选择 fail-loud 还是 partial-submit)
    /// 4. 调 `submit_global_desired_set(NEW set)` 让 backend ack
    ///
    /// 命名: keys = `(sec_key, sub_type)` 二元组, 不是 `(stock_id, market)` —
    /// stock_id resolve 是 caller 责任 (因为依赖 StaticDataCache).
    #[inline]
    pub fn qot_global_desired_keys(&self) -> Vec<(String, i32)> {
        self.compute_global_desired_set()
    }

    /// C++ delay-statistics gate equivalent for QOT push reporting.
    ///
    /// Ref: `APIServer_Qot_StockBasic.cpp:218-223`,
    /// `APIServer_Qot_OrderBook.cpp:238-243`, and
    /// `APIServer_Qot_Broker.cpp:194-199`: `IsHasSubFewTime(..., 3)` must be
    /// true before `Push_Count_Add`.
    pub fn qot_sub_elapsed_at_least_broker(
        &self,
        sec_key: &QotSecurityKey,
        sub_type: i32,
        elapsed: Duration,
    ) -> bool {
        self.qot_sub_times
            .read()
            .get(&(Self::broker_key(sec_key), sub_type))
            .map(|instant| instant.elapsed() >= elapsed)
            .unwrap_or(true)
    }
}