[rustc.git] / vendor / fsevent / src / lib.rs

#![cfg(target_os="macos")]
#![deny(
    trivial_numeric_casts,
    unstable_features,
    unused_import_braces,
    unused_qualifications
)]
#![cfg_attr(feature = "cargo-clippy", allow(unreadable_literal))]

#[macro_use]
extern crate bitflags;

extern crate fsevent_sys as fsevent;

use fsevent as fs;
use fsevent::core_foundation as cf;

use std::convert::AsRef;
use std::ffi::CStr;
use std::ptr;
use std::slice;
use std::slice::from_raw_parts_mut;
use std::str::from_utf8;

use std::sync::mpsc::Sender;

#[cfg(target_pointer_width = "64")]
type SafePointer = u64;

#[cfg(target_pointer_width = "32")]
type SafePointer = u32;

#[derive(Clone, Copy, Debug)]
pub struct FsEventRefWrapper {
    ptr: SafePointer,
}

impl From<*mut ::std::os::raw::c_void> for FsEventRefWrapper {
    fn from(raw: *mut ::std::os::raw::c_void) -> FsEventRefWrapper {
        let ptr = raw as SafePointer;
        Self { ptr }
    }
}

impl From<FsEventRefWrapper> for *mut ::std::os::raw::c_void {
    fn from(safe: FsEventRefWrapper) -> *mut ::std::os::raw::c_void {
        safe.ptr as *mut ::std::os::raw::c_void
    }
}

pub struct FsEvent {
    paths: Vec<String>,
    since_when: fs::FSEventStreamEventId,
    latency: cf::CFTimeInterval,
    flags: fs::FSEventStreamCreateFlags,
}

#[derive(Debug)]
pub struct Event {
    pub event_id: u64,
    pub flag: StreamFlags,
    pub path: String,
}

// Synchronize with
// /System/Library/Frameworks/CoreServices.framework/Versions/A/Frameworks/FSEvents.framework/Versions/A/Headers/FSEvents.h
bitflags! {
  #[repr(C)]
  pub struct StreamFlags: u32 {
    const NONE = 0x00000000;
    const MUST_SCAN_SUBDIRS = 0x00000001;
    const USER_DROPPED = 0x00000002;
    const KERNEL_DROPPED = 0x00000004;
    const IDS_WRAPPED = 0x00000008;
    const HISTORY_DONE = 0x00000010;
    const ROOT_CHANGED = 0x00000020;
    const MOUNT = 0x00000040;
    const UNMOUNT = 0x00000080;
    const ITEM_CREATED = 0x00000100;
    const ITEM_REMOVED = 0x00000200;
    const INODE_META_MOD = 0x00000400;
    const ITEM_RENAMED = 0x00000800;
    const ITEM_MODIFIED = 0x00001000;
    const FINDER_INFO_MOD = 0x00002000;
    const ITEM_CHANGE_OWNER = 0x00004000;
    const ITEM_XATTR_MOD = 0x00008000;
    const IS_FILE = 0x00010000;
    const IS_DIR = 0x00020000;
    const IS_SYMLINK = 0x00040000;
    const OWN_EVENT = 0x00080000;
    const IS_HARDLINK = 0x00100000;
    const IS_LAST_HARDLINK = 0x00200000;
    const ITEM_CLONED = 0x400000;
  }
}

impl std::fmt::Display for StreamFlags {
    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
        if self.contains(StreamFlags::MUST_SCAN_SUBDIRS) {
            let _d = write!(f, "MUST_SCAN_SUBDIRS ");
        }
        if self.contains(StreamFlags::USER_DROPPED) {
            let _d = write!(f, "USER_DROPPED ");
        }
        if self.contains(StreamFlags::KERNEL_DROPPED) {
            let _d = write!(f, "KERNEL_DROPPED ");
        }
        if self.contains(StreamFlags::IDS_WRAPPED) {
            let _d = write!(f, "IDS_WRAPPED ");
        }
        if self.contains(StreamFlags::HISTORY_DONE) {
            let _d = write!(f, "HISTORY_DONE ");
        }
        if self.contains(StreamFlags::ROOT_CHANGED) {
            let _d = write!(f, "ROOT_CHANGED ");
        }
        if self.contains(StreamFlags::MOUNT) {
            let _d = write!(f, "MOUNT ");
        }
        if self.contains(StreamFlags::UNMOUNT) {
            let _d = write!(f, "UNMOUNT ");
        }
        if self.contains(StreamFlags::ITEM_CREATED) {
            let _d = write!(f, "ITEM_CREATED ");
        }
        if self.contains(StreamFlags::ITEM_REMOVED) {
            let _d = write!(f, "ITEM_REMOVED ");
        }
        if self.contains(StreamFlags::INODE_META_MOD) {
            let _d = write!(f, "INODE_META_MOD ");
        }
        if self.contains(StreamFlags::ITEM_RENAMED) {
            let _d = write!(f, "ITEM_RENAMED ");
        }
        if self.contains(StreamFlags::ITEM_MODIFIED) {
            let _d = write!(f, "ITEM_MODIFIED ");
        }
        if self.contains(StreamFlags::FINDER_INFO_MOD) {
            let _d = write!(f, "FINDER_INFO_MOD ");
        }
        if self.contains(StreamFlags::ITEM_CHANGE_OWNER) {
            let _d = write!(f, "ITEM_CHANGE_OWNER ");
        }
        if self.contains(StreamFlags::ITEM_XATTR_MOD) {
            let _d = write!(f, "ITEM_XATTR_MOD ");
        }
        if self.contains(StreamFlags::IS_FILE) {
            let _d = write!(f, "IS_FILE ");
        }
        if self.contains(StreamFlags::IS_DIR) {
            let _d = write!(f, "IS_DIR ");
        }
        if self.contains(StreamFlags::IS_SYMLINK) {
            let _d = write!(f, "IS_SYMLINK ");
        }
        if self.contains(StreamFlags::OWN_EVENT) {
            let _d = write!(f, "OWN_EVENT ");
        }
        if self.contains(StreamFlags::IS_LAST_HARDLINK) {
            let _d = write!(f, "IS_LAST_HARDLINK ");
        }
        if self.contains(StreamFlags::IS_HARDLINK) {
            let _d = write!(f, "IS_HARDLINK ");
        }
        if self.contains(StreamFlags::ITEM_CLONED) {
            let _d = write!(f, "ITEM_CLONED ");
        }
        write!(f, "")
    }
}

fn default_stream_context(event_sender: *const Sender<Event>) -> fs::FSEventStreamContext {
    let ptr = event_sender as *mut ::std::os::raw::c_void;
    fs::FSEventStreamContext {
        version: 0,
        info: ptr,
        retain: None,
        release: None,
        copy_description: None,
    }
}

pub type Result<T> = std::result::Result<T, Error>;

#[derive(Debug)]
pub struct Error {
    msg: String,
}

impl std::error::Error for Error {
    fn description(&self) -> &str {
        &self.msg
    }
}

impl std::fmt::Display for Error {
    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
        self.msg.fmt(f)
    }
}

impl From<std::sync::mpsc::RecvTimeoutError> for Error {
    fn from(err: std::sync::mpsc::RecvTimeoutError) -> Error {
        Self {
            msg: err.to_string(),
        }
    }
}

impl FsEvent {
    pub fn new(paths: Vec<String>) -> Self {
        Self {
            paths,
            since_when: fs::kFSEventStreamEventIdSinceNow,
            latency: 0.0,
            flags: fs::kFSEventStreamCreateFlagFileEvents | fs::kFSEventStreamCreateFlagNoDefer,
        }
    }

    // https://github.com/thibaudgg/rb-fsevent/blob/master/ext/fsevent_watch/main.c
    pub fn append_path(&mut self, source: &str) -> Result<()> {
        self.paths.push(source.to_string());
        Ok(())
    }

    fn build_native_paths(&self) -> Result<cf::CFMutableArrayRef> {
        let native_paths = unsafe {
            cf::CFArrayCreateMutable(cf::kCFAllocatorDefault, 0, &cf::kCFTypeArrayCallBacks)
        };

        if native_paths == std::ptr::null_mut() {
            Err(Error {
                msg: "Unable to allocate CFMutableArrayRef".to_string(),
            })
        } else {
            for path in &self.paths {
                unsafe {
                    let mut err = ptr::null_mut();
                    let cf_path = cf::str_path_to_cfstring_ref(path, &mut err);
                    if !err.is_null() {
                        let cf_str = cf::CFCopyDescription(err as cf::CFRef);
                        let mut buf = [0; 1024];
                        cf::CFStringGetCString(
                            cf_str,
                            buf.as_mut_ptr(),
                            buf.len() as cf::CFIndex,
                            cf::kCFStringEncodingUTF8,
                        );
                        return Err(Error {
                            msg: CStr::from_ptr(buf.as_ptr())
                                .to_str()
                                .unwrap_or("Unknown error")
                                .to_string(),
                        });
                    } else {
                        cf::CFArrayAppendValue(native_paths, cf_path);
                        cf::CFRelease(cf_path);
                    }
                }
            }

            Ok(native_paths)
        }
    }

    fn internal_observe(
        since_when: fs::FSEventStreamEventId,
        latency: cf::CFTimeInterval,
        flags: fs::FSEventStreamCreateFlags,
        paths: FsEventRefWrapper,
        event_sender: Sender<Event>,
        subscription_handle_sender: Option<Sender<FsEventRefWrapper>>,
    ) -> Result<()> {
        let stream_context = default_stream_context(&event_sender);
        let paths = paths.into();

        unsafe {
            let stream = fs::FSEventStreamCreate(
                cf::kCFAllocatorDefault,
                callback,
                &stream_context,
                paths,
                since_when,
                latency,
                flags,
            );

            // fs::FSEventStreamShow(stream);

            match subscription_handle_sender {
                Some(ret_tx) => {
                    let runloop_ref = cf::CFRunLoopGetCurrent();
                    let runloop_ref_safe = FsEventRefWrapper::from(runloop_ref);
                    let ptr_val = runloop_ref_safe.ptr.clone();
                    ret_tx
                        .send(runloop_ref_safe)
                        .expect(&format!("Unable to return CFRunLoopRef ({:#X})", ptr_val));
                }
                None => {}
            }

            fs::FSEventStreamScheduleWithRunLoop(
                stream,
                cf::CFRunLoopGetCurrent(),
                cf::kCFRunLoopDefaultMode,
            );

            fs::FSEventStreamStart(stream);
            cf::CFRunLoopRun();

            fs::FSEventStreamFlushSync(stream);
            fs::FSEventStreamStop(stream);
        }

        Ok(())
    }

    pub fn observe(&self, event_sender: Sender<Event>) {
        let native_paths = self
            .build_native_paths()
            .expect("Unable to build CFMutableArrayRef of watched paths.");
        let safe_native_paths = FsEventRefWrapper::from(native_paths);
        Self::internal_observe(
            self.since_when,
            self.latency,
            self.flags,
            safe_native_paths,
            event_sender,
            None,
        )
        .unwrap();
    }

    pub fn observe_async(&self, event_sender: Sender<Event>) -> Result<FsEventRefWrapper> {
        let (ret_tx, ret_rx) = std::sync::mpsc::channel();
        let native_paths = self.build_native_paths()?;
        let safe_native_paths = FsEventRefWrapper::from(native_paths);

        let since_when = self.since_when;
        let latency = self.latency;
        let flags = self.flags;
        std::thread::spawn(move || {
            Self::internal_observe(
                since_when,
                latency,
                flags,
                safe_native_paths,
                event_sender,
                Some(ret_tx),
            )
        });

        match ret_rx.recv_timeout(std::time::Duration::from_secs(5)) {
            Ok(v) => Ok(v),
            Err(e) => Err(Error::from(e)),
        }
    }

    pub fn shutdown_observe(&self, handle: FsEventRefWrapper) {
        unsafe { cf::CFRunLoopStop(handle.into()) };
    }
}

#[allow(unused_variables)]
extern "C" fn callback(
    stream_ref: fs::FSEventStreamRef,
    info: *mut ::std::os::raw::c_void,
    num_events: usize,                          // size_t numEvents
    event_paths: *mut ::std::os::raw::c_void,   // void *eventPaths
    event_flags: *const ::std::os::raw::c_void, // const FSEventStreamEventFlags eventFlags[]
    event_ids: *const ::std::os::raw::c_void,   // const FSEventStreamEventId eventIds[]
) {
    unsafe {
        let event_paths = event_paths as *const *const ::std::os::raw::c_char;
        let num = num_events;
        let e_ptr = event_flags as *mut u32;
        let i_ptr = event_ids as *mut u64;
        let sender = info as *mut Sender<Event>;

        let paths: &[*const ::std::os::raw::c_char] =
            std::mem::transmute(slice::from_raw_parts(event_paths, num));
        let flags = from_raw_parts_mut(e_ptr, num);
        let ids = from_raw_parts_mut(i_ptr, num);

        for p in 0..num {
            let i = CStr::from_ptr(paths[p]).to_bytes();
            let path = from_utf8(i).expect("Invalid UTF8 string.");
            let flag: StreamFlags = StreamFlags::from_bits(flags[p]).expect(
                format!("Unable to decode StreamFlags: {} for {}", flags[p], path).as_ref(),
            );
            // println!("{}: {}", ids[p], flag);

            let event = Event {
                event_id: ids[p],
                flag,
                path: path.to_string(),
            };
            let _s = (*sender).send(event);
        }
    }
}
Commit	Line	Data
f035d41b XL	1	#![cfg(target_os="macos")]
	2	#![deny(
	3	trivial_numeric_casts,
	4	unstable_features,
	5	unused_import_braces,
	6	unused_qualifications
	7	)]
	8	#![cfg_attr(feature = "cargo-clippy", allow(unreadable_literal))]
	9
	10	#[macro_use]
	11	extern crate bitflags;
	12
	13	extern crate fsevent_sys as fsevent;
	14
	15	use fsevent as fs;
	16	use fsevent::core_foundation as cf;
	17
	18	use std::convert::AsRef;
	19	use std::ffi::CStr;
	20	use std::ptr;
	21	use std::slice;
	22	use std::slice::from_raw_parts_mut;
	23	use std::str::from_utf8;
	24
	25	use std::sync::mpsc::Sender;
	26
	27	#[cfg(target_pointer_width = "64")]
	28	type SafePointer = u64;
	29
	30	#[cfg(target_pointer_width = "32")]
	31	type SafePointer = u32;
	32
	33	#[derive(Clone, Copy, Debug)]
	34	pub struct FsEventRefWrapper {
	35	ptr: SafePointer,
	36	}
	37
	38	impl From<*mut ::std::os::raw::c_void> for FsEventRefWrapper {
	39	fn from(raw: *mut ::std::os::raw::c_void) -> FsEventRefWrapper {
	40	let ptr = raw as SafePointer;
	41	Self { ptr }
	42	}
	43	}
	44
	45	impl From<FsEventRefWrapper> for *mut ::std::os::raw::c_void {
	46	fn from(safe: FsEventRefWrapper) -> *mut ::std::os::raw::c_void {
	47	safe.ptr as *mut ::std::os::raw::c_void
	48	}
	49	}
	50
	51	pub struct FsEvent {
	52	paths: Vec<String>,
	53	since_when: fs::FSEventStreamEventId,
	54	latency: cf::CFTimeInterval,
	55	flags: fs::FSEventStreamCreateFlags,
	56	}
	57
	58	#[derive(Debug)]
	59	pub struct Event {
	60	pub event_id: u64,
	61	pub flag: StreamFlags,
	62	pub path: String,
	63	}
	64
65	// Synchronize with
66	// /System/Library/Frameworks/CoreServices.framework/Versions/A/Frameworks/FSEvents.framework/Versions/A/Headers/FSEvents.h
67	bitflags! {
68	#[repr(C)]
69	pub struct StreamFlags: u32 {
70	const NONE = 0x00000000;
71	const MUST_SCAN_SUBDIRS = 0x00000001;
72	const USER_DROPPED = 0x00000002;
73	const KERNEL_DROPPED = 0x00000004;
74	const IDS_WRAPPED = 0x00000008;
75	const HISTORY_DONE = 0x00000010;
76	const ROOT_CHANGED = 0x00000020;
77	const MOUNT = 0x00000040;
78	const UNMOUNT = 0x00000080;
79	const ITEM_CREATED = 0x00000100;
80	const ITEM_REMOVED = 0x00000200;
81	const INODE_META_MOD = 0x00000400;
82	const ITEM_RENAMED = 0x00000800;
83	const ITEM_MODIFIED = 0x00001000;
84	const FINDER_INFO_MOD = 0x00002000;
85	const ITEM_CHANGE_OWNER = 0x00004000;
86	const ITEM_XATTR_MOD = 0x00008000;
87	const IS_FILE = 0x00010000;
88	const IS_DIR = 0x00020000;
89	const IS_SYMLINK = 0x00040000;
90	const OWN_EVENT = 0x00080000;
91	const IS_HARDLINK = 0x00100000;
92	const IS_LAST_HARDLINK = 0x00200000;
93	const ITEM_CLONED = 0x400000;
94	}
95	}
96
97	impl std::fmt::Display for StreamFlags {
98	fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
99	if self.contains(StreamFlags::MUST_SCAN_SUBDIRS) {
100	let _d = write!(f, "MUST_SCAN_SUBDIRS ");
101	}
102	if self.contains(StreamFlags::USER_DROPPED) {
103	let _d = write!(f, "USER_DROPPED ");
104	}
105	if self.contains(StreamFlags::KERNEL_DROPPED) {
106	let _d = write!(f, "KERNEL_DROPPED ");
107	}
108	if self.contains(StreamFlags::IDS_WRAPPED) {
109	let _d = write!(f, "IDS_WRAPPED ");
110	}
111	if self.contains(StreamFlags::HISTORY_DONE) {
112	let _d = write!(f, "HISTORY_DONE ");
113	}
114	if self.contains(StreamFlags::ROOT_CHANGED) {
115	let _d = write!(f, "ROOT_CHANGED ");
116	}
117	if self.contains(StreamFlags::MOUNT) {
118	let _d = write!(f, "MOUNT ");
119	}
120	if self.contains(StreamFlags::UNMOUNT) {
121	let _d = write!(f, "UNMOUNT ");
122	}
123	if self.contains(StreamFlags::ITEM_CREATED) {
124	let _d = write!(f, "ITEM_CREATED ");
125	}
126	if self.contains(StreamFlags::ITEM_REMOVED) {
127	let _d = write!(f, "ITEM_REMOVED ");
128	}
129	if self.contains(StreamFlags::INODE_META_MOD) {
130	let _d = write!(f, "INODE_META_MOD ");
131	}
132	if self.contains(StreamFlags::ITEM_RENAMED) {
133	let _d = write!(f, "ITEM_RENAMED ");
134	}
135	if self.contains(StreamFlags::ITEM_MODIFIED) {
136	let _d = write!(f, "ITEM_MODIFIED ");
137	}
138	if self.contains(StreamFlags::FINDER_INFO_MOD) {
139	let _d = write!(f, "FINDER_INFO_MOD ");
140	}
141	if self.contains(StreamFlags::ITEM_CHANGE_OWNER) {
142	let _d = write!(f, "ITEM_CHANGE_OWNER ");
143	}
144	if self.contains(StreamFlags::ITEM_XATTR_MOD) {
145	let _d = write!(f, "ITEM_XATTR_MOD ");
146	}
147	if self.contains(StreamFlags::IS_FILE) {
148	let _d = write!(f, "IS_FILE ");
149	}
150	if self.contains(StreamFlags::IS_DIR) {
151	let _d = write!(f, "IS_DIR ");
152	}
153	if self.contains(StreamFlags::IS_SYMLINK) {
154	let _d = write!(f, "IS_SYMLINK ");
155	}
156	if self.contains(StreamFlags::OWN_EVENT) {
157	let _d = write!(f, "OWN_EVENT ");
158	}
159	if self.contains(StreamFlags::IS_LAST_HARDLINK) {
160	let _d = write!(f, "IS_LAST_HARDLINK ");
161	}
162	if self.contains(StreamFlags::IS_HARDLINK) {
163	let _d = write!(f, "IS_HARDLINK ");
164	}
165	if self.contains(StreamFlags::ITEM_CLONED) {
166	let _d = write!(f, "ITEM_CLONED ");
167	}
168	write!(f, "")
169	}
170	}
171
172	fn default_stream_context(event_sender: *const Sender<Event>) -> fs::FSEventStreamContext {
173	let ptr = event_sender as *mut ::std::os::raw::c_void;
174	fs::FSEventStreamContext {
175	version: 0,
176	info: ptr,
177	retain: None,
178	release: None,
179	copy_description: None,
180	}
181	}
182
183	pub type Result<T> = std::result::Result<T, Error>;
184
185	#[derive(Debug)]
186	pub struct Error {
187	msg: String,
188	}
189
190	impl std::error::Error for Error {
191	fn description(&self) -> &str {
192	&self.msg
193	}
194	}
195
196	impl std::fmt::Display for Error {
197	fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
198	self.msg.fmt(f)
199	}
200	}
201
202	impl From<std::sync::mpsc::RecvTimeoutError> for Error {
203	fn from(err: std::sync::mpsc::RecvTimeoutError) -> Error {
204	Self {
205	msg: err.to_string(),
206	}
207	}
208	}
209
210	impl FsEvent {
211	pub fn new(paths: Vec<String>) -> Self {
212	Self {
213	paths,
214	since_when: fs::kFSEventStreamEventIdSinceNow,
215	latency: 0.0,
216	flags: fs::kFSEventStreamCreateFlagFileEvents \| fs::kFSEventStreamCreateFlagNoDefer,
217	}
218	}
219
220	// https://github.com/thibaudgg/rb-fsevent/blob/master/ext/fsevent_watch/main.c
221	pub fn append_path(&mut self, source: &str) -> Result<()> {
222	self.paths.push(source.to_string());
223	Ok(())
224	}
225
226	fn build_native_paths(&self) -> Result<cf::CFMutableArrayRef> {
227	let native_paths = unsafe {
228	cf::CFArrayCreateMutable(cf::kCFAllocatorDefault, 0, &cf::kCFTypeArrayCallBacks)
229	};
230
231	if native_paths == std::ptr::null_mut() {
232	Err(Error {
233	msg: "Unable to allocate CFMutableArrayRef".to_string(),
234	})
235	} else {
236	for path in &self.paths {
237	unsafe {
238	let mut err = ptr::null_mut();
239	let cf_path = cf::str_path_to_cfstring_ref(path, &mut err);
240	if !err.is_null() {
241	let cf_str = cf::CFCopyDescription(err as cf::CFRef);
242	let mut buf = [0; 1024];
243	cf::CFStringGetCString(
244	cf_str,
245	buf.as_mut_ptr(),
246	buf.len() as cf::CFIndex,
247	cf::kCFStringEncodingUTF8,
248	);
249	return Err(Error {
250	msg: CStr::from_ptr(buf.as_ptr())
251	.to_str()
252	.unwrap_or("Unknown error")
253	.to_string(),
254	});
255	} else {
256	cf::CFArrayAppendValue(native_paths, cf_path);
257	cf::CFRelease(cf_path);
258	}
259	}
260	}
261
262	Ok(native_paths)
263	}
264	}
265
266	fn internal_observe(
267	since_when: fs::FSEventStreamEventId,
268	latency: cf::CFTimeInterval,
269	flags: fs::FSEventStreamCreateFlags,
270	paths: FsEventRefWrapper,
271	event_sender: Sender<Event>,
272	subscription_handle_sender: Option<Sender<FsEventRefWrapper>>,
273	) -> Result<()> {
274	let stream_context = default_stream_context(&event_sender);
275	let paths = paths.into();
276
277	unsafe {
278	let stream = fs::FSEventStreamCreate(
279	cf::kCFAllocatorDefault,
280	callback,
281	&stream_context,
282	paths,
283	since_when,
284	latency,
285	flags,
286	);
287
288	// fs::FSEventStreamShow(stream);
289
290	match subscription_handle_sender {
291	Some(ret_tx) => {
292	let runloop_ref = cf::CFRunLoopGetCurrent();
293	let runloop_ref_safe = FsEventRefWrapper::from(runloop_ref);
294	let ptr_val = runloop_ref_safe.ptr.clone();
295	ret_tx
296	.send(runloop_ref_safe)
297	.expect(&format!("Unable to return CFRunLoopRef ({:#X})", ptr_val));
298	}
299	None => {}
300	}
301
302	fs::FSEventStreamScheduleWithRunLoop(
303	stream,
304	cf::CFRunLoopGetCurrent(),
305	cf::kCFRunLoopDefaultMode,
306	);
307
308	fs::FSEventStreamStart(stream);
309	cf::CFRunLoopRun();
310
311	fs::FSEventStreamFlushSync(stream);
312	fs::FSEventStreamStop(stream);
313	}
314
315	Ok(())
316	}
317
318	pub fn observe(&self, event_sender: Sender<Event>) {
319	let native_paths = self
320	.build_native_paths()
321	.expect("Unable to build CFMutableArrayRef of watched paths.");
322	let safe_native_paths = FsEventRefWrapper::from(native_paths);
323	Self::internal_observe(
324	self.since_when,
325	self.latency,
326	self.flags,
327	safe_native_paths,
328	event_sender,
329	None,
330	)
331	.unwrap();
332	}
333
334	pub fn observe_async(&self, event_sender: Sender<Event>) -> Result<FsEventRefWrapper> {
335	let (ret_tx, ret_rx) = std::sync::mpsc::channel();
336	let native_paths = self.build_native_paths()?;
337	let safe_native_paths = FsEventRefWrapper::from(native_paths);
338
339	let since_when = self.since_when;
340	let latency = self.latency;
341	let flags = self.flags;
342	std::thread::spawn(move \|\| {
343	Self::internal_observe(
344	since_when,
345	latency,
346	flags,
347	safe_native_paths,
348	event_sender,
349	Some(ret_tx),
350	)
351	});
352
353	match ret_rx.recv_timeout(std::time::Duration::from_secs(5)) {
354	Ok(v) => Ok(v),
355	Err(e) => Err(Error::from(e)),
356	}
357	}
358
359	pub fn shutdown_observe(&self, handle: FsEventRefWrapper) {
360	unsafe { cf::CFRunLoopStop(handle.into()) };
361	}
362	}
363
364	#[allow(unused_variables)]
365	extern "C" fn callback(
366	stream_ref: fs::FSEventStreamRef,
367	info: *mut ::std::os::raw::c_void,
368	num_events: usize, // size_t numEvents
369	event_paths: mut ::std::os::raw::c_void, // void eventPaths
370	event_flags: *const ::std::os::raw::c_void, // const FSEventStreamEventFlags eventFlags[]
371	event_ids: *const ::std::os::raw::c_void, // const FSEventStreamEventId eventIds[]
372	) {
373	unsafe {
374	let event_paths = event_paths as const const ::std::os::raw::c_char;
375	let num = num_events;
376	let e_ptr = event_flags as *mut u32;
377	let i_ptr = event_ids as *mut u64;
378	let sender = info as *mut Sender<Event>;
379
380	let paths: &[*const ::std::os::raw::c_char] =
381	std::mem::transmute(slice::from_raw_parts(event_paths, num));
382	let flags = from_raw_parts_mut(e_ptr, num);
383	let ids = from_raw_parts_mut(i_ptr, num);
384
385	for p in 0..num {
386	let i = CStr::from_ptr(paths[p]).to_bytes();
387	let path = from_utf8(i).expect("Invalid UTF8 string.");
388	let flag: StreamFlags = StreamFlags::from_bits(flags[p]).expect(
389	format!("Unable to decode StreamFlags: {} for {}", flags[p], path).as_ref(),
390	);
391	// println!("{}: {}", ids[p], flag);
392
393	let event = Event {
394	event_id: ids[p],
395	flag,
396	path: path.to_string(),
397	};
398	let _s = (*sender).send(event);
399	}
400	}
401	}