small_bin/
watcher.rs

1use crate::{
2    config::AppConfig,
3    database::{Database, QueueItem},
4    *,
5};
6use anyhow::Result;
7use notify::{
8    Config, Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher,
9    event::{MetadataKind, ModifyKind, RenameMode},
10};
11use std::{path::Path, sync::Arc};
12use tokio::sync::mpsc;
13
14
15#[derive(Debug)]
16pub struct FileWatcher {
17    config: Arc<AppConfig>,
18    database: Arc<Database>,
19}
20
21
22impl FileWatcher {
23    pub fn new(config: Arc<AppConfig>, database: Arc<Database>) -> Self {
24        FileWatcher {
25            config,
26            database,
27        }
28    }
29
30
31    pub async fn start(self: Arc<Self>) -> Result<()> {
32        info!("Launching Small Filesystem Handler");
33        let config = self
34            .config
35            .select_config()
36            .expect("One of configs should always be selected!");
37        info!("Watching path: {}", config.watch_path);
38
39        let (tx, mut rx) = mpsc::unbounded_channel();
40        let mut watcher = RecommendedWatcher::new(
41            move |res: Result<Event, notify::Error>| {
42                if let Ok(event) = res {
43                    let _ = tx.send(event);
44                }
45            },
46            Config::default(),
47        )?;
48
49        watcher.watch(Path::new(&config.watch_path), RecursiveMode::NonRecursive)?;
50        info!("Filesystem non-recursive events watcher initialized");
51
52        // Keep watcher alive and process events
53        while let Some(event) = rx.recv().await {
54            if let Err(e) = self.handle_event(event).await {
55                error!("Error handling file event: {e:?}");
56            }
57        }
58
59        Ok(())
60    }
61
62
63    async fn handle_event(&self, event: Event) -> Result<()> {
64        for path in event.paths {
65            let path_str = path.to_string_lossy().to_string();
66            debug!("Handling event: {:?} for path {path_str}", event.kind);
67            if !path.exists() || !path.is_file() {
68                debug!(
69                    "Path doesn't exist or is not a file: {path_str} after event {:?}. Skipped process_event",
70                    event.kind
71                );
72                continue;
73            }
74            if TEMP_PATTERN.is_match(&path_str) {
75                debug!("{path_str} matches temp file name! Skipping");
76                continue;
77            }
78
79            // Wait for file to be fully written
80            if let Err(e) = wait_for_file(&path).await {
81                error!("Error waiting for file {path_str}: {e:?}");
82                continue;
83            }
84
85            // NOTE: do not trigger metadata change events
86            if event.kind == EventKind::Modify(ModifyKind::Metadata(MetadataKind::Any)) {
87                debug!("Skipping event kind: {:?}", event.kind);
88                continue;
89            }
90
91            self.process_event(&path_str)?;
92        }
93
94        Ok(())
95    }
96
97
98    fn process_event(&self, file_path: &str) -> Result<()> {
99        debug!("Processing event for path: {file_path}");
100        let config = self
101            .config
102            .select_config()
103            .expect("One of configs should always be selected!");
104        let uuid_from_file =
105            uuid::Uuid::new_v3(&uuid::Uuid::NAMESPACE_OID, file_path.as_bytes()).to_string();
106        let remote_dest_file = format!("{}/{uuid_from_file}", config.remote_path);
107
108        // Add to queue
109        let queue_item = QueueItem {
110            local_file: file_path.to_string(),
111            remote_file: remote_dest_file,
112            uuid: uuid_from_file,
113        };
114        self.database.add_to_queue(&queue_item)?;
115        debug!("Added file to queue: {file_path}");
116        Ok(())
117    }
118}
small_bin/watcher.rs

small_bin/
watcher.rs
