mirror of
https://github.com/encounter/notify.git
synced 2026-07-10 12:18:48 -07:00
PollWatcher: Immediately scan path when adding a watch
This commit is contained in:
Daniel Faust
+132
-125
@@ -4,7 +4,7 @@
|
||||
//! cross-platform APIs; it should function on any platform that the Rust standard library does.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::sync::{Arc, RwLock};
|
||||
use std::sync::{Arc, RwLock, Mutex};
|
||||
use std::sync::mpsc::Sender;
|
||||
use std::fs;
|
||||
use std::thread;
|
||||
@@ -18,10 +18,20 @@ use filetime::FileTime;
|
||||
extern crate walkdir;
|
||||
extern crate time;
|
||||
|
||||
struct PathData {
|
||||
mtime: u64,
|
||||
last_check: f64,
|
||||
}
|
||||
|
||||
struct WatchData {
|
||||
is_recursive: bool,
|
||||
paths: HashMap<PathBuf, PathData>,
|
||||
}
|
||||
|
||||
/// Polling based `Watcher` implementation
|
||||
pub struct PollWatcher {
|
||||
tx: Sender<Event>,
|
||||
watches: Arc<RwLock<HashMap<PathBuf, bool>>>,
|
||||
watches: Arc<Mutex<HashMap<PathBuf, WatchData>>>,
|
||||
open: Arc<RwLock<bool>>,
|
||||
}
|
||||
|
||||
@@ -30,7 +40,7 @@ impl PollWatcher {
|
||||
pub fn with_delay(tx: Sender<Event>, delay: u32) -> Result<PollWatcher> {
|
||||
let mut p = PollWatcher {
|
||||
tx: tx,
|
||||
watches: Arc::new(RwLock::new(HashMap::new())),
|
||||
watches: Arc::new(Mutex::new(HashMap::new())),
|
||||
open: Arc::new(RwLock::new(true)),
|
||||
};
|
||||
p.run(delay);
|
||||
@@ -46,122 +56,77 @@ impl PollWatcher {
|
||||
// TODO: handle chmod events
|
||||
// TODO: handle renames
|
||||
// TODO: DRY it up
|
||||
let mut mtimes: HashMap<PathBuf, (u64, f64)> = HashMap::new();
|
||||
|
||||
let mut current_time = time::precise_time_s();
|
||||
|
||||
for (watch, is_recursive) in watches.read().unwrap().iter() {
|
||||
match fs::metadata(watch) {
|
||||
Err(e) => {
|
||||
let _ = tx.send(Event {
|
||||
path: Some(watch.clone()),
|
||||
op: Err(Error::Io(e)),
|
||||
});
|
||||
continue;
|
||||
}
|
||||
Ok(metadata) => {
|
||||
if !metadata.is_dir() {
|
||||
let modified = FileTime::from_last_modification_time(&metadata).seconds();
|
||||
mtimes.insert(watch.clone(), (modified, current_time));
|
||||
} else {
|
||||
let depth = if *is_recursive {
|
||||
usize::max_value()
|
||||
} else {
|
||||
1
|
||||
};
|
||||
for entry in WalkDir::new(watch)
|
||||
.follow_links(true)
|
||||
.max_depth(depth)
|
||||
.into_iter()
|
||||
.filter_map(|e| e.ok()) {
|
||||
let path = entry.path();
|
||||
|
||||
match entry.metadata() {
|
||||
Err(e) => {
|
||||
let _ = tx.send(Event {
|
||||
path: Some(path.to_path_buf()),
|
||||
op: Err(Error::Io(e.into())),
|
||||
});
|
||||
}
|
||||
Ok(m) => {
|
||||
let modified = FileTime::from_last_modification_time(&m).seconds();
|
||||
mtimes.insert(path.to_path_buf(), (modified, current_time));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
loop {
|
||||
if !(*open.read().unwrap()) {
|
||||
break;
|
||||
}
|
||||
|
||||
current_time = time::precise_time_s();
|
||||
if let Ok(mut watches) = watches.lock() {
|
||||
let current_time = time::precise_time_s();
|
||||
|
||||
for (watch, is_recursive) in watches.read().unwrap().iter() {
|
||||
match fs::metadata(watch) {
|
||||
Err(e) => {
|
||||
let _ = tx.send(Event {
|
||||
path: Some(watch.clone()),
|
||||
op: Err(Error::Io(e)),
|
||||
});
|
||||
continue;
|
||||
}
|
||||
Ok(metadata) => {
|
||||
if !metadata.is_dir() {
|
||||
let modified = FileTime::from_last_modification_time(&metadata).seconds();
|
||||
match mtimes.insert(watch.clone(), (modified, current_time)) {
|
||||
None => {
|
||||
unreachable!();
|
||||
}
|
||||
Some((old_modified, _)) => {
|
||||
if modified > old_modified {
|
||||
let _ = tx.send(Event {
|
||||
path: Some(watch.clone()),
|
||||
op: Ok(op::WRITE),
|
||||
});
|
||||
for (watch, &mut WatchData{is_recursive, ref mut paths}) in watches.iter_mut() {
|
||||
match fs::metadata(watch) {
|
||||
Err(e) => {
|
||||
let _ = tx.send(Event {
|
||||
path: Some(watch.clone()),
|
||||
op: Err(Error::Io(e)),
|
||||
});
|
||||
continue;
|
||||
}
|
||||
Ok(metadata) => {
|
||||
if !metadata.is_dir() {
|
||||
let mtime = FileTime::from_last_modification_time(&metadata).seconds();
|
||||
match paths.insert(watch.clone(), PathData{mtime: mtime, last_check: current_time}) {
|
||||
None => {
|
||||
unreachable!();
|
||||
}
|
||||
Some(PathData{mtime: old_mtime, ..}) => {
|
||||
if mtime > old_mtime {
|
||||
let _ = tx.send(Event {
|
||||
path: Some(watch.clone()),
|
||||
op: Ok(op::WRITE),
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
let depth = if *is_recursive {
|
||||
usize::max_value()
|
||||
} else {
|
||||
1
|
||||
};
|
||||
for entry in WalkDir::new(watch)
|
||||
.follow_links(true)
|
||||
.max_depth(depth)
|
||||
.into_iter()
|
||||
.filter_map(|e| e.ok()) {
|
||||
let path = entry.path();
|
||||
let depth = if is_recursive {
|
||||
usize::max_value()
|
||||
} else {
|
||||
1
|
||||
};
|
||||
for entry in WalkDir::new(watch)
|
||||
.follow_links(true)
|
||||
.max_depth(depth)
|
||||
.into_iter()
|
||||
.filter_map(|e| e.ok()) {
|
||||
let path = entry.path();
|
||||
|
||||
match entry.metadata() {
|
||||
Err(e) => {
|
||||
let _ = tx.send(Event {
|
||||
path: Some(path.to_path_buf()),
|
||||
op: Err(Error::Io(e.into())),
|
||||
});
|
||||
}
|
||||
Ok(m) => {
|
||||
let modified = FileTime::from_last_modification_time(&m).seconds();
|
||||
match mtimes.insert(path.to_path_buf(), (modified, current_time)) {
|
||||
None => {
|
||||
let _ = tx.send(Event {
|
||||
path: Some(path.to_path_buf()),
|
||||
op: Ok(op::CREATE),
|
||||
});
|
||||
}
|
||||
Some((old_modified, _)) => {
|
||||
if modified > old_modified {
|
||||
match entry.metadata() {
|
||||
Err(e) => {
|
||||
let _ = tx.send(Event {
|
||||
path: Some(path.to_path_buf()),
|
||||
op: Err(Error::Io(e.into())),
|
||||
});
|
||||
}
|
||||
Ok(m) => {
|
||||
let mtime = FileTime::from_last_modification_time(&m).seconds();
|
||||
match paths.insert(path.to_path_buf(), PathData{mtime: mtime, last_check: current_time}) {
|
||||
None => {
|
||||
let _ = tx.send(Event {
|
||||
path: Some(path.to_path_buf()),
|
||||
op: Ok(op::WRITE),
|
||||
op: Ok(op::CREATE),
|
||||
});
|
||||
}
|
||||
Some(PathData{mtime: old_mtime, ..}) => {
|
||||
if mtime > old_mtime {
|
||||
let _ = tx.send(Event {
|
||||
path: Some(path.to_path_buf()),
|
||||
op: Ok(op::WRITE),
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -170,33 +135,26 @@ impl PollWatcher {
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let mut removed: Vec<PathBuf> = Vec::new();
|
||||
|
||||
'paths: for (ref path, &(_, last_checked)) in &mtimes {
|
||||
for (watch, _) in watches.read().unwrap().iter() {
|
||||
if path.starts_with(watch) {
|
||||
if last_checked < current_time {
|
||||
for (_, &mut WatchData{ref mut paths, ..}) in watches.iter_mut() {
|
||||
let mut removed = Vec::new();
|
||||
for (path, &PathData{last_check, ..}) in paths.iter() {
|
||||
if last_check < current_time {
|
||||
let _ = tx.send(Event {
|
||||
path: Some(path.to_path_buf()),
|
||||
path: Some(path.clone()),
|
||||
op: Ok(op::REMOVE),
|
||||
});
|
||||
removed.push(path.to_path_buf());
|
||||
removed.push(path.clone());
|
||||
}
|
||||
continue 'paths;
|
||||
}
|
||||
for path in removed {
|
||||
(*paths).remove(&path);
|
||||
}
|
||||
}
|
||||
// not found in watches
|
||||
removed.push(path.to_path_buf());
|
||||
}
|
||||
|
||||
for path in removed {
|
||||
mtimes.remove(&path);
|
||||
}
|
||||
|
||||
if delay != 0 {
|
||||
thread::sleep(Duration::from_millis(delay as u64));
|
||||
if delay != 0 {
|
||||
thread::sleep(Duration::from_millis(delay as u64));
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
@@ -209,12 +167,61 @@ impl Watcher for PollWatcher {
|
||||
}
|
||||
|
||||
fn watch<P: AsRef<Path>>(&mut self, path: P, recursive_mode: RecursiveMode) -> Result<()> {
|
||||
(*self.watches).write().unwrap().insert(path.as_ref().to_path_buf(), recursive_mode.is_recursive());
|
||||
if let Ok(mut watches) = self.watches.lock() {
|
||||
let current_time = time::precise_time_s();
|
||||
|
||||
let watch = path.as_ref().to_owned();
|
||||
|
||||
match fs::metadata(path) {
|
||||
Err(e) => {
|
||||
let _ = self.tx.send(Event {
|
||||
path: Some(watch.clone()),
|
||||
op: Err(Error::Io(e)),
|
||||
});
|
||||
}
|
||||
Ok(metadata) => {
|
||||
if !metadata.is_dir() {
|
||||
let mut paths = HashMap::new();
|
||||
let mtime = FileTime::from_last_modification_time(&metadata).seconds();
|
||||
paths.insert(watch.clone(), PathData{mtime: mtime, last_check: current_time});
|
||||
watches.insert(watch, WatchData{is_recursive: recursive_mode.is_recursive(), paths: paths});
|
||||
} else {
|
||||
let mut paths = HashMap::new();
|
||||
let depth = if recursive_mode.is_recursive() {
|
||||
usize::max_value()
|
||||
} else {
|
||||
1
|
||||
};
|
||||
for entry in WalkDir::new(watch.clone())
|
||||
.follow_links(true)
|
||||
.max_depth(depth)
|
||||
.into_iter()
|
||||
.filter_map(|e| e.ok()) {
|
||||
let path = entry.path();
|
||||
|
||||
match entry.metadata() {
|
||||
Err(e) => {
|
||||
let _ = self.tx.send(Event {
|
||||
path: Some(path.to_path_buf()),
|
||||
op: Err(Error::Io(e.into())),
|
||||
});
|
||||
}
|
||||
Ok(m) => {
|
||||
let mtime = FileTime::from_last_modification_time(&m).seconds();
|
||||
paths.insert(path.to_path_buf(), PathData{mtime: mtime, last_check: current_time});
|
||||
}
|
||||
}
|
||||
}
|
||||
watches.insert(watch, WatchData{is_recursive: recursive_mode.is_recursive(), paths: paths});
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn unwatch<P: AsRef<Path>>(&mut self, path: P) -> Result<()> {
|
||||
if (*self.watches).write().unwrap().remove(path.as_ref()).is_some() {
|
||||
if (*self.watches).lock().unwrap().remove(path.as_ref()).is_some() {
|
||||
Ok(())
|
||||
} else {
|
||||
Err(Error::WatchNotFound)
|
||||
|
||||
Reference in New Issue
Block a user