use crate::value::ConfigValue;
use serde::{Deserialize, Serialize};
use std::time::SystemTime;
use tokio::sync::broadcast;
use tracing::debug;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConfigChange {
pub key: String,
pub old_value: Option<ConfigValue>,
pub new_value: ConfigValue,
pub timestamp: SystemTime,
pub changed_by: String,
}
pub struct ConfigWatcher {
key_pattern: String,
receiver: broadcast::Receiver<ConfigChange>,
}
impl ConfigWatcher {
pub fn new(key_pattern: String, receiver: broadcast::Receiver<ConfigChange>) -> Self {
Self {
key_pattern,
receiver,
}
}
pub async fn next(&mut self) -> Option<ConfigChange> {
loop {
match self.receiver.recv().await {
Ok(change) => {
if self.matches_pattern(&change.key) {
debug!(
"Configuration change matched pattern '{}': {}",
self.key_pattern, change.key
);
return Some(change);
}
}
Err(broadcast::error::RecvError::Closed) => {
debug!("Configuration change channel closed");
return None;
}
Err(broadcast::error::RecvError::Lagged(_)) => {
debug!("Configuration watcher lagged, continuing...");
continue;
}
}
}
}
pub fn pattern(&self) -> &str {
&self.key_pattern
}
fn matches_pattern(&self, key: &str) -> bool {
matches_pattern(&self.key_pattern, key)
}
}
fn matches_pattern(pattern: &str, key: &str) -> bool {
if pattern == key {
return true;
}
if pattern.is_empty() {
return true;
}
if pattern.contains('*') {
return matches_wildcard_pattern(pattern, key);
}
if pattern.ends_with('.') {
return key.starts_with(pattern);
}
let pattern_with_dot = format!("{pattern}.");
if key.starts_with(&pattern_with_dot) {
return true;
}
false
}
fn matches_wildcard_pattern(pattern: &str, key: &str) -> bool {
let pattern_parts: Vec<&str> = pattern.split('*').collect();
if pattern_parts.len() == 1 {
return pattern == key;
}
let mut key_pos = 0;
for (i, part) in pattern_parts.iter().enumerate() {
if part.is_empty() {
continue;
}
if i == 0 {
if !key[key_pos..].starts_with(part) {
return false;
}
key_pos += part.len();
} else if i == pattern_parts.len() - 1 {
return key[key_pos..].ends_with(part);
} else {
if let Some(pos) = key[key_pos..].find(part) {
key_pos += pos + part.len();
} else {
return false;
}
}
}
true
}
pub struct WatcherBuilder {
patterns: Vec<String>,
}
impl WatcherBuilder {
pub fn new() -> Self {
Self {
patterns: Vec::new(),
}
}
pub fn watch<S: Into<String>>(mut self, pattern: S) -> Self {
self.patterns.push(pattern.into());
self
}
pub fn build(self, receiver: broadcast::Receiver<ConfigChange>) -> Vec<ConfigWatcher> {
self.patterns
.into_iter()
.map(|pattern| ConfigWatcher::new(pattern, receiver.resubscribe()))
.collect()
}
}
impl Default for WatcherBuilder {
fn default() -> Self {
Self::new()
}
}
pub struct ChangeFilter {
include_patterns: Vec<String>,
exclude_patterns: Vec<String>,
debounce_duration: Option<std::time::Duration>,
last_notification: Option<SystemTime>,
}
impl ChangeFilter {
pub fn new() -> Self {
Self {
include_patterns: Vec::new(),
exclude_patterns: Vec::new(),
debounce_duration: None,
last_notification: None,
}
}
pub fn include<S: Into<String>>(mut self, pattern: S) -> Self {
self.include_patterns.push(pattern.into());
self
}
pub fn exclude<S: Into<String>>(mut self, pattern: S) -> Self {
self.exclude_patterns.push(pattern.into());
self
}
pub fn debounce(mut self, duration: std::time::Duration) -> Self {
self.debounce_duration = Some(duration);
self
}
pub fn should_process(&mut self, change: &ConfigChange) -> bool {
if let Some(debounce_duration) = self.debounce_duration {
if let Some(last_time) = self.last_notification {
if change
.timestamp
.duration_since(last_time)
.unwrap_or_default()
< debounce_duration
{
return false;
}
}
}
for exclude_pattern in &self.exclude_patterns {
if matches_pattern(exclude_pattern, &change.key) {
return false;
}
}
if self.include_patterns.is_empty() {
self.last_notification = Some(change.timestamp);
return true;
}
for include_pattern in &self.include_patterns {
if matches_pattern(include_pattern, &change.key) {
self.last_notification = Some(change.timestamp);
return true;
}
}
false
}
}
impl Default for ChangeFilter {
fn default() -> Self {
Self::new()
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::time::Duration;
use tokio::sync::broadcast;
#[test]
fn test_pattern_matching() {
assert!(matches_pattern("app.database", "app.database"));
assert!(matches_pattern("app.database", "app.database.host"));
assert!(matches_pattern("app.", "app.database.host"));
assert!(matches_pattern("", "anything"));
assert!(matches_pattern("app.*", "app.database"));
assert!(matches_pattern("app.*", "app.cache"));
assert!(matches_pattern("*.host", "database.host"));
assert!(matches_pattern("*.host", "cache.host"));
assert!(!matches_pattern("app.database", "app.cache"));
assert!(!matches_pattern("app.database.host", "app.database"));
assert!(!matches_pattern("database", "app.database"));
}
#[tokio::test]
async fn test_config_watcher() {
let (tx, rx) = broadcast::channel(10);
let mut watcher = ConfigWatcher::new("app.database".to_string(), rx);
let change = ConfigChange {
key: "app.database.host".to_string(),
old_value: None,
new_value: ConfigValue::String("localhost".to_string()),
timestamp: SystemTime::now(),
changed_by: "test".to_string(),
};
tx.send(change.clone()).unwrap();
let received = watcher.next().await.unwrap();
assert_eq!(received.key, "app.database.host");
}
#[tokio::test]
async fn test_change_filter() {
let mut filter = ChangeFilter::new()
.include("app.*")
.exclude("app.secret.*")
.debounce(Duration::from_millis(100));
let change1 = ConfigChange {
key: "app.database.host".to_string(),
old_value: None,
new_value: ConfigValue::String("localhost".to_string()),
timestamp: SystemTime::now(),
changed_by: "test".to_string(),
};
let change2 = ConfigChange {
key: "app.secret.key".to_string(),
old_value: None,
new_value: ConfigValue::String("secret".to_string()),
timestamp: SystemTime::now(),
changed_by: "test".to_string(),
};
assert!(filter.should_process(&change1));
assert!(!filter.should_process(&change2));
let change3 = ConfigChange {
key: "app.database.port".to_string(),
old_value: None,
new_value: ConfigValue::Integer(5432),
timestamp: SystemTime::now(),
changed_by: "test".to_string(),
};
assert!(!filter.should_process(&change3)); }
#[test]
fn test_wildcard_matching() {
assert!(matches_wildcard_pattern("app.*", "app.database"));
assert!(matches_wildcard_pattern("app.*", "app.cache"));
assert!(matches_wildcard_pattern("*.host", "database.host"));
assert!(matches_wildcard_pattern("app.*.host", "app.database.host"));
assert!(matches_wildcard_pattern("*", "anything"));
assert!(!matches_wildcard_pattern("app.*", "database.host"));
assert!(!matches_wildcard_pattern("*.host", "database.port"));
assert!(!matches_wildcard_pattern("app.*.host", "app.database.port"));
}
}