myrddin/file_monitor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Finished coding up the Rust challenge.

Browse Source
This commit is contained in:
Myrddin Dundragon 2025-03-20 02:18:30 -04:00
parent 52c5d40420
commit ded4064f4b
9 changed files with 1419 additions and 152 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

934
Cargo.lock generated
View File

File diff suppressed because it is too large Load Diff

2
Cargo.toml
View File

@ -15,3 +15,5 @@ clap = { version = "4.5.32", features = ["derive"] }
chrono = "0.4.40"
notify = "8.0.0"
tokio = { version = "1.44.1", features = ["macros", "rt-multi-thread", "io-std", "fs", "signal", "sync"] }
console-subscriber = "0.4.1"
home = "0.5.11"

BIN
docs/rust_coding_challenge_filemonitor.pdf Normal file
View File

Binary file not shown.

114
src/dir_monitor.rs
View File

@ -1,8 +1,11 @@
use std::path::{Path, PathBuf};
use notify::{EventKind, RecommendedWatcher, Watcher};
use notify::{RecommendedWatcher, Watcher};
use crate::event::Event;
use crate::monitored_files::MonitoredFiles;
use crate::notify_sender::NotifySender;
use crate::util::strip_path_prefix;
@ -15,10 +18,12 @@ pub struct DirMonitor
/// The directory to monitor.
dir: PathBuf,
/// The files that are being monitored within the directory.
monitored_files: MonitoredFiles,
}
impl DirMonitor
{
/// Create a new directory monitor for the desired directory.
@ -40,7 +45,7 @@ impl DirMonitor
// Start from the directory and add each file to our list,
// then recurse through all sub directories and add them to
// the list and repeat.
match scan_dir(&mut self.monitored_files, &self.dir)
match scan_dir(&self.dir, &mut self.monitored_files, &self.dir)
{
Ok(_) =>
{
@ -61,80 +66,65 @@ impl DirMonitor
/// ```
pub fn print_monitored_files(&self)
{
self.monitored_files.print(&self.dir);
println!("{}", self.monitored_files);
}
///
/// Start monitoring the directory asyncronously and process changes within the directory
/// until a termination message is received.
pub async fn monitor(&mut self, mut term_receiver: tokio::sync::watch::Receiver<bool>)
{
let mut running: bool = true;
//let (notify_sender, notify_receiver) = std::sync::mpsc::channel::<Result<notify::Event, notify::Error>>();
let (notify_sender, notify_receiver) = tokio::sync::mpsc::channel::<Result<notify::Event, notify::Error>>(100);
let mut fs_watcher: RecommendedWatcher = match notify::recommended_watcher(notify_sender)
{
Ok(watcher) => { watcher }
Err(e) => { eprintln!("Unable to create watcher: {}", e); panic!(); }
};
fs_watcher.watch(&self.dir, notify::RecursiveMode::Recursive).unwrap();
// Setup the notify crate to watch the INBOX directory.
let tokio_runtime = tokio::runtime::Handle::current();
let (notify_sender, mut notify_receiver) = tokio::sync::mpsc::channel::<Event>(10);
let wrapped_sender = NotifySender::new(tokio_runtime, notify_sender);
let mut fs_watcher: RecommendedWatcher =
match notify::recommended_watcher(wrapped_sender)
{
Ok(watcher) => { watcher }
Err(e) =>
{
// Just panic because we cannot watch the directories so the program
// would be useless and this saves from having to press CTRL-C.
eprintln!("Unable to create watcher: {}", e);
panic!();
}
};
if let Err(e) = fs_watcher.watch(&self.dir, notify::RecursiveMode::Recursive)
{
// Just panic because we cannot watch the directories so the program
// would be useless and this saves from having to press CTRL-C.
eprintln!("Error trying to watch the directory: {}", e);
panic!();
}
// Loop until termination processing events from the notify watcher.
while running
{
// Listen for file changes until termination signal is received.
match notify_receiver.recv().await
// We are listening for messages from either the notify receiver or
// the termination receiver. When ever one of them comes across we
// will process it.
tokio::select!
{
Ok(msg) =>
// Handle listening for the notify watcher events.
// These are the changes that we care about from the file system.
Some(mut event) = notify_receiver.recv() =>
{
match msg
{
Ok(event) =>
{
match event.kind
{
EventKind::Create(_) =>
{
println!("File Created: {:?}", event.paths);
}
EventKind::Modify(_) =>
{
println!("File Modified: {:?}", event.paths);
}
EventKind::Remove(_) =>
{
println!("File Removed: {:?}", event.paths);
}
_ =>
{
}
}
}
Err(e) =>
{
eprintln!("{}", e);
}
}
event.make_paths_relative(&self.dir);
println!("{}", event);
self.monitored_files.process_event(event);
}
Err(e) =>
{
eprintln!("Error receiving notify event: {}", e);
}
}
// Handle listening for the termination message, the boolean value will
// be changed to false when we are meant to terminate.
match term_receiver.has_changed()
{
Ok(_) =>
// Handle listening for the termination message, the boolean value will
// be changed to false when we are meant to terminate.
_ = term_receiver.changed() =>
{
running = *term_receiver.borrow_and_update();
}
Err(e) =>
{
eprintln!("Unable to receive: {}", e);
}
}
}
}
@ -144,7 +134,7 @@ impl DirMonitor
/// Scans a directory, and all of its sub directories, and creates a list of the files
/// inside and their last modification time.
fn scan_dir(monitored_files: &mut MonitoredFiles, dir: &Path) -> std::io::Result<()>
fn scan_dir(base_dir: &Path, monitored_files: &mut MonitoredFiles, dir: &Path) -> std::io::Result<()>
{
let mut dir_list: Vec<PathBuf> = Vec::new();
@ -169,14 +159,14 @@ fn scan_dir(monitored_files: &mut MonitoredFiles, dir: &Path) -> std::io::Result
{
let last_mod_time: std::time::SystemTime = meta.modified()?;
monitored_files.add(&file.path(), last_mod_time);
monitored_files.add(&strip_path_prefix(&file.path(), base_dir), last_mod_time);
}
}
// Recursively scan the sub directories.
for sub_dir in dir_list
{
scan_dir(monitored_files, &sub_dir)?;
scan_dir(base_dir, monitored_files, &sub_dir)?;
}
Ok(())

87
src/event.rs
View File

@ -1,35 +1,83 @@
use std::time::SystemTime;
use crate::util::strip_path_prefix;
/// The events we are worried about.
///
/// These will be sent as messages in a channel.
pub enum Event
{
///
/// A file has been created.
New
{
///
path: String,
/// The path to the file.
path: std::path::PathBuf,
///
/// The modification time of the file.
mod_time: SystemTime
},
///
/// A file has been modified.
Modify
{
///
path: String,
/// The path to the file.
path: std::path::PathBuf,
///
/// The modification time of the file.
mod_time: SystemTime
},
///
/// A file has been renamed, and it stayed in a monitored directory.
Rename
{
/// The path of the file that was renamed.
from: std::path::PathBuf,
/// The path that the file was renamed to.
to: std::path::PathBuf,
/// The modification time of the file.
mod_time: SystemTime
},
/// A file has been removed.
Delete
{
///
path: String
/// The path of the removed file.
path: std::path::PathBuf
}
}
impl Event
{
/// Take all the paths in the Event and makes them relative
/// to the base_dir directory.
pub fn make_paths_relative(&mut self, base_dir: &std::path::Path)
{
match self
{
Event::New { path, mod_time: _ } =>
{
*path = strip_path_prefix(&path, base_dir);
}
Event::Modify { path, mod_time: _ } =>
{
*path = strip_path_prefix(&path, base_dir);
}
Event::Rename { from, to, mod_time: _ } =>
{
*from = strip_path_prefix(&from, base_dir);
*to = strip_path_prefix(&to, base_dir);
}
Event::Delete { path } =>
{
*path = strip_path_prefix(&path, base_dir);
}
}
}
}
@ -40,17 +88,22 @@ impl std::fmt::Display for Event
{
match self
{
Event::New(_) =>
Event::New { path, mod_time: _ } =>
{
write!(f, "[NEW] {}", self.path)
write!(f, "[NEW] {}", path.display())
}
Event::Modify(_) =>
Event::Modify { path, mod_time: _ } =>
{
write!(f, "[MOD] {}", self.path)
write!(f, "[MOD] {}", path.display())
}
Event::Delete =>
Event::Rename { from, to, mod_time: _ } =>
{
write!(f, "[DEL] {}", self.path)
writeln!(f, "[DEL] {}", from.display())?;
write!(f, "[NEW] {}", to.display())
}
Event::Delete { path } =>
{
write!(f, "[DEL] {}", path.display())
}
}
}

46
src/main.rs
View File

@ -5,6 +5,8 @@
mod dir_monitor;
mod event;
mod monitored_files;
mod notify_sender;
mod util;
@ -33,19 +35,28 @@ struct Options
/// Creates the given directory, and its parents, if it does not exist.
fn create_directory(dir: &str) -> bool
fn create_directory(dir: &str) -> Option<String>
{
match std::fs::create_dir_all(dir)
let mut inbox_dir = String::from(dir);
if inbox_dir.starts_with("~")
{
if let Some(path) = home::home_dir()
{
inbox_dir = inbox_dir.replace("~", &path.display().to_string());
}
}
match std::fs::create_dir_all(inbox_dir.clone())
{
Ok(_) =>
{
true
Some(inbox_dir)
}
Err(e) =>
{
eprintln!("Creating directory error: {}", e);
false
None
}
}
}
@ -72,14 +83,13 @@ async fn main()
{
let options: Options = Options::parse();
println!("Inbox: `{}`", options.inbox_dir);
if create_directory(&options.inbox_dir)
if let Some(inbox_dir) = create_directory(&options.inbox_dir)
{
// This is our keep running channel. False will be sent
// when the application is meant to be closed.
let (sender, receiver) = watch::channel(true);
let mut directory_monitor: DirMonitor = DirMonitor::new(&options.inbox_dir);
let mut directory_monitor: DirMonitor = DirMonitor::new(&inbox_dir);
// Program workflow step 1.
// Async is not needed here. This can be done syncronously because
@ -99,11 +109,20 @@ async fn main()
// preferably a green thread, but a fiber is sufficient
// and less resource intensive.
directory_monitor.monitor(receiver).await;
// Program workflow step 4.
// Async is not needed here. This can be done
// syncronously because it is the final/clean up
// step of the program.
//
// This is done here because we have moved the directory
// monitor inside this task.
directory_monitor.print_monitored_files();
});
// Run until Ctrl-C is pressed.
// Once it is pressed it will send a message to stop.
let term_task = tokio::spawn(async move
let _term_task = tokio::spawn(async move
{
// Program workflow step 3.
// Async with message passing so it can run and listen
@ -115,11 +134,10 @@ async fn main()
// Only need to wait for the monitor task to finish because it will only
// finish once the termination task has finished.
mon_task.await;
// Program workflow step 4.
// Async is not needed here. This can be done syncronously because it
// is the final/clean up step of the program.
directory_monitor.print_monitored_files();
match mon_task.await
{
Ok(_) => {}
Err(e) => { eprintln!("Error during monitoring task: {}", e); }
}
}
}

107
src/monitored_files.rs
View File

@ -1,3 +1,4 @@
use std::collections::HashMap;
use std::path::{PathBuf, Path};
use std::time::SystemTime;
@ -7,104 +8,96 @@ use crate::event::Event;
///
/// The files currently being monitored and their last modification time.
pub struct MonitoredFiles
{
///
mod_dates: Vec<SystemTime>,
///
paths: Vec<PathBuf>
/// A mapping of a Path to the last known modification time.
files: HashMap<PathBuf, SystemTime>
}
impl MonitoredFiles
{
/// Create a new empty set of monitored files.
pub fn new() -> Self
{
MonitoredFiles
{
mod_dates: Vec::new(),
paths: Vec::new()
files: HashMap::new()
}
}
pub fn process_event(&mut self, path: &Path, event: Event)
/// Process an event and change the monitored files.
pub fn process_event(&mut self, event: Event)
{
match event
{
Event::New(creation_time) =>
Event::New { path, mod_time } =>
{
self.add(path, creation_time);
self.add(&path, mod_time);
}
Event::Modify(mod_time) =>
Event::Modify { path, mod_time} =>
{
self.modify(&path, mod_time);
}
Event::Delete =>
Event::Rename { from, to, mod_time } =>
{
self.remove(&from);
self.add(&to, mod_time);
}
Event::Delete { path } =>
{
self.remove(&path);
}
}
}
pub fn add(&mut self, path: &Path, time: SystemTime)
/// Add a file that is to be monitored..
pub fn add(&mut self, path: &Path, mod_time: SystemTime)
{
self.paths.push(PathBuf::from(path));
self.mod_dates.push(time);
self.files.insert(PathBuf::from(path), mod_time);
}
/// Modify the stored modifcation time of a monitored file.
pub fn modify(&mut self, path: &Path, time: SystemTime)
{
if let Some(index) = self.get_path_index(path)
if let Some(entry) = self.files.get_mut(path)
{
self.mod_dates[index] = time;
*entry = time;
}
}
/// Remove a file from being monitored.
pub fn remove(&mut self, path: &Path)
{
if let Some(index) = self.get_path_index(path)
{
self.mod_dates.remove(index);
self.paths.remove(index);
}
}
pub fn print(&self, base: &Path)
{
for file in 0..self.paths.len()
{
let date_time: DateTime<Local> = DateTime::from(self.mod_dates[file]);
match &self.paths[file].strip_prefix(base)
{
Ok(rel_path) =>
{
println!("[{}] {}", date_time.format("%d/%m/%Y %H:%M"), rel_path.display());
}
Err(e) =>
{
eprintln!("Unable to strip base directory: {}", e);
}
}
}
}
fn get_path_index(&self, path: &Path) -> Option<usize>
{
for (index, stored_path) in self.paths.iter().enumerate()
{
if path == stored_path
{
return Some(index);
}
}
None
self.files.remove(path);
}
}
impl std::fmt::Display for MonitoredFiles
{
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result
{
let mut it = self.files.iter().peekable();
while let Some((path, mod_time)) = it.next()
{
let date_time: DateTime<Local> = DateTime::from(*mod_time);
if !it.peek().is_none()
{
writeln!(f, "[{}] {}", date_time.format("%m/%d/%Y %H:%M"), path.display())?;
}
else
{
write!(f, "[{}] {}", date_time.format("%m/%d/%Y %H:%M"), path.display())?;
}
}
Ok(())
}
}

245
src/notify_sender.rs Normal file
View File

@ -0,0 +1,245 @@
use std::path::PathBuf;
use crate::util::get_file_mod_time;
/// NotifySender is a newtype idiom around the tokio mpsc Sender.
///
/// This was done to allow the tokio mpsc Sender to be a Notify crate
/// EventHandler.
pub struct NotifySender
{
/// The tokio runtime to spawn event handling tasks on.
tokio_runtime: tokio::runtime::Handle,
/// The channel to send event messages on.
sender: tokio::sync::mpsc::Sender<crate::event::Event>
}
impl NotifySender
{
/// Create a new type that wraps the tokio mpsc Sender.
pub fn new(tokio_runtime: tokio::runtime::Handle,
sender: tokio::sync::mpsc::Sender<crate::event::Event>)
-> Self
{
NotifySender
{
tokio_runtime,
sender
}
}
}
impl notify::EventHandler for NotifySender
{
fn handle_event(&mut self, event: notify::Result<notify::Event>)
{
match event
{
Ok(notify_event) =>
{
// Process the notify event and turn it into an event
// our crate is interested in, one of our Events. If
// we get a valid one then create a tokio task that will
// send the event out as a message.
if let Some(monitor_event) = process_notify_event(notify_event)
{
let notify_sender = self.sender.clone();
self.tokio_runtime.spawn(async move
{
if let Err(e) = notify_sender.send(monitor_event).await
{
eprintln!("Notify EventHandler: {}", e);
}
});
}
}
Err(e) =>
{
eprintln!("Notify crate error: {}", e);
}
}
}
}
/// Processes the notify crate events into events that our program cares about.
fn process_notify_event(event: notify::Event) -> Option<crate::event::Event>
{
match event.kind
{
notify::EventKind::Create(kind) =>
{
process_create_event(kind, event.paths)
}
notify::EventKind::Modify(kind) =>
{
process_modify_event(kind, event.paths)
}
notify::EventKind::Remove(kind) =>
{
process_remove_event(kind, event.paths)
}
_ => { None }
}
}
/// Process the Create event into a New event.
fn process_create_event(event: notify::event::CreateKind,
paths: Vec<PathBuf>)
-> Option<crate::event::Event>
{
if event == notify::event::CreateKind::File
{
let mod_time: std::time::SystemTime =
match get_file_mod_time(&paths[0])
{
Ok(time) => { time }
Err(e) =>
{
eprintln!("Unable to open file to retrieve modification time: {}",
e);
std::time::SystemTime::now()
}
};
return Some(crate::event::Event::New
{
path: paths[0].clone(),
mod_time: mod_time
});
}
None
}
/// Process the modify event into either a Modify event or a Rename event.
fn process_modify_event(event: notify::event::ModifyKind,
paths: Vec<PathBuf>)
-> Option<crate::event::Event>
{
match event
{
// This is just handling the events. On my system they came across as Any for
// the MetadataKind, but the documentation seemed to hint that modification
// times should be on write events. So for the sake of being done without testing
// this on different platforms the Any modification event will trigger a time lookup.
notify::event::ModifyKind::Any | notify::event::ModifyKind::Metadata(_) =>
{
let mod_time: std::time::SystemTime =
match get_file_mod_time(&paths[0])
{
Ok(time) => { time }
Err(e) =>
{
eprintln!("Unable to open file to retrieve modification time: {}",
e);
std::time::SystemTime::now()
}
};
Some(crate::event::Event::Modify
{
path: paths[0].clone(),
mod_time: mod_time
})
}
// Handling a rename event so that during testing when a file is
// moved it is still tracked.
notify::event::ModifyKind::Name(mode) =>
{
match mode
{
// This is for when a file is renamed and both the to and from
// are in the monitored directories.
notify::event::RenameMode::Both =>
{
let mod_time: std::time::SystemTime =
match get_file_mod_time(&paths[1])
{
Ok(time) => { time }
Err(e) =>
{
eprintln!("Unable to open file to retrieve modification time: {}",
e);
std::time::SystemTime::now()
}
};
Some(crate::event::Event::Rename
{
from: paths[0].clone(),
to: paths[1].clone(),
mod_time: mod_time
})
}
// This is for when a file is renamed and only the from
// file is in the monitored directories.
notify::event::RenameMode::From =>
{
Some(crate::event::Event::Delete
{
path: paths[0].clone(),
})
}
// This is for when a file is renamed and both the to file
// is in the monitored directories.
notify::event::RenameMode::To =>
{
let mod_time: std::time::SystemTime =
match get_file_mod_time(&paths[0])
{
Ok(time) => { time }
Err(e) =>
{
eprintln!("Unable to open file to retrieve modification time: {}",
e);
std::time::SystemTime::now()
}
};
Some(crate::event::Event::New
{
path: paths[0].clone(),
mod_time: mod_time
})
}
_ => { None }
}
}
_ => { None }
}
}
/// Process the remove event into a Delete event.
fn process_remove_event(event: notify::event::RemoveKind,
paths: Vec<PathBuf>)
-> Option<crate::event::Event>
{
if event == notify::event::RemoveKind::File
{
return Some(crate::event::Event::Delete
{
path: paths[0].clone(),
});
}
None
}

36
src/util.rs Normal file
View File

@ -0,0 +1,36 @@
/// Return the modification time of a file.
///
/// If it is unable to get the modification time from the file system
/// it will default to the current system time.
pub fn get_file_mod_time(path: &std::path::Path) -> std::io::Result<std::time::SystemTime>
{
let file = std::fs::File::open(path)?;
let meta = file.metadata()?;
if meta.is_file()
{
return Ok(meta.modified()?);
}
Ok(std::time::SystemTime::now())
}
/// Create a new PathBuf that is the same as path but has the base removed if it can.
pub fn strip_path_prefix(path: &std::path::Path, base: &std::path::Path) -> std::path::PathBuf
{
match &path.strip_prefix(base)
{
Ok(rel_path) =>
{
let mut final_path = std::path::PathBuf::new();
final_path.push("./");
final_path.push(rel_path);
final_path
}
Err(_) =>
{
std::path::PathBuf::from(path)
}
}
}