#![allow(non_upper_case_globals, dead_code)]
use crate::event::*;
use crate::{unbounded, Config, Error, EventHandler, RecursiveMode, Result, Sender, Watcher};
use fsevent_sys as fs;
use fsevent_sys::core_foundation as cf;
use std::collections::HashMap;
use std::ffi::CStr;
use std::fmt;
use std::os::raw;
use std::path::{Path, PathBuf};
use std::ptr;
use std::sync::{Arc, Mutex};
use std::thread;
bitflags::bitflags! {
#[repr(C)]
struct StreamFlags: u32 {
const NONE = fs::kFSEventStreamEventFlagNone;
const MUST_SCAN_SUBDIRS = fs::kFSEventStreamEventFlagMustScanSubDirs;
const USER_DROPPED = fs::kFSEventStreamEventFlagUserDropped;
const KERNEL_DROPPED = fs::kFSEventStreamEventFlagKernelDropped;
const IDS_WRAPPED = fs::kFSEventStreamEventFlagEventIdsWrapped;
const HISTORY_DONE = fs::kFSEventStreamEventFlagHistoryDone;
const ROOT_CHANGED = fs::kFSEventStreamEventFlagRootChanged;
const MOUNT = fs::kFSEventStreamEventFlagMount;
const UNMOUNT = fs::kFSEventStreamEventFlagUnmount;
const ITEM_CREATED = fs::kFSEventStreamEventFlagItemCreated;
const ITEM_REMOVED = fs::kFSEventStreamEventFlagItemRemoved;
const INODE_META_MOD = fs::kFSEventStreamEventFlagItemInodeMetaMod;
const ITEM_RENAMED = fs::kFSEventStreamEventFlagItemRenamed;
const ITEM_MODIFIED = fs::kFSEventStreamEventFlagItemModified;
const FINDER_INFO_MOD = fs::kFSEventStreamEventFlagItemFinderInfoMod;
const ITEM_CHANGE_OWNER = fs::kFSEventStreamEventFlagItemChangeOwner;
const ITEM_XATTR_MOD = fs::kFSEventStreamEventFlagItemXattrMod;
const IS_FILE = fs::kFSEventStreamEventFlagItemIsFile;
const IS_DIR = fs::kFSEventStreamEventFlagItemIsDir;
const IS_SYMLINK = fs::kFSEventStreamEventFlagItemIsSymlink;
const OWN_EVENT = fs::kFSEventStreamEventFlagOwnEvent;
const IS_HARDLINK = fs::kFSEventStreamEventFlagItemIsHardlink;
const IS_LAST_HARDLINK = fs::kFSEventStreamEventFlagItemIsLastHardlink;
const ITEM_CLONED = fs::kFSEventStreamEventFlagItemCloned;
}
}
pub struct FsEventWatcher {
paths: cf::CFMutableArrayRef,
since_when: fs::FSEventStreamEventId,
latency: cf::CFTimeInterval,
flags: fs::FSEventStreamCreateFlags,
event_handler: Arc<Mutex<dyn EventHandler>>,
runloop: Option<(cf::CFRunLoopRef, thread::JoinHandle<()>)>,
recursive_info: HashMap<PathBuf, bool>,
}
impl fmt::Debug for FsEventWatcher {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.debug_struct("FsEventWatcher")
.field("paths", &self.paths)
.field("since_when", &self.since_when)
.field("latency", &self.latency)
.field("flags", &self.flags)
.field("event_handler", &Arc::as_ptr(&self.event_handler))
.field("runloop", &self.runloop)
.field("recursive_info", &self.recursive_info)
.finish()
}
}
unsafe impl Send for FsEventWatcher {}
unsafe impl Sync for FsEventWatcher {}
fn translate_flags(flags: StreamFlags, precise: bool) -> Vec<Event> {
let mut evs = Vec::new();
if flags.contains(StreamFlags::HISTORY_DONE) {
return evs;
}
if flags.contains(StreamFlags::MUST_SCAN_SUBDIRS) {
let e = Event::new(EventKind::Other).set_flag(Flag::Rescan);
evs.push(if flags.contains(StreamFlags::USER_DROPPED) {
e.set_info("rescan: user dropped")
} else if flags.contains(StreamFlags::KERNEL_DROPPED) {
e.set_info("rescan: kernel dropped")
} else {
e
});
}
if !precise {
evs.push(Event::new(EventKind::Any));
return evs;
}
if flags.contains(StreamFlags::ROOT_CHANGED) {
evs.push(
Event::new(EventKind::Modify(ModifyKind::Name(RenameMode::From)))
.set_info("root changed"),
);
}
if flags.contains(StreamFlags::MOUNT) {
evs.push(Event::new(EventKind::Create(CreateKind::Other)).set_info("mount"));
}
if flags.contains(StreamFlags::UNMOUNT) {
evs.push(Event::new(EventKind::Remove(RemoveKind::Other)).set_info("mount"));
}
if flags.contains(StreamFlags::ITEM_CREATED) {
evs.push(if flags.contains(StreamFlags::IS_DIR) {
Event::new(EventKind::Create(CreateKind::Folder))
} else if flags.contains(StreamFlags::IS_FILE) {
Event::new(EventKind::Create(CreateKind::File))
} else {
let e = Event::new(EventKind::Create(CreateKind::Other));
if flags.contains(StreamFlags::IS_SYMLINK) {
e.set_info("is: symlink")
} else if flags.contains(StreamFlags::IS_HARDLINK) {
e.set_info("is: hardlink")
} else if flags.contains(StreamFlags::ITEM_CLONED) {
e.set_info("is: clone")
} else {
Event::new(EventKind::Create(CreateKind::Any))
}
});
}
if flags.contains(StreamFlags::ITEM_REMOVED) {
evs.push(if flags.contains(StreamFlags::IS_DIR) {
Event::new(EventKind::Remove(RemoveKind::Folder))
} else if flags.contains(StreamFlags::IS_FILE) {
Event::new(EventKind::Remove(RemoveKind::File))
} else {
let e = Event::new(EventKind::Remove(RemoveKind::Other));
if flags.contains(StreamFlags::IS_SYMLINK) {
e.set_info("is: symlink")
} else if flags.contains(StreamFlags::IS_HARDLINK) {
e.set_info("is: hardlink")
} else if flags.contains(StreamFlags::ITEM_CLONED) {
e.set_info("is: clone")
} else {
Event::new(EventKind::Remove(RemoveKind::Any))
}
});
}
if flags.contains(StreamFlags::ITEM_RENAMED) {
evs.push(Event::new(EventKind::Modify(ModifyKind::Name(
RenameMode::Any,
))));
}
if flags.contains(StreamFlags::INODE_META_MOD) {
evs.push(Event::new(EventKind::Modify(ModifyKind::Metadata(
MetadataKind::Any,
))));
}
if flags.contains(StreamFlags::FINDER_INFO_MOD) {
evs.push(
Event::new(EventKind::Modify(ModifyKind::Metadata(MetadataKind::Other)))
.set_info("meta: finder info"),
);
}
if flags.contains(StreamFlags::ITEM_CHANGE_OWNER) {
evs.push(Event::new(EventKind::Modify(ModifyKind::Metadata(
MetadataKind::Ownership,
))));
}
if flags.contains(StreamFlags::ITEM_XATTR_MOD) {
evs.push(Event::new(EventKind::Modify(ModifyKind::Metadata(
MetadataKind::Extended,
))));
}
if flags.contains(StreamFlags::ITEM_MODIFIED) {
evs.push(Event::new(EventKind::Modify(ModifyKind::Data(
DataChange::Content,
))));
}
if flags.contains(StreamFlags::OWN_EVENT) {
for ev in &mut evs {
*ev = std::mem::take(ev).set_process_id(std::process::id());
}
}
evs
}
struct StreamContextInfo {
event_handler: Arc<Mutex<dyn EventHandler>>,
recursive_info: HashMap<PathBuf, bool>,
}
extern "C" fn release_context(info: *const libc::c_void) {
unsafe {
drop(Box::from_raw(
info as *const StreamContextInfo as *mut StreamContextInfo,
));
}
}
extern "C" {
fn CFRunLoopIsWaiting(runloop: cf::CFRunLoopRef) -> cf::Boolean;
}
impl FsEventWatcher {
fn from_event_handler(event_handler: Arc<Mutex<dyn EventHandler>>) -> Result<Self> {
Ok(FsEventWatcher {
paths: unsafe {
cf::CFArrayCreateMutable(cf::kCFAllocatorDefault, 0, &cf::kCFTypeArrayCallBacks)
},
since_when: fs::kFSEventStreamEventIdSinceNow,
latency: 0.0,
flags: fs::kFSEventStreamCreateFlagFileEvents | fs::kFSEventStreamCreateFlagNoDefer,
event_handler,
runloop: None,
recursive_info: HashMap::new(),
})
}
fn watch_inner(&mut self, path: &Path, recursive_mode: RecursiveMode) -> Result<()> {
self.stop();
let result = self.append_path(path, recursive_mode);
let _ = self.run();
result
}
fn unwatch_inner(&mut self, path: &Path) -> Result<()> {
self.stop();
let result = self.remove_path(path);
let _ = self.run();
result
}
#[inline]
fn is_running(&self) -> bool {
self.runloop.is_some()
}
fn stop(&mut self) {
if !self.is_running() {
return;
}
if let Some((runloop, thread_handle)) = self.runloop.take() {
unsafe {
let runloop = runloop as *mut raw::c_void;
while CFRunLoopIsWaiting(runloop) == 0 {
thread::yield_now();
}
cf::CFRunLoopStop(runloop);
}
thread_handle.join().expect("thread to shut down");
}
}
fn remove_path(&mut self, path: &Path) -> Result<()> {
let str_path = path.to_str().unwrap();
unsafe {
let mut err: cf::CFErrorRef = ptr::null_mut();
let cf_path = cf::str_path_to_cfstring_ref(str_path, &mut err);
if cf_path.is_null() {
cf::CFRelease(err as cf::CFRef);
return Err(Error::watch_not_found().add_path(path.into()));
}
let mut to_remove = Vec::new();
for idx in 0..cf::CFArrayGetCount(self.paths) {
let item = cf::CFArrayGetValueAtIndex(self.paths, idx);
if cf::CFStringCompare(item, cf_path, cf::kCFCompareCaseInsensitive)
== cf::kCFCompareEqualTo
{
to_remove.push(idx);
}
}
cf::CFRelease(cf_path);
for idx in to_remove.iter().rev() {
cf::CFArrayRemoveValueAtIndex(self.paths, *idx);
}
}
let p = if let Ok(canonicalized_path) = path.canonicalize() {
canonicalized_path
} else {
path.to_owned()
};
match self.recursive_info.remove(&p) {
Some(_) => Ok(()),
None => Err(Error::watch_not_found()),
}
}
fn append_path(&mut self, path: &Path, recursive_mode: RecursiveMode) -> Result<()> {
if !path.exists() {
return Err(Error::path_not_found().add_path(path.into()));
}
let canonical_path = path.to_path_buf().canonicalize()?;
let str_path = path.to_str().unwrap();
unsafe {
let mut err: cf::CFErrorRef = ptr::null_mut();
let cf_path = cf::str_path_to_cfstring_ref(str_path, &mut err);
if cf_path.is_null() {
cf::CFRelease(err as cf::CFRef);
return Err(Error::path_not_found().add_path(path.into()));
}
cf::CFArrayAppendValue(self.paths, cf_path);
cf::CFRelease(cf_path);
}
self.recursive_info
.insert(canonical_path, recursive_mode.is_recursive());
Ok(())
}
fn run(&mut self) -> Result<()> {
if unsafe { cf::CFArrayGetCount(self.paths) } == 0 {
return Err(Error::path_not_found());
}
let context = Box::into_raw(Box::new(StreamContextInfo {
event_handler: self.event_handler.clone(),
recursive_info: self.recursive_info.clone(),
}));
let stream_context = fs::FSEventStreamContext {
version: 0,
info: context as *mut libc::c_void,
retain: None,
release: Some(release_context),
copy_description: None,
};
let stream = unsafe {
fs::FSEventStreamCreate(
cf::kCFAllocatorDefault,
callback,
&stream_context,
self.paths,
self.since_when,
self.latency,
self.flags,
)
};
struct CFSendWrapper(cf::CFRef);
unsafe impl Send for CFSendWrapper {}
let stream = CFSendWrapper(stream);
let (rl_tx, rl_rx) = unbounded();
let thread_handle = thread::Builder::new()
.name("notify-rs fsevents loop".to_string())
.spawn(move || {
let _ = &stream;
let stream = stream.0;
unsafe {
let cur_runloop = cf::CFRunLoopGetCurrent();
fs::FSEventStreamScheduleWithRunLoop(
stream,
cur_runloop,
cf::kCFRunLoopDefaultMode,
);
fs::FSEventStreamStart(stream);
rl_tx
.send(CFSendWrapper(cur_runloop))
.expect("Unable to send runloop to watcher");
cf::CFRunLoopRun();
fs::FSEventStreamStop(stream);
fs::FSEventStreamInvalidate(stream);
fs::FSEventStreamRelease(stream);
}
})?;
self.runloop = Some((rl_rx.recv().unwrap().0, thread_handle));
Ok(())
}
fn configure_raw_mode(&mut self, _config: Config, tx: Sender<Result<bool>>) {
tx.send(Ok(false))
.expect("configuration channel disconnect");
}
}
extern "C" fn callback(
stream_ref: fs::FSEventStreamRef,
info: *mut libc::c_void,
num_events: libc::size_t, event_paths: *mut libc::c_void, event_flags: *const fs::FSEventStreamEventFlags, event_ids: *const fs::FSEventStreamEventId, ) {
unsafe {
callback_impl(
stream_ref,
info,
num_events,
event_paths,
event_flags,
event_ids,
)
}
}
unsafe fn callback_impl(
_stream_ref: fs::FSEventStreamRef,
info: *mut libc::c_void,
num_events: libc::size_t, event_paths: *mut libc::c_void, event_flags: *const fs::FSEventStreamEventFlags, _event_ids: *const fs::FSEventStreamEventId, ) {
let event_paths = event_paths as *const *const libc::c_char;
let info = info as *const StreamContextInfo;
let event_handler = &(*info).event_handler;
for p in 0..num_events {
let path = CStr::from_ptr(*event_paths.add(p))
.to_str()
.expect("Invalid UTF8 string.");
let path = PathBuf::from(path);
let flag = *event_flags.add(p);
let flag = StreamFlags::from_bits(flag).unwrap_or_else(|| {
panic!("Unable to decode StreamFlags: {}", flag);
});
let mut handle_event = false;
for (p, r) in &(*info).recursive_info {
if path.starts_with(p) {
if *r || &path == p {
handle_event = true;
break;
} else if let Some(parent_path) = path.parent() {
if parent_path == p {
handle_event = true;
break;
}
}
}
}
if !handle_event {
continue;
}
for ev in translate_flags(flag, true).into_iter() {
let ev = ev.add_path(path.clone());
let mut event_handler = event_handler.lock().expect("lock not to be poisoned");
event_handler.handle_event(Ok(ev));
}
}
}
impl Watcher for FsEventWatcher {
fn new<F: EventHandler>(event_handler: F, _config: Config) -> Result<Self> {
Self::from_event_handler(Arc::new(Mutex::new(event_handler)))
}
fn watch(&mut self, path: &Path, recursive_mode: RecursiveMode) -> Result<()> {
self.watch_inner(path, recursive_mode)
}
fn unwatch(&mut self, path: &Path) -> Result<()> {
self.unwatch_inner(path)
}
fn configure(&mut self, config: Config) -> Result<bool> {
let (tx, rx) = unbounded();
self.configure_raw_mode(config, tx);
rx.recv()?
}
fn kind() -> crate::WatcherKind {
crate::WatcherKind::Fsevent
}
}
impl Drop for FsEventWatcher {
fn drop(&mut self) {
self.stop();
unsafe {
cf::CFRelease(self.paths);
}
}
}
#[test]
fn test_fsevent_watcher_drop() {
use super::*;
use std::time::Duration;
let dir = tempfile::tempdir().unwrap();
let (tx, rx) = std::sync::mpsc::channel();
{
let mut watcher = FsEventWatcher::new(tx, Default::default()).unwrap();
watcher.watch(dir.path(), RecursiveMode::Recursive).unwrap();
thread::sleep(Duration::from_millis(2000));
println!("is running -> {}", watcher.is_running());
thread::sleep(Duration::from_millis(1000));
watcher.unwatch(dir.path()).unwrap();
println!("is running -> {}", watcher.is_running());
}
thread::sleep(Duration::from_millis(1000));
for res in rx {
let e = res.unwrap();
println!("debug => {:?} {:?}", e.kind, e.paths);
}
println!("in test: {} works", file!());
}
#[test]
fn test_steam_context_info_send_and_sync() {
fn check_send<T: Send + Sync>() {}
check_send::<StreamContextInfo>();
}