1#![cfg(target_os = "macos")]
2
3use bitflags::bitflags;
4use fsevent_sys::{self as fs, core_foundation as cf};
5use parking_lot::Mutex;
6use std::{
7 convert::AsRef,
8 ffi::{c_void, CStr, OsStr},
9 os::unix::ffi::OsStrExt,
10 path::{Path, PathBuf},
11 slice,
12 sync::Arc,
13 time::Duration,
14};
15
16#[derive(Clone, Debug)]
17pub struct Event {
18 pub event_id: u64,
19 pub flags: StreamFlags,
20 pub path: PathBuf,
21}
22
23pub struct EventStream {
24 stream: fs::FSEventStreamRef,
25 state: Arc<Mutex<Lifecycle>>,
26 callback: Box<Option<RunCallback>>,
27}
28
29type RunCallback = Box<dyn FnMut(Vec<Event>) -> bool>;
30
31enum Lifecycle {
32 New,
33 Running(cf::CFRunLoopRef),
34 Stopped,
35}
36
37pub struct Handle(Arc<Mutex<Lifecycle>>);
38
39unsafe impl Send for EventStream {}
40unsafe impl Send for Lifecycle {}
41
42impl EventStream {
43 pub fn new(paths: &[&Path], latency: Duration) -> (Self, Handle) {
44 unsafe {
45 let callback = Box::new(None);
46 let stream_context = fs::FSEventStreamContext {
47 version: 0,
48 info: callback.as_ref() as *const _ as *mut c_void,
49 retain: None,
50 release: None,
51 copy_description: None,
52 };
53
54 let cf_paths =
55 cf::CFArrayCreateMutable(cf::kCFAllocatorDefault, 0, &cf::kCFTypeArrayCallBacks);
56 assert!(!cf_paths.is_null());
57
58 for path in paths {
59 let path_bytes = path.as_os_str().as_bytes();
60 let cf_url = cf::CFURLCreateFromFileSystemRepresentation(
61 cf::kCFAllocatorDefault,
62 path_bytes.as_ptr() as *const i8,
63 path_bytes.len() as cf::CFIndex,
64 false,
65 );
66 let cf_path = cf::CFURLCopyFileSystemPath(cf_url, cf::kCFURLPOSIXPathStyle);
67 cf::CFArrayAppendValue(cf_paths, cf_path);
68 cf::CFRelease(cf_path);
69 cf::CFRelease(cf_url);
70 }
71
72 let stream = fs::FSEventStreamCreate(
73 cf::kCFAllocatorDefault,
74 Self::trampoline,
75 &stream_context,
76 cf_paths,
77 fs::kFSEventStreamEventIdSinceNow,
78 latency.as_secs_f64(),
79 fs::kFSEventStreamCreateFlagFileEvents
80 | fs::kFSEventStreamCreateFlagNoDefer
81 | fs::kFSEventStreamCreateFlagWatchRoot,
82 );
83 cf::CFRelease(cf_paths);
84
85 let state = Arc::new(Mutex::new(Lifecycle::New));
86
87 (
88 EventStream {
89 stream,
90 state: state.clone(),
91 callback,
92 },
93 Handle(state),
94 )
95 }
96 }
97
98 pub fn run<F>(mut self, f: F)
99 where
100 F: FnMut(Vec<Event>) -> bool + 'static,
101 {
102 *self.callback = Some(Box::new(f));
103 unsafe {
104 let run_loop = cf::CFRunLoopGetCurrent();
105 {
106 let mut state = self.state.lock();
107 match *state {
108 Lifecycle::New => *state = Lifecycle::Running(run_loop),
109 Lifecycle::Running(_) => unreachable!(),
110 Lifecycle::Stopped => return,
111 }
112 }
113 fs::FSEventStreamScheduleWithRunLoop(self.stream, run_loop, cf::kCFRunLoopDefaultMode);
114
115 fs::FSEventStreamStart(self.stream);
116 cf::CFRunLoopRun();
117
118 fs::FSEventStreamFlushSync(self.stream);
119 fs::FSEventStreamStop(self.stream);
120 fs::FSEventStreamRelease(self.stream);
121 }
122 }
123
124 extern "C" fn trampoline(
125 stream_ref: fs::FSEventStreamRef,
126 info: *mut ::std::os::raw::c_void,
127 num: usize, // size_t numEvents
128 event_paths: *mut ::std::os::raw::c_void, // void *eventPaths
129 event_flags: *const ::std::os::raw::c_void, // const FSEventStreamEventFlags eventFlags[]
130 event_ids: *const ::std::os::raw::c_void, // const FSEventStreamEventId eventIds[]
131 ) {
132 unsafe {
133 let event_paths = event_paths as *const *const ::std::os::raw::c_char;
134 let e_ptr = event_flags as *mut u32;
135 let i_ptr = event_ids as *mut u64;
136 let callback = (info as *mut Option<RunCallback>)
137 .as_mut()
138 .unwrap()
139 .as_mut()
140 .unwrap();
141
142 let paths = slice::from_raw_parts(event_paths, num);
143 let flags = slice::from_raw_parts_mut(e_ptr, num);
144 let ids = slice::from_raw_parts_mut(i_ptr, num);
145
146 let mut events = Vec::with_capacity(num);
147 for p in 0..num {
148 let path_c_str = CStr::from_ptr(paths[p]);
149 let path = PathBuf::from(OsStr::from_bytes(path_c_str.to_bytes()));
150 if let Some(flag) = StreamFlags::from_bits(flags[p]) {
151 events.push(Event {
152 event_id: ids[p],
153 flags: flag,
154 path,
155 });
156 } else {
157 debug_assert!(false, "unknown flag set for fs event: {}", flags[p]);
158 }
159 }
160
161 if !callback(events) {
162 fs::FSEventStreamStop(stream_ref);
163 cf::CFRunLoopStop(cf::CFRunLoopGetCurrent());
164 }
165 }
166 }
167}
168
169impl Drop for Handle {
170 fn drop(&mut self) {
171 let mut state = self.0.lock();
172 if let Lifecycle::Running(run_loop) = *state {
173 unsafe {
174 cf::CFRunLoopStop(run_loop);
175 }
176 }
177 *state = Lifecycle::Stopped;
178 }
179}
180
181// Synchronize with
182// /System/Library/Frameworks/CoreServices.framework/Versions/A/Frameworks/FSEvents.framework/Versions/A/Headers/FSEvents.h
183bitflags! {
184 #[repr(C)]
185 pub struct StreamFlags: u32 {
186 const NONE = 0x00000000;
187 const MUST_SCAN_SUBDIRS = 0x00000001;
188 const USER_DROPPED = 0x00000002;
189 const KERNEL_DROPPED = 0x00000004;
190 const IDS_WRAPPED = 0x00000008;
191 const HISTORY_DONE = 0x00000010;
192 const ROOT_CHANGED = 0x00000020;
193 const MOUNT = 0x00000040;
194 const UNMOUNT = 0x00000080;
195 const ITEM_CREATED = 0x00000100;
196 const ITEM_REMOVED = 0x00000200;
197 const INODE_META_MOD = 0x00000400;
198 const ITEM_RENAMED = 0x00000800;
199 const ITEM_MODIFIED = 0x00001000;
200 const FINDER_INFO_MOD = 0x00002000;
201 const ITEM_CHANGE_OWNER = 0x00004000;
202 const ITEM_XATTR_MOD = 0x00008000;
203 const IS_FILE = 0x00010000;
204 const IS_DIR = 0x00020000;
205 const IS_SYMLINK = 0x00040000;
206 const OWN_EVENT = 0x00080000;
207 const IS_HARDLINK = 0x00100000;
208 const IS_LAST_HARDLINK = 0x00200000;
209 const ITEM_CLONED = 0x400000;
210 }
211}
212
213impl std::fmt::Display for StreamFlags {
214 fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
215 if self.contains(StreamFlags::MUST_SCAN_SUBDIRS) {
216 let _d = write!(f, "MUST_SCAN_SUBDIRS ");
217 }
218 if self.contains(StreamFlags::USER_DROPPED) {
219 let _d = write!(f, "USER_DROPPED ");
220 }
221 if self.contains(StreamFlags::KERNEL_DROPPED) {
222 let _d = write!(f, "KERNEL_DROPPED ");
223 }
224 if self.contains(StreamFlags::IDS_WRAPPED) {
225 let _d = write!(f, "IDS_WRAPPED ");
226 }
227 if self.contains(StreamFlags::HISTORY_DONE) {
228 let _d = write!(f, "HISTORY_DONE ");
229 }
230 if self.contains(StreamFlags::ROOT_CHANGED) {
231 let _d = write!(f, "ROOT_CHANGED ");
232 }
233 if self.contains(StreamFlags::MOUNT) {
234 let _d = write!(f, "MOUNT ");
235 }
236 if self.contains(StreamFlags::UNMOUNT) {
237 let _d = write!(f, "UNMOUNT ");
238 }
239 if self.contains(StreamFlags::ITEM_CREATED) {
240 let _d = write!(f, "ITEM_CREATED ");
241 }
242 if self.contains(StreamFlags::ITEM_REMOVED) {
243 let _d = write!(f, "ITEM_REMOVED ");
244 }
245 if self.contains(StreamFlags::INODE_META_MOD) {
246 let _d = write!(f, "INODE_META_MOD ");
247 }
248 if self.contains(StreamFlags::ITEM_RENAMED) {
249 let _d = write!(f, "ITEM_RENAMED ");
250 }
251 if self.contains(StreamFlags::ITEM_MODIFIED) {
252 let _d = write!(f, "ITEM_MODIFIED ");
253 }
254 if self.contains(StreamFlags::FINDER_INFO_MOD) {
255 let _d = write!(f, "FINDER_INFO_MOD ");
256 }
257 if self.contains(StreamFlags::ITEM_CHANGE_OWNER) {
258 let _d = write!(f, "ITEM_CHANGE_OWNER ");
259 }
260 if self.contains(StreamFlags::ITEM_XATTR_MOD) {
261 let _d = write!(f, "ITEM_XATTR_MOD ");
262 }
263 if self.contains(StreamFlags::IS_FILE) {
264 let _d = write!(f, "IS_FILE ");
265 }
266 if self.contains(StreamFlags::IS_DIR) {
267 let _d = write!(f, "IS_DIR ");
268 }
269 if self.contains(StreamFlags::IS_SYMLINK) {
270 let _d = write!(f, "IS_SYMLINK ");
271 }
272 if self.contains(StreamFlags::OWN_EVENT) {
273 let _d = write!(f, "OWN_EVENT ");
274 }
275 if self.contains(StreamFlags::IS_LAST_HARDLINK) {
276 let _d = write!(f, "IS_LAST_HARDLINK ");
277 }
278 if self.contains(StreamFlags::IS_HARDLINK) {
279 let _d = write!(f, "IS_HARDLINK ");
280 }
281 if self.contains(StreamFlags::ITEM_CLONED) {
282 let _d = write!(f, "ITEM_CLONED ");
283 }
284 write!(f, "")
285 }
286}
287
288#[test]
289fn test_event_stream() {
290 use std::{fs, sync::mpsc, time::Duration};
291 use tempdir::TempDir;
292
293 let dir = TempDir::new("test_observe").unwrap();
294 let path = dir.path().canonicalize().unwrap();
295 fs::write(path.join("a"), "a contents").unwrap();
296
297 let (tx, rx) = mpsc::channel();
298 let (stream, handle) = EventStream::new(&[&path], Duration::from_millis(50));
299 std::thread::spawn(move || stream.run(move |events| tx.send(events.to_vec()).is_ok()));
300
301 fs::write(path.join("b"), "b contents").unwrap();
302 let events = rx.recv_timeout(Duration::from_millis(500)).unwrap();
303 let event = events.last().unwrap();
304 assert_eq!(event.path, path.join("b"));
305 assert!(event.flags.contains(StreamFlags::ITEM_CREATED));
306
307 fs::remove_file(path.join("a")).unwrap();
308 let events = rx.recv_timeout(Duration::from_millis(500)).unwrap();
309 let event = events.last().unwrap();
310 assert_eq!(event.path, path.join("a"));
311 assert!(event.flags.contains(StreamFlags::ITEM_REMOVED));
312 drop(handle);
313}
314
315#[test]
316fn test_event_stream_shutdown() {
317 use std::{fs, sync::mpsc, time::Duration};
318 use tempdir::TempDir;
319
320 let dir = TempDir::new("test_observe").unwrap();
321 let path = dir.path().canonicalize().unwrap();
322
323 let (tx, rx) = mpsc::channel();
324 let (stream, handle) = EventStream::new(&[&path], Duration::from_millis(50));
325 std::thread::spawn(move || {
326 stream.run({
327 let tx = tx.clone();
328 move |_| {
329 tx.send(()).unwrap();
330 true
331 }
332 });
333 tx.send(()).unwrap();
334 });
335
336 fs::write(path.join("b"), "b contents").unwrap();
337 rx.recv_timeout(Duration::from_millis(500)).unwrap();
338
339 drop(handle);
340 rx.recv_timeout(Duration::from_millis(500)).unwrap();
341}
342
343#[test]
344fn test_event_stream_shutdown_before_run() {
345 use std::time::Duration;
346 use tempdir::TempDir;
347
348 let dir = TempDir::new("test_observe").unwrap();
349 let path = dir.path().canonicalize().unwrap();
350
351 let (stream, handle) = EventStream::new(&[&path], Duration::from_millis(50));
352 drop(handle);
353 stream.run(|_| true);
354}