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 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 if flag.contains(StreamFlags::HISTORY_DONE) {
152 events.clear();
153 } else {
154 events.push(Event {
155 event_id: ids[p],
156 flags: flag,
157 path,
158 });
159 }
160 } else {
161 debug_assert!(false, "unknown flag set for fs event: {}", flags[p]);
162 }
163 }
164
165 if !events.is_empty() {
166 if !callback(events) {
167 fs::FSEventStreamStop(stream_ref);
168 cf::CFRunLoopStop(cf::CFRunLoopGetCurrent());
169 }
170 }
171 }
172 }
173}
174
175impl Drop for Handle {
176 fn drop(&mut self) {
177 let mut state = self.0.lock();
178 if let Lifecycle::Running(run_loop) = *state {
179 unsafe {
180 cf::CFRunLoopStop(run_loop);
181 }
182 }
183 *state = Lifecycle::Stopped;
184 }
185}
186
187// Synchronize with
188// /System/Library/Frameworks/CoreServices.framework/Versions/A/Frameworks/FSEvents.framework/Versions/A/Headers/FSEvents.h
189bitflags! {
190 #[repr(C)]
191 pub struct StreamFlags: u32 {
192 const NONE = 0x00000000;
193 const MUST_SCAN_SUBDIRS = 0x00000001;
194 const USER_DROPPED = 0x00000002;
195 const KERNEL_DROPPED = 0x00000004;
196 const IDS_WRAPPED = 0x00000008;
197 const HISTORY_DONE = 0x00000010;
198 const ROOT_CHANGED = 0x00000020;
199 const MOUNT = 0x00000040;
200 const UNMOUNT = 0x00000080;
201 const ITEM_CREATED = 0x00000100;
202 const ITEM_REMOVED = 0x00000200;
203 const INODE_META_MOD = 0x00000400;
204 const ITEM_RENAMED = 0x00000800;
205 const ITEM_MODIFIED = 0x00001000;
206 const FINDER_INFO_MOD = 0x00002000;
207 const ITEM_CHANGE_OWNER = 0x00004000;
208 const ITEM_XATTR_MOD = 0x00008000;
209 const IS_FILE = 0x00010000;
210 const IS_DIR = 0x00020000;
211 const IS_SYMLINK = 0x00040000;
212 const OWN_EVENT = 0x00080000;
213 const IS_HARDLINK = 0x00100000;
214 const IS_LAST_HARDLINK = 0x00200000;
215 const ITEM_CLONED = 0x400000;
216 }
217}
218
219impl std::fmt::Display for StreamFlags {
220 fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
221 if self.contains(StreamFlags::MUST_SCAN_SUBDIRS) {
222 let _d = write!(f, "MUST_SCAN_SUBDIRS ");
223 }
224 if self.contains(StreamFlags::USER_DROPPED) {
225 let _d = write!(f, "USER_DROPPED ");
226 }
227 if self.contains(StreamFlags::KERNEL_DROPPED) {
228 let _d = write!(f, "KERNEL_DROPPED ");
229 }
230 if self.contains(StreamFlags::IDS_WRAPPED) {
231 let _d = write!(f, "IDS_WRAPPED ");
232 }
233 if self.contains(StreamFlags::HISTORY_DONE) {
234 let _d = write!(f, "HISTORY_DONE ");
235 }
236 if self.contains(StreamFlags::ROOT_CHANGED) {
237 let _d = write!(f, "ROOT_CHANGED ");
238 }
239 if self.contains(StreamFlags::MOUNT) {
240 let _d = write!(f, "MOUNT ");
241 }
242 if self.contains(StreamFlags::UNMOUNT) {
243 let _d = write!(f, "UNMOUNT ");
244 }
245 if self.contains(StreamFlags::ITEM_CREATED) {
246 let _d = write!(f, "ITEM_CREATED ");
247 }
248 if self.contains(StreamFlags::ITEM_REMOVED) {
249 let _d = write!(f, "ITEM_REMOVED ");
250 }
251 if self.contains(StreamFlags::INODE_META_MOD) {
252 let _d = write!(f, "INODE_META_MOD ");
253 }
254 if self.contains(StreamFlags::ITEM_RENAMED) {
255 let _d = write!(f, "ITEM_RENAMED ");
256 }
257 if self.contains(StreamFlags::ITEM_MODIFIED) {
258 let _d = write!(f, "ITEM_MODIFIED ");
259 }
260 if self.contains(StreamFlags::FINDER_INFO_MOD) {
261 let _d = write!(f, "FINDER_INFO_MOD ");
262 }
263 if self.contains(StreamFlags::ITEM_CHANGE_OWNER) {
264 let _d = write!(f, "ITEM_CHANGE_OWNER ");
265 }
266 if self.contains(StreamFlags::ITEM_XATTR_MOD) {
267 let _d = write!(f, "ITEM_XATTR_MOD ");
268 }
269 if self.contains(StreamFlags::IS_FILE) {
270 let _d = write!(f, "IS_FILE ");
271 }
272 if self.contains(StreamFlags::IS_DIR) {
273 let _d = write!(f, "IS_DIR ");
274 }
275 if self.contains(StreamFlags::IS_SYMLINK) {
276 let _d = write!(f, "IS_SYMLINK ");
277 }
278 if self.contains(StreamFlags::OWN_EVENT) {
279 let _d = write!(f, "OWN_EVENT ");
280 }
281 if self.contains(StreamFlags::IS_LAST_HARDLINK) {
282 let _d = write!(f, "IS_LAST_HARDLINK ");
283 }
284 if self.contains(StreamFlags::IS_HARDLINK) {
285 let _d = write!(f, "IS_HARDLINK ");
286 }
287 if self.contains(StreamFlags::ITEM_CLONED) {
288 let _d = write!(f, "ITEM_CLONED ");
289 }
290 write!(f, "")
291 }
292}
293
294#[link(name = "CoreServices", kind = "framework")]
295extern "C" {
296 pub fn FSEventsGetCurrentEventId() -> u64;
297}
298
299#[test]
300fn test_event_stream() {
301 use std::{fs, sync::mpsc, time::Duration};
302 use tempdir::TempDir;
303
304 let dir = TempDir::new("test_observe").unwrap();
305 let path = dir.path().canonicalize().unwrap();
306 fs::write(path.join("a"), "a contents").unwrap();
307
308 let (tx, rx) = mpsc::channel();
309 let (stream, handle) = EventStream::new(&[&path], Duration::from_millis(50));
310 std::thread::spawn(move || stream.run(move |events| tx.send(events.to_vec()).is_ok()));
311
312 fs::write(path.join("b"), "b contents").unwrap();
313 let events = rx.recv_timeout(Duration::from_millis(500)).unwrap();
314 let event = events.last().unwrap();
315 assert_eq!(event.path, path.join("b"));
316 assert!(event.flags.contains(StreamFlags::ITEM_CREATED));
317
318 fs::remove_file(path.join("a")).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("a"));
322 assert!(event.flags.contains(StreamFlags::ITEM_REMOVED));
323 drop(handle);
324}
325
326#[test]
327fn test_event_stream_shutdown() {
328 use std::{fs, sync::mpsc, time::Duration};
329 use tempdir::TempDir;
330
331 let dir = TempDir::new("test_observe").unwrap();
332 let path = dir.path().canonicalize().unwrap();
333
334 let (tx, rx) = mpsc::channel();
335 let (stream, handle) = EventStream::new(&[&path], Duration::from_millis(50));
336 std::thread::spawn(move || {
337 stream.run({
338 let tx = tx.clone();
339 move |_| {
340 tx.send(()).unwrap();
341 true
342 }
343 });
344 tx.send(()).unwrap();
345 });
346
347 fs::write(path.join("b"), "b contents").unwrap();
348 rx.recv_timeout(Duration::from_millis(500)).unwrap();
349
350 drop(handle);
351 rx.recv_timeout(Duration::from_millis(500)).unwrap();
352}
353
354#[test]
355fn test_event_stream_shutdown_before_run() {
356 use std::time::Duration;
357 use tempdir::TempDir;
358
359 let dir = TempDir::new("test_observe").unwrap();
360 let path = dir.path().canonicalize().unwrap();
361
362 let (stream, handle) = EventStream::new(&[&path], Duration::from_millis(50));
363 drop(handle);
364 stream.run(|_| true);
365}