ntp_timer/
sync.rs

1use crate::config::Config;
2use crate::error::TimerError;
3use crate::ntp;
4use crate::timer::GlobalTimer;
5use lazy_static::lazy_static;
6use std::sync::{Arc, Mutex};
7use std::time::Duration;
8use tracing::{debug, info, warn};
9
10lazy_static! {
11    static ref GLOBAL_TIMER: Arc<Mutex<Option<GlobalTimer>>> = Arc::new(Mutex::new(None));
12    static ref CONFIG: Arc<Mutex<Option<Arc<Config>>>> = Arc::new(Mutex::new(None));
13}
14
15/// Global timer manager providing unified interface for time synchronization and access
16pub struct TimerManager;
17
18impl TimerManager {
19    /// Initializes the global timer (lazy initialization).
20    ///
21    /// On first call, fetches NTP timestamp and starts background sync task.
22    /// Subsequent calls are no-op if already initialized.
23    pub async fn init_global_timer(config: Arc<Config>) -> Result<(), TimerError> {
24        config.validate().map_err(TimerError::ConfigError)?;
25
26        let mut config_guard = CONFIG
27            .lock()
28            .map_err(|e| TimerError::LockError(e.to_string()))?;
29
30        if config_guard.is_none() {
31            *config_guard = Some(Arc::clone(&config));
32        }
33        drop(config_guard);
34
35        let mut timer_guard = GLOBAL_TIMER
36            .lock()
37            .map_err(|e| TimerError::LockError(e.to_string()))?;
38
39        if timer_guard.is_none() {
40            let base_timestamp = ntp::get_ntp_timestamp(&config)?;
41            let timer = GlobalTimer::new(base_timestamp);
42            *timer_guard = Some(timer);
43            drop(timer_guard);
44
45            TimerManager::spawn_background_sync();
46        }
47
48        Ok(())
49    }
50
51    /// Gets the current timestamp via fast local path (<0.1ms).
52    ///
53    /// # Errors
54    /// Returns `TimerError::NotInitialized` if `init_global_timer` was not called.
55    pub fn get_timestamp() -> Result<i64, TimerError> {
56        let timer_guard = GLOBAL_TIMER
57            .lock()
58            .map_err(|e| TimerError::LockError(e.to_string()))?;
59
60        match timer_guard.as_ref() {
61            Some(timer) => Ok(timer.get_current_timestamp()),
62            None => Err(TimerError::NotInitialized),
63        }
64    }
65
66    /// 手动同步 NTP
67    pub async fn manual_sync() -> Result<i64, TimerError> {
68        let config_opt = {
69            let config_guard = CONFIG
70                .lock()
71                .map_err(|e| TimerError::LockError(e.to_string()))?;
72            config_guard.as_ref().map(Arc::clone)
73        };
74
75        let config = config_opt.ok_or(TimerError::NotInitialized)?;
76
77        let new_ntp_timestamp = ntp::get_ntp_timestamp(&config)?;
78
79        let mut timer_guard = GLOBAL_TIMER
80            .lock()
81            .map_err(|e| TimerError::LockError(e.to_string()))?;
82
83        match timer_guard.as_mut() {
84            Some(timer) => {
85                if timer.detect_clock_jump(new_ntp_timestamp, config.clock_jump_threshold_secs) {
86                    timer.update_timestamp(new_ntp_timestamp, true);
87                } else {
88                    timer.update_timestamp(new_ntp_timestamp, false);
89                }
90
91                let timestamp = timer.get_current_timestamp();
92                info!(timestamp, "NTP 同步完成");
93                Ok(timestamp)
94            }
95            None => Err(TimerError::NotInitialized),
96        }
97    }
98
99    /// 启动后台同步任务
100    fn spawn_background_sync() {
101        tokio::spawn(async {
102            loop {
103                let sync_interval = {
104                    let config_guard = CONFIG.lock().ok();
105                    config_guard.and_then(|g| g.as_ref().map(|c| c.sync_interval_secs))
106                };
107
108                if let Some(interval) = sync_interval {
109                    tokio::time::sleep(Duration::from_secs(interval)).await;
110
111                    match TimerManager::manual_sync().await {
112                        Ok(_) => {
113                            debug!("后台同步完成");
114                        }
115                        Err(e) => {
116                            warn!("后台同步失败，继续使用本地计时器: {}", e);
117                        }
118                    }
119                } else {
120                    tokio::time::sleep(Duration::from_secs(1)).await;
121                }
122            }
123        });
124    }
125
126    /// 重置计时器（用于测试）
127    #[cfg(test)]
128    pub fn reset() {
129        if let Ok(mut timer_guard) = GLOBAL_TIMER.lock() {
130            *timer_guard = None;
131        }
132        if let Ok(mut config_guard) = CONFIG.lock() {
133            *config_guard = None;
134        }
135    }
136}
137
138#[cfg(test)]
139mod tests {
140    use super::*;
141
142    #[tokio::test]
143    async fn test_manual_sync_without_init() {
144        TimerManager::reset();
145        let result = TimerManager::manual_sync().await;
146        assert!(result.is_err());
147    }
148}
ntp_timer/sync.rs

ntp_timer/
sync.rs
