1#![cfg(target_os = "macos")]
2
3use bitflags::bitflags;
4use fsevent_sys::{self as fs, core_foundation as cf};
5use std::{
6 convert::AsRef,
7 ffi::{c_void, CStr, OsStr},
8 os::unix::ffi::OsStrExt,
9 path::{Path, PathBuf},
10 slice,
11 sync::mpsc::Sender,
12 time::Duration,
13};
14
15#[derive(Debug)]
16pub struct Event {
17 pub event_id: u64,
18 pub flags: StreamFlags,
19 pub path: PathBuf,
20}
21
22pub struct EventStream {
23 stream: fs::FSEventStreamRef,
24 _sender: Box<Sender<Vec<Event>>>,
25}
26
27unsafe impl Send for EventStream {}
28
29impl EventStream {
30 pub fn new(paths: &[&Path], latency: Duration, event_sender: Sender<Vec<Event>>) -> Self {
31 unsafe {
32 let sender = Box::new(event_sender);
33 let stream_context = fs::FSEventStreamContext {
34 version: 0,
35 info: sender.as_ref() as *const _ as *mut c_void,
36 retain: None,
37 release: None,
38 copy_description: None,
39 };
40
41 let cf_paths =
42 cf::CFArrayCreateMutable(cf::kCFAllocatorDefault, 0, &cf::kCFTypeArrayCallBacks);
43 assert!(!cf_paths.is_null());
44
45 for path in paths {
46 let path_bytes = path.as_os_str().as_bytes();
47 let cf_url = cf::CFURLCreateFromFileSystemRepresentation(
48 cf::kCFAllocatorDefault,
49 path_bytes.as_ptr() as *const i8,
50 path_bytes.len() as cf::CFIndex,
51 false,
52 );
53 let cf_path = cf::CFURLCopyFileSystemPath(cf_url, cf::kCFURLPOSIXPathStyle);
54 cf::CFArrayAppendValue(cf_paths, cf_path);
55 cf::CFRelease(cf_path);
56 cf::CFRelease(cf_url);
57 }
58
59 let stream = fs::FSEventStreamCreate(
60 cf::kCFAllocatorDefault,
61 callback,
62 &stream_context,
63 cf_paths,
64 fs::kFSEventStreamEventIdSinceNow,
65 latency.as_secs_f64(),
66 fs::kFSEventStreamCreateFlagFileEvents | fs::kFSEventStreamCreateFlagNoDefer,
67 );
68 cf::CFRelease(cf_paths);
69
70 EventStream {
71 stream,
72 _sender: sender,
73 }
74 }
75 }
76
77 pub fn run(self) {
78 unsafe {
79 fs::FSEventStreamScheduleWithRunLoop(
80 self.stream,
81 cf::CFRunLoopGetCurrent(),
82 cf::kCFRunLoopDefaultMode,
83 );
84
85 fs::FSEventStreamStart(self.stream);
86 cf::CFRunLoopRun();
87
88 fs::FSEventStreamFlushSync(self.stream);
89 fs::FSEventStreamStop(self.stream);
90 fs::FSEventStreamRelease(self.stream);
91 }
92 }
93}
94
95extern "C" fn callback(
96 stream_ref: fs::FSEventStreamRef,
97 info: *mut ::std::os::raw::c_void,
98 num: usize, // size_t numEvents
99 event_paths: *mut ::std::os::raw::c_void, // void *eventPaths
100 event_flags: *const ::std::os::raw::c_void, // const FSEventStreamEventFlags eventFlags[]
101 event_ids: *const ::std::os::raw::c_void, // const FSEventStreamEventId eventIds[]
102) {
103 unsafe {
104 let event_paths = event_paths as *const *const ::std::os::raw::c_char;
105 let e_ptr = event_flags as *mut u32;
106 let i_ptr = event_ids as *mut u64;
107 let sender = (info as *mut Sender<Vec<Event>>).as_mut().unwrap();
108
109 let paths = slice::from_raw_parts(event_paths, num);
110 let flags = slice::from_raw_parts_mut(e_ptr, num);
111 let ids = slice::from_raw_parts_mut(i_ptr, num);
112
113 let mut events = Vec::with_capacity(num);
114 for p in 0..num {
115 let path_c_str = CStr::from_ptr(paths[p]);
116 let path = PathBuf::from(OsStr::from_bytes(path_c_str.to_bytes()));
117 if let Some(flag) = StreamFlags::from_bits(flags[p]) {
118 events.push(Event {
119 event_id: ids[p],
120 flags: flag,
121 path,
122 });
123 } else {
124 debug_assert!(false, "unknown flag set for fs event: {}", flags[p]);
125 }
126 }
127
128 if sender.send(events).is_err() {
129 fs::FSEventStreamStop(stream_ref);
130 cf::CFRunLoopStop(cf::CFRunLoopGetCurrent());
131 }
132 }
133}
134
135// Synchronize with
136// /System/Library/Frameworks/CoreServices.framework/Versions/A/Frameworks/FSEvents.framework/Versions/A/Headers/FSEvents.h
137bitflags! {
138 #[repr(C)]
139 pub struct StreamFlags: u32 {
140 const NONE = 0x00000000;
141 const MUST_SCAN_SUBDIRS = 0x00000001;
142 const USER_DROPPED = 0x00000002;
143 const KERNEL_DROPPED = 0x00000004;
144 const IDS_WRAPPED = 0x00000008;
145 const HISTORY_DONE = 0x00000010;
146 const ROOT_CHANGED = 0x00000020;
147 const MOUNT = 0x00000040;
148 const UNMOUNT = 0x00000080;
149 const ITEM_CREATED = 0x00000100;
150 const ITEM_REMOVED = 0x00000200;
151 const INODE_META_MOD = 0x00000400;
152 const ITEM_RENAMED = 0x00000800;
153 const ITEM_MODIFIED = 0x00001000;
154 const FINDER_INFO_MOD = 0x00002000;
155 const ITEM_CHANGE_OWNER = 0x00004000;
156 const ITEM_XATTR_MOD = 0x00008000;
157 const IS_FILE = 0x00010000;
158 const IS_DIR = 0x00020000;
159 const IS_SYMLINK = 0x00040000;
160 const OWN_EVENT = 0x00080000;
161 const IS_HARDLINK = 0x00100000;
162 const IS_LAST_HARDLINK = 0x00200000;
163 const ITEM_CLONED = 0x400000;
164 }
165}
166
167impl std::fmt::Display for StreamFlags {
168 fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
169 if self.contains(StreamFlags::MUST_SCAN_SUBDIRS) {
170 let _d = write!(f, "MUST_SCAN_SUBDIRS ");
171 }
172 if self.contains(StreamFlags::USER_DROPPED) {
173 let _d = write!(f, "USER_DROPPED ");
174 }
175 if self.contains(StreamFlags::KERNEL_DROPPED) {
176 let _d = write!(f, "KERNEL_DROPPED ");
177 }
178 if self.contains(StreamFlags::IDS_WRAPPED) {
179 let _d = write!(f, "IDS_WRAPPED ");
180 }
181 if self.contains(StreamFlags::HISTORY_DONE) {
182 let _d = write!(f, "HISTORY_DONE ");
183 }
184 if self.contains(StreamFlags::ROOT_CHANGED) {
185 let _d = write!(f, "ROOT_CHANGED ");
186 }
187 if self.contains(StreamFlags::MOUNT) {
188 let _d = write!(f, "MOUNT ");
189 }
190 if self.contains(StreamFlags::UNMOUNT) {
191 let _d = write!(f, "UNMOUNT ");
192 }
193 if self.contains(StreamFlags::ITEM_CREATED) {
194 let _d = write!(f, "ITEM_CREATED ");
195 }
196 if self.contains(StreamFlags::ITEM_REMOVED) {
197 let _d = write!(f, "ITEM_REMOVED ");
198 }
199 if self.contains(StreamFlags::INODE_META_MOD) {
200 let _d = write!(f, "INODE_META_MOD ");
201 }
202 if self.contains(StreamFlags::ITEM_RENAMED) {
203 let _d = write!(f, "ITEM_RENAMED ");
204 }
205 if self.contains(StreamFlags::ITEM_MODIFIED) {
206 let _d = write!(f, "ITEM_MODIFIED ");
207 }
208 if self.contains(StreamFlags::FINDER_INFO_MOD) {
209 let _d = write!(f, "FINDER_INFO_MOD ");
210 }
211 if self.contains(StreamFlags::ITEM_CHANGE_OWNER) {
212 let _d = write!(f, "ITEM_CHANGE_OWNER ");
213 }
214 if self.contains(StreamFlags::ITEM_XATTR_MOD) {
215 let _d = write!(f, "ITEM_XATTR_MOD ");
216 }
217 if self.contains(StreamFlags::IS_FILE) {
218 let _d = write!(f, "IS_FILE ");
219 }
220 if self.contains(StreamFlags::IS_DIR) {
221 let _d = write!(f, "IS_DIR ");
222 }
223 if self.contains(StreamFlags::IS_SYMLINK) {
224 let _d = write!(f, "IS_SYMLINK ");
225 }
226 if self.contains(StreamFlags::OWN_EVENT) {
227 let _d = write!(f, "OWN_EVENT ");
228 }
229 if self.contains(StreamFlags::IS_LAST_HARDLINK) {
230 let _d = write!(f, "IS_LAST_HARDLINK ");
231 }
232 if self.contains(StreamFlags::IS_HARDLINK) {
233 let _d = write!(f, "IS_HARDLINK ");
234 }
235 if self.contains(StreamFlags::ITEM_CLONED) {
236 let _d = write!(f, "ITEM_CLONED ");
237 }
238 write!(f, "")
239 }
240}
241
242#[test]
243fn test_observe() {
244 use std::{fs, sync::mpsc, time::Duration};
245 use tempdir::TempDir;
246
247 let dir = TempDir::new("test_observe").unwrap();
248 let path = dir.path().canonicalize().unwrap();
249 fs::write(path.join("a"), "a contents").unwrap();
250
251 let (tx, rx) = mpsc::channel();
252 let stream = EventStream::new(&[&path], Duration::from_millis(50), tx);
253 std::thread::spawn(move || stream.run());
254
255 fs::write(path.join("b"), "b contents").unwrap();
256 let events = rx.recv_timeout(Duration::from_millis(500)).unwrap();
257 let event = events.last().unwrap();
258 assert_eq!(event.path, path.join("b"));
259 assert!(event.flags.contains(StreamFlags::ITEM_CREATED));
260
261 fs::remove_file(path.join("a")).unwrap();
262 let events = rx.recv_timeout(Duration::from_millis(500)).unwrap();
263 let event = events.last().unwrap();
264 assert_eq!(event.path, path.join("a"));
265 assert!(event.flags.contains(StreamFlags::ITEM_REMOVED));
266}