unistore_watcher/

watcher.rs

//! 【文件监控器】- 核心监控实现
//!
//! 职责：
//! - 管理文件系统监控
//! - 处理事件分发

use crate::config::WatcherConfig;
use crate::deps::*;
use crate::error::WatcherError;
use crate::event::FileEvent;
use std::sync::atomic::{AtomicBool, Ordering};
use tracing::{error, info, warn};

/// 文件监控器
pub struct FileWatcher {
    /// 底层 notify 监控器
    watcher: RecommendedWatcher,
    /// 配置
    config: WatcherConfig,
    /// 已监控的路径
    watched_paths: Arc<RwLock<HashSet<PathBuf>>>,
    /// 事件广播发送器
    event_tx: broadcast::Sender<FileEvent>,
    /// 是否已关闭
    closed: Arc<AtomicBool>,
}

impl FileWatcher {
    /// 创建新的文件监控器
    pub fn new(config: WatcherConfig) -> Result<Self, WatcherError> {
        let (event_tx, _) = broadcast::channel(config.buffer_size);
        let event_tx_clone = event_tx.clone();

        // 创建 notify 监控器
        let watcher = notify::recommended_watcher(move |res: Result<NotifyEvent, notify::Error>| {
            match res {
                Ok(event) => {
                    if let Some(file_event) = FileEvent::from_notify(event) {
                        // 忽略发送失败（没有订阅者）
                        let _ = event_tx_clone.send(file_event);
                    }
                }
                Err(e) => {
                    error!(error = ?e, "监控错误");
                }
            }
        })
        .map_err(WatcherError::NotifyError)?;

        info!("文件监控器已创建");

        Ok(Self {
            watcher,
            config,
            watched_paths: Arc::new(RwLock::new(HashSet::new())),
            event_tx,
            closed: Arc::new(AtomicBool::new(false)),
        })
    }

    /// 使用默认配置创建
    pub fn with_defaults() -> Result<Self, WatcherError> {
        Self::new(WatcherConfig::default())
    }

    /// 添加监控路径
    pub fn watch<P: AsRef<Path>>(&mut self, path: P) -> Result<(), WatcherError> {
        let path = path.as_ref().to_path_buf();

        if self.closed.load(Ordering::SeqCst) {
            return Err(WatcherError::WatcherClosed);
        }

        // 检查路径是否存在
        if !path.exists() {
            return Err(WatcherError::PathNotFound(path));
        }

        // 检查是否已监控
        {
            let watched = self.watched_paths.read();
            if watched.contains(&path) {
                return Err(WatcherError::AlreadyWatched(path));
            }
        }

        // 确定递归模式
        let mode = if self.config.recursive {
            RecursiveMode::Recursive
        } else {
            RecursiveMode::NonRecursive
        };

        // 添加监控
        self.watcher.watch(&path, mode)?;

        // 记录路径
        self.watched_paths.write().insert(path.clone());

        info!(path = ?path, recursive = self.config.recursive, "添加监控路径");

        Ok(())
    }

    /// 移除监控路径
    pub fn unwatch<P: AsRef<Path>>(&mut self, path: P) -> Result<(), WatcherError> {
        let path = path.as_ref().to_path_buf();

        if self.closed.load(Ordering::SeqCst) {
            return Err(WatcherError::WatcherClosed);
        }

        // 检查是否已监控
        {
            let watched = self.watched_paths.read();
            if !watched.contains(&path) {
                return Err(WatcherError::NotWatched(path));
            }
        }

        // 移除监控
        self.watcher.unwatch(&path)?;

        // 移除记录
        self.watched_paths.write().remove(&path);

        info!(path = ?path, "移除监控路径");

        Ok(())
    }

    /// 获取事件接收器
    pub fn subscribe(&self) -> broadcast::Receiver<FileEvent> {
        self.event_tx.subscribe()
    }

    /// 获取已监控的路径列表
    pub fn watched_paths(&self) -> Vec<PathBuf> {
        self.watched_paths.read().iter().cloned().collect()
    }

    /// 检查路径是否被监控
    pub fn is_watching<P: AsRef<Path>>(&self, path: P) -> bool {
        self.watched_paths.read().contains(path.as_ref())
    }

    /// 获取监控路径数量
    pub fn watch_count(&self) -> usize {
        self.watched_paths.read().len()
    }

    /// 检查监控器是否已关闭
    pub fn is_closed(&self) -> bool {
        self.closed.load(Ordering::SeqCst)
    }

    /// 关闭监控器
    pub fn close(&mut self) {
        if self.closed.swap(true, Ordering::SeqCst) {
            return; // 已关闭
        }

        // 移除所有监控
        let paths: Vec<PathBuf> = self.watched_paths.read().iter().cloned().collect();
        for path in paths {
            if let Err(e) = self.watcher.unwatch(&path) {
                warn!(path = ?path, error = ?e, "移除监控失败");
            }
        }

        self.watched_paths.write().clear();
        info!("文件监控器已关闭");
    }

    /// 获取配置引用
    pub fn config(&self) -> &WatcherConfig {
        &self.config
    }
}

impl Drop for FileWatcher {
    fn drop(&mut self) {
        self.close();
    }
}

/// 简化的监控 API
pub async fn watch_path<P, F>(path: P, mut handler: F) -> Result<(), WatcherError>
where
    P: AsRef<Path>,
    F: FnMut(FileEvent) + Send + 'static,
{
    let mut watcher = FileWatcher::with_defaults()?;
    watcher.watch(path)?;

    let mut rx = watcher.subscribe();

    while let Ok(event) = rx.recv().await {
        handler(event);
    }

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::tempdir;

    #[test]
    fn test_watcher_creation() {
        let watcher = FileWatcher::with_defaults();
        assert!(watcher.is_ok());
    }

    #[test]
    fn test_watch_nonexistent_path() {
        let mut watcher = FileWatcher::with_defaults().unwrap();
        let result = watcher.watch("/nonexistent/path/that/does/not/exist");
        assert!(result.is_err());
        assert!(matches!(result.unwrap_err(), WatcherError::PathNotFound(_)));
    }

    #[test]
    fn test_watch_and_unwatch() {
        let dir = tempdir().unwrap();
        let mut watcher = FileWatcher::with_defaults().unwrap();

        // 添加监控
        watcher.watch(dir.path()).unwrap();
        assert!(watcher.is_watching(dir.path()));
        assert_eq!(watcher.watch_count(), 1);

        // 移除监控
        watcher.unwatch(dir.path()).unwrap();
        assert!(!watcher.is_watching(dir.path()));
        assert_eq!(watcher.watch_count(), 0);
    }

    #[test]
    fn test_double_watch() {
        let dir = tempdir().unwrap();
        let mut watcher = FileWatcher::with_defaults().unwrap();

        watcher.watch(dir.path()).unwrap();
        let result = watcher.watch(dir.path());
        assert!(result.is_err());
        assert!(matches!(result.unwrap_err(), WatcherError::AlreadyWatched(_)));
    }

    #[test]
    fn test_close_watcher() {
        let dir = tempdir().unwrap();
        let mut watcher = FileWatcher::with_defaults().unwrap();

        watcher.watch(dir.path()).unwrap();
        assert!(!watcher.is_closed());

        watcher.close();
        assert!(watcher.is_closed());
        assert_eq!(watcher.watch_count(), 0);
    }
}