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 FSEventsGetCurrentEventId(),
78 latency.as_secs_f64(),
79 fs::kFSEventStreamCreateFlagFileEvents
80 | fs::kFSEventStreamCreateFlagNoDefer
81 | fs::kFSEventStreamCreateFlagWatchRoot,
82 );
83 cf::CFRelease(cf_paths);
84
85 fs::FSEventStreamScheduleWithRunLoop(
86 stream,
87 cf::CFRunLoopGetCurrent(),
88 cf::kCFRunLoopDefaultMode,
89 );
90 fs::FSEventStreamStart(stream);
91 fs::FSEventStreamFlushSync(stream);
92 fs::FSEventStreamStop(stream);
93
94 let state = Arc::new(Mutex::new(Lifecycle::New));
95
96 (
97 EventStream {
98 stream,
99 state: state.clone(),
100 callback,
101 },
102 Handle(state),
103 )
104 }
105 }
106
107 pub fn run<F>(mut self, f: F)
108 where
109 F: FnMut(Vec<Event>) -> bool + 'static,
110 {
111 *self.callback = Some(Box::new(f));
112 unsafe {
113 let run_loop = cf::CFRunLoopGetCurrent();
114 {
115 let mut state = self.state.lock();
116 match *state {
117 Lifecycle::New => *state = Lifecycle::Running(run_loop),
118 Lifecycle::Running(_) => unreachable!(),
119 Lifecycle::Stopped => return,
120 }
121 }
122 fs::FSEventStreamScheduleWithRunLoop(self.stream, run_loop, cf::kCFRunLoopDefaultMode);
123 fs::FSEventStreamStart(self.stream);
124 cf::CFRunLoopRun();
125 fs::FSEventStreamRelease(self.stream);
126 }
127 }
128
129 extern "C" fn trampoline(
130 stream_ref: fs::FSEventStreamRef,
131 info: *mut ::std::os::raw::c_void,
132 num: usize, // size_t numEvents
133 event_paths: *mut ::std::os::raw::c_void, // void *eventPaths
134 event_flags: *const ::std::os::raw::c_void, // const FSEventStreamEventFlags eventFlags[]
135 event_ids: *const ::std::os::raw::c_void, // const FSEventStreamEventId eventIds[]
136 ) {
137 unsafe {
138 let event_paths = event_paths as *const *const ::std::os::raw::c_char;
139 let e_ptr = event_flags as *mut u32;
140 let i_ptr = event_ids as *mut u64;
141 let callback_ptr = (info as *mut Option<RunCallback>).as_mut().unwrap();
142 let callback = if let Some(callback) = callback_ptr.as_mut() {
143 callback
144 } else {
145 return;
146 };
147
148 let paths = slice::from_raw_parts(event_paths, num);
149 let flags = slice::from_raw_parts_mut(e_ptr, num);
150 let ids = slice::from_raw_parts_mut(i_ptr, num);
151
152 let mut events = Vec::with_capacity(num);
153 for p in 0..num {
154 let path_c_str = CStr::from_ptr(paths[p]);
155 let path = PathBuf::from(OsStr::from_bytes(path_c_str.to_bytes()));
156 if let Some(flag) = StreamFlags::from_bits(flags[p]) {
157 if flag.contains(StreamFlags::HISTORY_DONE) {
158 events.clear();
159 } else {
160 events.push(Event {
161 event_id: ids[p],
162 flags: flag,
163 path,
164 });
165 }
166 } else {
167 debug_assert!(false, "unknown flag set for fs event: {}", flags[p]);
168 }
169 }
170
171 if !events.is_empty() {
172 if !callback(events) {
173 fs::FSEventStreamStop(stream_ref);
174 cf::CFRunLoopStop(cf::CFRunLoopGetCurrent());
175 }
176 }
177 }
178 }
179}
180
181impl Drop for Handle {
182 fn drop(&mut self) {
183 let mut state = self.0.lock();
184 if let Lifecycle::Running(run_loop) = *state {
185 unsafe {
186 cf::CFRunLoopStop(run_loop);
187 }
188 }
189 *state = Lifecycle::Stopped;
190 }
191}
192
193// Synchronize with
194// /System/Library/Frameworks/CoreServices.framework/Versions/A/Frameworks/FSEvents.framework/Versions/A/Headers/FSEvents.h
195bitflags! {
196 #[repr(C)]
197 pub struct StreamFlags: u32 {
198 const NONE = 0x00000000;
199 const MUST_SCAN_SUBDIRS = 0x00000001;
200 const USER_DROPPED = 0x00000002;
201 const KERNEL_DROPPED = 0x00000004;
202 const IDS_WRAPPED = 0x00000008;
203 const HISTORY_DONE = 0x00000010;
204 const ROOT_CHANGED = 0x00000020;
205 const MOUNT = 0x00000040;
206 const UNMOUNT = 0x00000080;
207 const ITEM_CREATED = 0x00000100;
208 const ITEM_REMOVED = 0x00000200;
209 const INODE_META_MOD = 0x00000400;
210 const ITEM_RENAMED = 0x00000800;
211 const ITEM_MODIFIED = 0x00001000;
212 const FINDER_INFO_MOD = 0x00002000;
213 const ITEM_CHANGE_OWNER = 0x00004000;
214 const ITEM_XATTR_MOD = 0x00008000;
215 const IS_FILE = 0x00010000;
216 const IS_DIR = 0x00020000;
217 const IS_SYMLINK = 0x00040000;
218 const OWN_EVENT = 0x00080000;
219 const IS_HARDLINK = 0x00100000;
220 const IS_LAST_HARDLINK = 0x00200000;
221 const ITEM_CLONED = 0x400000;
222 }
223}
224
225impl std::fmt::Display for StreamFlags {
226 fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
227 if self.contains(StreamFlags::MUST_SCAN_SUBDIRS) {
228 let _d = write!(f, "MUST_SCAN_SUBDIRS ");
229 }
230 if self.contains(StreamFlags::USER_DROPPED) {
231 let _d = write!(f, "USER_DROPPED ");
232 }
233 if self.contains(StreamFlags::KERNEL_DROPPED) {
234 let _d = write!(f, "KERNEL_DROPPED ");
235 }
236 if self.contains(StreamFlags::IDS_WRAPPED) {
237 let _d = write!(f, "IDS_WRAPPED ");
238 }
239 if self.contains(StreamFlags::HISTORY_DONE) {
240 let _d = write!(f, "HISTORY_DONE ");
241 }
242 if self.contains(StreamFlags::ROOT_CHANGED) {
243 let _d = write!(f, "ROOT_CHANGED ");
244 }
245 if self.contains(StreamFlags::MOUNT) {
246 let _d = write!(f, "MOUNT ");
247 }
248 if self.contains(StreamFlags::UNMOUNT) {
249 let _d = write!(f, "UNMOUNT ");
250 }
251 if self.contains(StreamFlags::ITEM_CREATED) {
252 let _d = write!(f, "ITEM_CREATED ");
253 }
254 if self.contains(StreamFlags::ITEM_REMOVED) {
255 let _d = write!(f, "ITEM_REMOVED ");
256 }
257 if self.contains(StreamFlags::INODE_META_MOD) {
258 let _d = write!(f, "INODE_META_MOD ");
259 }
260 if self.contains(StreamFlags::ITEM_RENAMED) {
261 let _d = write!(f, "ITEM_RENAMED ");
262 }
263 if self.contains(StreamFlags::ITEM_MODIFIED) {
264 let _d = write!(f, "ITEM_MODIFIED ");
265 }
266 if self.contains(StreamFlags::FINDER_INFO_MOD) {
267 let _d = write!(f, "FINDER_INFO_MOD ");
268 }
269 if self.contains(StreamFlags::ITEM_CHANGE_OWNER) {
270 let _d = write!(f, "ITEM_CHANGE_OWNER ");
271 }
272 if self.contains(StreamFlags::ITEM_XATTR_MOD) {
273 let _d = write!(f, "ITEM_XATTR_MOD ");
274 }
275 if self.contains(StreamFlags::IS_FILE) {
276 let _d = write!(f, "IS_FILE ");
277 }
278 if self.contains(StreamFlags::IS_DIR) {
279 let _d = write!(f, "IS_DIR ");
280 }
281 if self.contains(StreamFlags::IS_SYMLINK) {
282 let _d = write!(f, "IS_SYMLINK ");
283 }
284 if self.contains(StreamFlags::OWN_EVENT) {
285 let _d = write!(f, "OWN_EVENT ");
286 }
287 if self.contains(StreamFlags::IS_LAST_HARDLINK) {
288 let _d = write!(f, "IS_LAST_HARDLINK ");
289 }
290 if self.contains(StreamFlags::IS_HARDLINK) {
291 let _d = write!(f, "IS_HARDLINK ");
292 }
293 if self.contains(StreamFlags::ITEM_CLONED) {
294 let _d = write!(f, "ITEM_CLONED ");
295 }
296 write!(f, "")
297 }
298}
299
300#[link(name = "CoreServices", kind = "framework")]
301extern "C" {
302 pub fn FSEventsGetCurrentEventId() -> u64;
303}
304
305#[test]
306fn test_event_stream() {
307 use std::{fs, sync::mpsc, time::Duration};
308 use tempdir::TempDir;
309
310 let dir = TempDir::new("test_observe").unwrap();
311 let path = dir.path().canonicalize().unwrap();
312 fs::write(path.join("a"), "a contents").unwrap();
313
314 let (tx, rx) = mpsc::channel();
315 let (stream, handle) = EventStream::new(&[&path], Duration::from_millis(50));
316 std::thread::spawn(move || stream.run(move |events| tx.send(events.to_vec()).is_ok()));
317
318 fs::write(path.join("b"), "b contents").unwrap();
319 let events = rx.recv_timeout(Duration::from_millis(500)).unwrap();
320 let event = events.last().unwrap();
321 assert_eq!(event.path, path.join("b"));
322 assert!(event.flags.contains(StreamFlags::ITEM_CREATED));
323
324 fs::remove_file(path.join("a")).unwrap();
325 let events = rx.recv_timeout(Duration::from_millis(500)).unwrap();
326 let event = events.last().unwrap();
327 assert_eq!(event.path, path.join("a"));
328 assert!(event.flags.contains(StreamFlags::ITEM_REMOVED));
329 drop(handle);
330}
331
332#[test]
333fn test_event_stream_shutdown() {
334 use std::{fs, sync::mpsc, time::Duration};
335 use tempdir::TempDir;
336
337 let dir = TempDir::new("test_observe").unwrap();
338 let path = dir.path().canonicalize().unwrap();
339
340 let (tx, rx) = mpsc::channel();
341 let (stream, handle) = EventStream::new(&[&path], Duration::from_millis(50));
342 std::thread::spawn(move || {
343 stream.run({
344 let tx = tx.clone();
345 move |_| {
346 tx.send(()).unwrap();
347 true
348 }
349 });
350 tx.send(()).unwrap();
351 });
352
353 fs::write(path.join("b"), "b contents").unwrap();
354 rx.recv_timeout(Duration::from_millis(500)).unwrap();
355
356 drop(handle);
357 rx.recv_timeout(Duration::from_millis(500)).unwrap();
358}
359
360#[test]
361fn test_event_stream_shutdown_before_run() {
362 use std::time::Duration;
363 use tempdir::TempDir;
364
365 let dir = TempDir::new("test_observe").unwrap();
366 let path = dir.path().canonicalize().unwrap();
367
368 let (stream, handle) = EventStream::new(&[&path], Duration::from_millis(50));
369 drop(handle);
370 stream.run(|_| true);
371}