futu_cache/
app_jump_sig.rs

1use std::collections::VecDeque;
2
3use parking_lot::Mutex;
4
5#[derive(Debug, Clone, Copy, PartialEq, Eq)]
6pub enum AppJumpSigPublishError {
7    StaleConnectionEpoch,
8    InvalidExpiry,
9}
10
11struct AppJumpSigTicket {
12    invalid_time_ms: u64,
13    encoded: String,
14}
15
16#[derive(Default)]
17struct AppJumpSigInner {
18    epoch: Option<u64>,
19    tickets: VecDeque<AppJumpSigTicket>,
20}
21
22pub struct AppJumpSigCache {
23    inner: Mutex<AppJumpSigInner>,
24    refill_notify: tokio::sync::Notify,
25}
26
27impl AppJumpSigCache {
28    pub fn new() -> Self {
29        Self {
30            inner: Mutex::new(AppJumpSigInner::default()),
31            refill_notify: tokio::sync::Notify::new(),
32        }
33    }
34
35    /// Adopt a Platform publication epoch and clear tickets owned by an older
36    /// connection. Returns true when replacement occurred.
37    pub fn ensure_epoch(&self, epoch: u64) -> bool {
38        let mut inner = self.inner.lock();
39        if inner.epoch == Some(epoch) {
40            return false;
41        }
42        inner.epoch = Some(epoch);
43        inner.tickets.clear();
44        true
45    }
46
47    pub fn publish_batch(
48        &self,
49        epoch: u64,
50        invalid_time_secs: u32,
51        tickets: Vec<Vec<u8>>,
52    ) -> Result<usize, AppJumpSigPublishError> {
53        let invalid_time_ms = u64::from(invalid_time_secs)
54            .checked_mul(1_000)
55            .ok_or(AppJumpSigPublishError::InvalidExpiry)?;
56        if invalid_time_ms == 0 {
57            return Err(AppJumpSigPublishError::InvalidExpiry);
58        }
59        let mut inner = self.inner.lock();
60        if inner.epoch != Some(epoch) {
61            return Err(AppJumpSigPublishError::StaleConnectionEpoch);
62        }
63        let mut published = 0;
64        for ticket in tickets {
65            if ticket.is_empty() {
66                continue;
67            }
68            inner.tickets.push_back(AppJumpSigTicket {
69                invalid_time_ms,
70                encoded: url_encode(&ticket),
71            });
72            published += 1;
73        }
74        Ok(published)
75    }
76
77    #[must_use]
78    pub fn consume(&self, now_ms: u64, epoch: u64) -> Option<String> {
79        let mut inner = self.inner.lock();
80        if inner.epoch != Some(epoch) {
81            return None;
82        }
83        prune_expired(&mut inner, now_ms);
84        inner.tickets.pop_front().map(|ticket| ticket.encoded)
85    }
86
87    #[must_use]
88    pub fn valid_len(&self, now_ms: u64, epoch: u64) -> usize {
89        let mut inner = self.inner.lock();
90        if inner.epoch != Some(epoch) {
91            return 0;
92        }
93        prune_expired(&mut inner, now_ms);
94        inner.tickets.len()
95    }
96
97    pub fn request_refill(&self) {
98        self.refill_notify.notify_one();
99    }
100
101    pub async fn wait_for_refill_request(&self) {
102        self.refill_notify.notified().await;
103    }
104}
105
106impl Default for AppJumpSigCache {
107    fn default() -> Self {
108        Self::new()
109    }
110}
111
112fn prune_expired(inner: &mut AppJumpSigInner, now_ms: u64) {
113    inner
114        .tickets
115        .retain(|ticket| now_ms < ticket.invalid_time_ms);
116}
117
118fn url_encode(input: &[u8]) -> String {
119    const HEX: &[u8; 16] = b"0123456789ABCDEF";
120    let mut encoded = String::with_capacity(input.len());
121    for &byte in input {
122        if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b'~') {
123            encoded.push(char::from(byte));
124        } else {
125            encoded.push('%');
126            encoded.push(char::from(HEX[usize::from(byte >> 4)]));
127            encoded.push(char::from(HEX[usize::from(byte & 0x0f)]));
128        }
129    }
130    encoded
131}
futu_cache/app_jump_sig.rs

futu_cache/
app_jump_sig.rs
