#![warn(missing_docs)]
extern crate kernel32;
use winapi::shared::minwindef::TRUE;
use winapi::shared::winerror::ERROR_OPERATION_ABORTED;
use winapi::um::fileapi;
use winapi::um::handleapi::INVALID_HANDLE_VALUE;
use winapi::um::minwinbase::{LPOVERLAPPED, OVERLAPPED};
use winapi::um::winbase::{self, INFINITE, WAIT_OBJECT_0};
use winapi::um::winnt::{self, FILE_NOTIFY_INFORMATION, HANDLE};
use super::debounce::{Debounce, EventTx};
use super::{op, DebouncedEvent, Error, Op, RawEvent, RecursiveMode, Result, Watcher};
use std::collections::HashMap;
use std::env;
use std::ffi::OsString;
use std::mem;
use std::os::raw::c_void;
use std::os::windows::ffi::{OsStrExt, OsStringExt};
use std::path::{Path, PathBuf};
use std::ptr;
use std::slice;
use std::sync::mpsc::{channel, Receiver, Sender};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;
const BUF_SIZE: u32 = 16384;
static mut COOKIE_COUNTER: u32 = 0;
#[derive(Clone)]
struct ReadData {
dir: PathBuf, file: Option<PathBuf>, complete_sem: HANDLE,
is_recursive: bool,
}
struct ReadDirectoryRequest {
event_tx: Arc<Mutex<EventTx>>,
buffer: [u8; BUF_SIZE as usize],
handle: HANDLE,
data: ReadData,
}
enum Action {
Watch(PathBuf, RecursiveMode),
Unwatch(PathBuf),
Stop,
}
pub enum MetaEvent {
SingleWatchComplete,
WatcherAwakened,
}
struct WatchState {
dir_handle: HANDLE,
complete_sem: HANDLE,
}
struct ReadDirectoryChangesServer {
rx: Receiver<Action>,
event_tx: Arc<Mutex<EventTx>>,
meta_tx: Sender<MetaEvent>,
cmd_tx: Sender<Result<PathBuf>>,
watches: HashMap<PathBuf, WatchState>,
wakeup_sem: HANDLE,
}
impl ReadDirectoryChangesServer {
fn start(
event_tx: EventTx,
meta_tx: Sender<MetaEvent>,
cmd_tx: Sender<Result<PathBuf>>,
wakeup_sem: HANDLE,
) -> Sender<Action> {
let (action_tx, action_rx) = channel();
let sem_temp = wakeup_sem as u64;
thread::spawn(move || {
let wakeup_sem = sem_temp as HANDLE;
let server = ReadDirectoryChangesServer {
rx: action_rx,
event_tx: Arc::new(Mutex::new(event_tx)),
meta_tx: meta_tx,
cmd_tx: cmd_tx,
watches: HashMap::new(),
wakeup_sem: wakeup_sem,
};
server.run();
});
action_tx
}
fn run(mut self) {
loop {
let mut stopped = false;
while let Ok(action) = self.rx.try_recv() {
match action {
Action::Watch(path, recursive_mode) => {
let res = self.add_watch(path, recursive_mode.is_recursive());
let _ = self.cmd_tx.send(res);
}
Action::Unwatch(path) => self.remove_watch(path),
Action::Stop => {
stopped = true;
for (_, ws) in &self.watches {
stop_watch(ws, &self.meta_tx);
}
break;
}
}
}
if stopped {
break;
}
unsafe {
let waitres = kernel32::WaitForSingleObjectEx(self.wakeup_sem, 100, TRUE);
if waitres == WAIT_OBJECT_0 {
let _ = self.meta_tx.send(MetaEvent::WatcherAwakened);
}
}
}
unsafe {
kernel32::CloseHandle(self.wakeup_sem);
}
}
fn add_watch(&mut self, path: PathBuf, is_recursive: bool) -> Result<PathBuf> {
if !path.is_dir() && !path.is_file() {
return Err(Error::Generic(
"Input watch path is neither a file nor a directory.".to_owned(),
));
}
let (watching_file, dir_target) = {
if path.is_dir() {
(false, path.clone())
} else {
(true, path.parent().unwrap().to_path_buf())
}
};
let encoded_path: Vec<u16> = dir_target
.as_os_str()
.encode_wide()
.chain(Some(0))
.collect();
let handle;
unsafe {
handle = kernel32::CreateFileW(
encoded_path.as_ptr(),
winnt::FILE_LIST_DIRECTORY,
winnt::FILE_SHARE_READ | winnt::FILE_SHARE_DELETE | winnt::FILE_SHARE_WRITE,
ptr::null_mut(),
fileapi::OPEN_EXISTING,
winbase::FILE_FLAG_BACKUP_SEMANTICS | winbase::FILE_FLAG_OVERLAPPED,
ptr::null_mut(),
);
if handle == INVALID_HANDLE_VALUE {
let err = if watching_file {
Err(Error::Generic(
"You attempted to watch a single file, but parent \
directory could not be opened."
.to_owned(),
))
} else {
Err(Error::PathNotFound)
};
return err;
}
}
let wf = if watching_file {
Some(path.clone())
} else {
None
};
let semaphore =
unsafe { kernel32::CreateSemaphoreW(ptr::null_mut(), 0, 1, ptr::null_mut()) };
if semaphore == ptr::null_mut() || semaphore == INVALID_HANDLE_VALUE {
unsafe {
kernel32::CloseHandle(handle);
}
return Err(Error::Generic(
"Failed to create semaphore for watch.".to_owned(),
));
}
let rd = ReadData {
dir: dir_target,
file: wf,
complete_sem: semaphore,
is_recursive: is_recursive,
};
let ws = WatchState {
dir_handle: handle,
complete_sem: semaphore,
};
self.watches.insert(path.clone(), ws);
start_read(&rd, self.event_tx.clone(), handle);
Ok(path.to_path_buf())
}
fn remove_watch(&mut self, path: PathBuf) {
if let Some(ws) = self.watches.remove(&path) {
stop_watch(&ws, &self.meta_tx);
}
}
}
fn stop_watch(ws: &WatchState, meta_tx: &Sender<MetaEvent>) {
unsafe {
let cio = kernel32::CancelIo(ws.dir_handle);
let ch = kernel32::CloseHandle(ws.dir_handle);
if cio != 0 && ch != 0 {
kernel32::WaitForSingleObjectEx(ws.complete_sem, INFINITE, TRUE);
}
kernel32::CloseHandle(ws.complete_sem);
}
let _ = meta_tx.send(MetaEvent::SingleWatchComplete);
}
fn start_read(rd: &ReadData, event_tx: Arc<Mutex<EventTx>>, handle: HANDLE) {
let mut request = Box::new(ReadDirectoryRequest {
event_tx: event_tx,
handle: handle,
buffer: [0u8; BUF_SIZE as usize],
data: rd.clone(),
});
let flags = winnt::FILE_NOTIFY_CHANGE_FILE_NAME
| winnt::FILE_NOTIFY_CHANGE_DIR_NAME
| winnt::FILE_NOTIFY_CHANGE_ATTRIBUTES
| winnt::FILE_NOTIFY_CHANGE_SIZE
| winnt::FILE_NOTIFY_CHANGE_LAST_WRITE
| winnt::FILE_NOTIFY_CHANGE_CREATION
| winnt::FILE_NOTIFY_CHANGE_SECURITY;
let monitor_subdir = if (&request.data.file).is_none() && request.data.is_recursive {
1
} else {
0
};
unsafe {
let mut overlapped: Box<OVERLAPPED> = Box::new(mem::zeroed());
let req_buf = request.buffer.as_mut_ptr() as *mut c_void;
let request_p = Box::into_raw(request) as *mut c_void;
overlapped.hEvent = request_p;
let ret = winbase::ReadDirectoryChangesW(
handle,
req_buf,
BUF_SIZE,
monitor_subdir,
flags,
&mut 0u32 as *mut u32, &mut *overlapped as *mut OVERLAPPED,
Some(handle_event),
);
if ret == 0 {
let request: Box<ReadDirectoryRequest> = mem::transmute(request_p);
kernel32::ReleaseSemaphore(request.data.complete_sem, 1, ptr::null_mut());
} else {
mem::forget(overlapped);
}
}
}
fn send_pending_rename_event(event: Option<RawEvent>, event_tx: &mut EventTx) {
if let Some(e) = event {
event_tx.send(RawEvent {
path: e.path,
op: Ok(op::Op::REMOVE),
cookie: None,
});
}
}
unsafe extern "system" fn handle_event(
error_code: u32,
_bytes_written: u32,
overlapped: LPOVERLAPPED,
) {
let overlapped: Box<OVERLAPPED> = Box::from_raw(overlapped);
let request: Box<ReadDirectoryRequest> = Box::from_raw(overlapped.hEvent as *mut _);
if error_code == ERROR_OPERATION_ABORTED {
kernel32::ReleaseSemaphore(request.data.complete_sem, 1, ptr::null_mut());
return;
}
start_read(&request.data, request.event_tx.clone(), request.handle);
let event_tx_lock = request.event_tx.lock();
if let Ok(mut event_tx) = event_tx_lock {
let mut rename_event = None;
let mut cur_offset: *const u8 = request.buffer.as_ptr();
let mut cur_entry: *const FILE_NOTIFY_INFORMATION = mem::transmute(cur_offset);
loop {
let len = (*cur_entry).FileNameLength as usize / 2;
let encoded_path: &[u16] = slice::from_raw_parts((*cur_entry).FileName.as_ptr(), len);
let path = request
.data
.dir
.join(PathBuf::from(OsString::from_wide(encoded_path)));
let skip = match request.data.file {
None => false,
Some(ref watch_path) => *watch_path != path,
};
if !skip {
if (*cur_entry).Action == winnt::FILE_ACTION_RENAMED_OLD_NAME {
send_pending_rename_event(rename_event, &mut event_tx);
if request.data.file.is_some() {
event_tx.send(RawEvent {
path: Some(path),
op: Ok(op::Op::RENAME),
cookie: None,
});
rename_event = None;
} else {
COOKIE_COUNTER = COOKIE_COUNTER.wrapping_add(1);
rename_event = Some(RawEvent {
path: Some(path),
op: Ok(op::Op::RENAME),
cookie: Some(COOKIE_COUNTER),
});
}
} else {
let mut o = Op::empty();
let mut c = None;
match (*cur_entry).Action {
winnt::FILE_ACTION_RENAMED_NEW_NAME => {
if let Some(e) = rename_event {
if let Some(cookie) = e.cookie {
event_tx.send(e);
o.insert(op::Op::RENAME);
c = Some(cookie);
} else {
o.insert(op::Op::CREATE);
}
} else {
o.insert(op::Op::CREATE);
}
rename_event = None;
}
winnt::FILE_ACTION_ADDED => o.insert(op::Op::CREATE),
winnt::FILE_ACTION_REMOVED => o.insert(op::Op::REMOVE),
winnt::FILE_ACTION_MODIFIED => o.insert(op::Op::WRITE),
_ => (),
};
send_pending_rename_event(rename_event, &mut event_tx);
rename_event = None;
event_tx.send(RawEvent {
path: Some(path),
op: Ok(o),
cookie: c,
});
}
}
if (*cur_entry).NextEntryOffset == 0 {
break;
}
cur_offset = cur_offset.offset((*cur_entry).NextEntryOffset as isize);
cur_entry = mem::transmute(cur_offset);
}
send_pending_rename_event(rename_event, &mut event_tx);
}
}
pub struct ReadDirectoryChangesWatcher {
tx: Sender<Action>,
cmd_rx: Receiver<Result<PathBuf>>,
wakeup_sem: HANDLE,
}
impl ReadDirectoryChangesWatcher {
pub fn create(
tx: Sender<RawEvent>,
meta_tx: Sender<MetaEvent>,
) -> Result<ReadDirectoryChangesWatcher> {
let (cmd_tx, cmd_rx) = channel();
let wakeup_sem =
unsafe { kernel32::CreateSemaphoreW(ptr::null_mut(), 0, 1, ptr::null_mut()) };
if wakeup_sem == ptr::null_mut() || wakeup_sem == INVALID_HANDLE_VALUE {
return Err(Error::Generic(
"Failed to create wakeup semaphore.".to_owned(),
));
}
let event_tx = EventTx::Raw { tx: tx };
let action_tx = ReadDirectoryChangesServer::start(event_tx, meta_tx, cmd_tx, wakeup_sem);
Ok(ReadDirectoryChangesWatcher {
tx: action_tx,
cmd_rx: cmd_rx,
wakeup_sem: wakeup_sem,
})
}
pub fn create_debounced(
tx: Sender<DebouncedEvent>,
meta_tx: Sender<MetaEvent>,
delay: Duration,
) -> Result<ReadDirectoryChangesWatcher> {
let (cmd_tx, cmd_rx) = channel();
let wakeup_sem =
unsafe { kernel32::CreateSemaphoreW(ptr::null_mut(), 0, 1, ptr::null_mut()) };
if wakeup_sem == ptr::null_mut() || wakeup_sem == INVALID_HANDLE_VALUE {
return Err(Error::Generic(
"Failed to create wakeup semaphore.".to_owned(),
));
}
let event_tx = EventTx::Debounced {
tx: tx.clone(),
debounce: Debounce::new(delay, tx),
};
let action_tx = ReadDirectoryChangesServer::start(event_tx, meta_tx, cmd_tx, wakeup_sem);
Ok(ReadDirectoryChangesWatcher {
tx: action_tx,
cmd_rx: cmd_rx,
wakeup_sem: wakeup_sem,
})
}
fn wakeup_server(&mut self) {
unsafe {
kernel32::ReleaseSemaphore(self.wakeup_sem, 1, ptr::null_mut());
}
}
fn send_action_require_ack(&mut self, action: Action, pb: &PathBuf) -> Result<()> {
match self.tx.send(action) {
Err(_) => Err(Error::Generic(
"Error sending to internal channel".to_owned(),
)),
Ok(_) => {
self.wakeup_server();
match self.cmd_rx.recv() {
Err(_) => Err(Error::Generic(
"Error receiving from command channel".to_owned(),
)),
Ok(ack_res) => match ack_res {
Err(e) => Err(Error::Generic(format!("Error in watcher: {:?}", e))),
Ok(ack_pb) => {
if pb.as_path() != ack_pb.as_path() {
Err(Error::Generic(format!(
"Expected ack for {:?} but got \
ack for {:?}",
pb, ack_pb
)))
} else {
Ok(())
}
}
},
}
}
}
}
}
impl Watcher for ReadDirectoryChangesWatcher {
fn new_raw(tx: Sender<RawEvent>) -> Result<ReadDirectoryChangesWatcher> {
let (meta_tx, _) = channel();
ReadDirectoryChangesWatcher::create(tx, meta_tx)
}
fn new(tx: Sender<DebouncedEvent>, delay: Duration) -> Result<ReadDirectoryChangesWatcher> {
let (meta_tx, _) = channel();
ReadDirectoryChangesWatcher::create_debounced(tx, meta_tx, delay)
}
fn watch<P: AsRef<Path>>(&mut self, path: P, recursive_mode: RecursiveMode) -> Result<()> {
let pb = if path.as_ref().is_absolute() {
path.as_ref().to_owned()
} else {
let p = try!(env::current_dir().map_err(Error::Io));
p.join(path)
};
if !pb.is_dir() && !pb.is_file() {
return Err(Error::Generic(
"Input watch path is neither a file nor a directory.".to_owned(),
));
}
self.send_action_require_ack(Action::Watch(pb.clone(), recursive_mode), &pb)
}
fn unwatch<P: AsRef<Path>>(&mut self, path: P) -> Result<()> {
let pb = if path.as_ref().is_absolute() {
path.as_ref().to_owned()
} else {
let p = try!(env::current_dir().map_err(Error::Io));
p.join(path)
};
let res = self
.tx
.send(Action::Unwatch(pb))
.map_err(|_| Error::Generic("Error sending to internal channel".to_owned()));
self.wakeup_server();
res
}
}
impl Drop for ReadDirectoryChangesWatcher {
fn drop(&mut self) {
let _ = self.tx.send(Action::Stop);
self.wakeup_server();
}
}
unsafe impl Send for ReadDirectoryChangesWatcher {}
unsafe impl Sync for ReadDirectoryChangesWatcher {}