pub(crate) mod debounce;
#[cfg(feature = "progressive-reload")]
pub(crate) mod progressive;
#[cfg(feature = "watch")]
pub(crate) mod fs_watcher;
pub use debounce::AdaptiveDebouncer;
#[cfg(feature = "progressive-reload")]
pub use progressive::{
HealthStatus, ProgressiveReloader, ProgressiveReloaderBuilder, ReloadHealthCheck,
ReloadOutcome, ReloadStrategy,
};
#[cfg(feature = "watch")]
pub use fs_watcher::{FsWatcher, MultiFsWatcher};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::sync::Mutex;
use std::time::Duration;
use crate::error::{ConfersResult, ConfigConfigError};
pub struct WatcherGuard {
running: Arc<AtomicBool>,
task_handle: Mutex<Option<tokio::task::JoinHandle<()>>>,
}
impl WatcherGuard {
pub fn new() -> Self {
Self {
running: Arc::new(AtomicBool::new(false)),
task_handle: Mutex::new(None),
}
}
pub fn from_running(running: Arc<AtomicBool>) -> Self {
Self {
running,
task_handle: Mutex::new(None),
}
}
#[allow(dead_code)]
pub(crate) fn with_task(
running: Arc<AtomicBool>,
task_handle: tokio::task::JoinHandle<()>,
) -> Self {
Self {
running,
task_handle: Mutex::new(Some(task_handle)),
}
}
pub fn is_running(&self) -> bool {
self.running.load(Ordering::SeqCst)
}
pub fn start(&self) {
self.running.store(true, Ordering::SeqCst);
}
pub fn stop(&self) {
self.running.store(false, Ordering::SeqCst);
}
pub fn running_flag(&self) -> &Arc<AtomicBool> {
&self.running
}
pub async fn shutdown(&self, timeout: Duration) -> crate::error::ConfersResult<bool> {
self.stop();
let handle = self.task_handle.lock().unwrap().take();
match handle {
None => Ok(true),
Some(handle) => {
match tokio::time::timeout(timeout, handle).await {
Ok(_) => Ok(true),
Err(_) => Ok(false),
}
}
}
}
#[allow(dead_code)]
pub(crate) fn set_task_handle(&self, handle: tokio::task::JoinHandle<()>) {
*self.task_handle.lock().unwrap() = Some(handle);
}
}
#[cfg(feature = "watch")]
impl WatcherGuard {
#[allow(dead_code)]
pub(crate) async fn lifecycle_start(&self) -> Result<(), ConfigConfigError> {
self.start();
Ok(())
}
#[allow(dead_code)]
pub(crate) async fn lifecycle_stop(&self) -> ConfersResult<()> {
self.shutdown(Duration::from_secs(5)).await?;
Ok(())
}
}
impl Default for WatcherGuard {
fn default() -> Self {
Self::new()
}
}
impl Drop for WatcherGuard {
fn drop(&mut self) {
self.stop();
if let Some(handle) = self.task_handle.get_mut().unwrap().take() {
handle.abort();
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::atomic::AtomicBool;
use std::sync::Arc;
#[tokio::test]
async fn test_shutdown_no_task_handle_returns_true() {
let guard = WatcherGuard::new();
guard.start();
assert!(guard.is_running());
let result = guard.shutdown(Duration::from_secs(5)).await;
assert!(result.is_ok(), "shutdown should return Ok");
assert!(result.unwrap(), "shutdown with no task should return true");
assert!(
!guard.is_running(),
"guard should be stopped after shutdown"
);
}
#[tokio::test]
async fn test_shutdown_task_completes_within_timeout_returns_true() {
let running = Arc::new(AtomicBool::new(true));
let handle = tokio::spawn(async {});
let guard = WatcherGuard::with_task(running, handle);
let result = guard.shutdown(Duration::from_secs(2)).await;
assert!(
result.unwrap(),
"shutdown should return true when task completes within timeout"
);
}
#[tokio::test]
async fn test_shutdown_task_exceeds_timeout_returns_false() {
let running = Arc::new(AtomicBool::new(true));
let handle = tokio::spawn(async {
tokio::time::sleep(Duration::from_secs(2)).await;
});
let guard = WatcherGuard::with_task(running, handle);
let result = guard.shutdown(Duration::from_millis(50)).await;
assert!(
!result.unwrap(),
"shutdown should return false when task exceeds timeout"
);
}
#[tokio::test]
async fn test_set_task_handle_then_shutdown() {
let guard = WatcherGuard::new();
guard.start();
guard.set_task_handle(tokio::spawn(async {}));
let result = guard.shutdown(Duration::from_secs(2)).await;
assert!(
result.unwrap(),
"shutdown should return true for a task that completes via set_task_handle"
);
assert!(!guard.is_running());
}
}
#[derive(Debug, Clone)]
pub struct WatcherConfig {
pub debounce_ms: u64,
pub min_reload_interval_ms: u64,
pub max_consecutive_failures: u32,
pub failure_pause_ms: u64,
pub rollback_on_validation_failure: bool,
}
impl Default for WatcherConfig {
fn default() -> Self {
Self {
debounce_ms: 200,
min_reload_interval_ms: 1000,
max_consecutive_failures: 5,
failure_pause_ms: 30000,
rollback_on_validation_failure: false,
}
}
}
impl WatcherConfig {
pub fn new() -> Self {
Self::default()
}
pub fn builder() -> WatcherConfigBuilder {
WatcherConfigBuilder::new()
}
pub fn with_debounce(mut self, ms: u64) -> Self {
self.debounce_ms = ms;
self
}
pub fn with_min_reload_interval(mut self, ms: u64) -> Self {
self.min_reload_interval_ms = ms;
self
}
pub fn with_max_consecutive_failures(mut self, count: u32) -> Self {
self.max_consecutive_failures = count;
self
}
pub fn with_failure_pause(mut self, ms: u64) -> Self {
self.failure_pause_ms = ms;
self
}
pub fn with_rollback_on_validation_failure(mut self, rollback: bool) -> Self {
self.rollback_on_validation_failure = rollback;
self
}
}
pub struct WatcherConfigBuilder {
debounce_ms: Option<u64>,
min_reload_interval_ms: Option<u64>,
max_consecutive_failures: Option<u32>,
failure_pause_ms: Option<u64>,
rollback_on_validation_failure: Option<bool>,
}
impl WatcherConfigBuilder {
pub fn new() -> Self {
Self {
debounce_ms: None,
min_reload_interval_ms: None,
max_consecutive_failures: None,
failure_pause_ms: None,
rollback_on_validation_failure: None,
}
}
pub fn debounce_ms(mut self, ms: u64) -> Self {
self.debounce_ms = Some(ms);
self
}
pub fn min_reload_interval_ms(mut self, ms: u64) -> Self {
self.min_reload_interval_ms = Some(ms);
self
}
pub fn max_consecutive_failures(mut self, count: u32) -> Self {
self.max_consecutive_failures = Some(count);
self
}
pub fn failure_pause_ms(mut self, ms: u64) -> Self {
self.failure_pause_ms = Some(ms);
self
}
pub fn rollback_on_validation_failure(mut self, rollback: bool) -> Self {
self.rollback_on_validation_failure = Some(rollback);
self
}
pub fn build(self) -> WatcherConfig {
WatcherConfig {
debounce_ms: self.debounce_ms.unwrap_or(200),
min_reload_interval_ms: self.min_reload_interval_ms.unwrap_or(1000),
max_consecutive_failures: self.max_consecutive_failures.unwrap_or(5),
failure_pause_ms: self.failure_pause_ms.unwrap_or(30000),
rollback_on_validation_failure: self.rollback_on_validation_failure.unwrap_or(false),
}
}
}
impl Default for WatcherConfigBuilder {
fn default() -> Self {
Self::new()
}
}