use crate::watcher::{FileWatcher, FileWatcherConfig};
use dampen_core::ir::DampenDocument;
use dampen_core::parser;
use dampen_core::parser::error::ParseError;
use iced::Subscription;
use iced::advanced::subscription::{EventStream, Hasher, Recipe};
use std::hash::Hash;
use std::path::PathBuf;
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;
#[derive(Debug, Clone)]
pub enum FileEvent {
Success {
path: PathBuf,
document: Box<DampenDocument>,
},
ParseError {
path: PathBuf,
error: ParseError,
content: String,
},
WatcherError {
path: PathBuf,
error: String,
},
}
#[derive(Debug, Clone)]
pub struct FileWatcherRecipe {
pub paths: Vec<PathBuf>,
pub debounce_ms: u64,
pub extension_filter: String,
pub recursive: bool,
}
impl FileWatcherRecipe {
pub fn new(paths: Vec<PathBuf>, debounce_ms: u64) -> Self {
Self {
paths,
debounce_ms,
extension_filter: ".dampen".to_string(),
recursive: true,
}
}
pub fn with_extension(mut self, extension: impl Into<String>) -> Self {
self.extension_filter = extension.into();
self
}
pub fn with_recursive(mut self, recursive: bool) -> Self {
self.recursive = recursive;
self
}
}
impl Recipe for FileWatcherRecipe {
type Output = FileEvent;
fn hash(&self, state: &mut Hasher) {
std::any::TypeId::of::<Self>().hash(state);
let mut sorted_paths = self.paths.clone();
sorted_paths.sort();
for path in &sorted_paths {
path.hash(state);
}
self.debounce_ms.hash(state);
self.extension_filter.hash(state);
self.recursive.hash(state);
}
fn stream(
self: Box<Self>,
_input: EventStream,
) -> futures::stream::BoxStream<'static, Self::Output> {
let paths = self.paths;
let debounce_ms = self.debounce_ms;
let extension_filter = self.extension_filter;
let recursive = self.recursive;
let (tx, rx) = mpsc::channel(1000);
let tx_monitor = tx.clone();
tokio::spawn(async move {
let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(30));
loop {
interval.tick().await;
let capacity = tx_monitor.max_capacity();
let available = tx_monitor.capacity();
let fill_percent = ((capacity - available) as f64 / capacity as f64) * 100.0;
if fill_percent > 80.0 {
eprintln!(
"[dampen-dev] Warning: File event channel {:.0}% full ({} of {} slots used)",
fill_percent,
capacity - available,
capacity
);
}
}
});
tokio::task::spawn_blocking(move || {
let config = FileWatcherConfig {
watch_paths: paths.clone(),
debounce_ms,
extension_filter,
recursive,
};
eprintln!(
"[dampen-dev] Creating file watcher with config: paths={:?}, debounce={}ms",
paths, debounce_ms
);
let mut watcher = match FileWatcher::new(config) {
Ok(w) => {
eprintln!("[dampen-dev] File watcher created successfully");
w
}
Err(e) => {
eprintln!("[dampen-dev] Failed to create file watcher: {}", e);
let _ = tx.blocking_send(FileEvent::WatcherError {
path: PathBuf::new(),
error: format!("Failed to create file watcher: {}", e),
});
return;
}
};
for path in &paths {
eprintln!("[dampen-dev] Attempting to watch: {}", path.display());
if let Err(e) = watcher.watch(path.clone()) {
eprintln!("[dampen-dev] Failed to watch {}: {}", path.display(), e);
let _ = tx.blocking_send(FileEvent::WatcherError {
path: path.clone(),
error: format!("Failed to watch path: {}", e),
});
} else {
eprintln!("[dampen-dev] Successfully watching: {}", path.display());
}
}
eprintln!("[dampen-dev] File watcher ready, waiting for events...");
let receiver = watcher.receiver();
loop {
match receiver.recv_timeout(std::time::Duration::from_millis(100)) {
Ok(path) => {
eprintln!("[dampen-dev] File changed: {}", path.display());
let content = match std::fs::read_to_string(&path) {
Ok(c) => c,
Err(e) => {
if tx
.blocking_send(FileEvent::WatcherError {
path: path.clone(),
error: format!("Failed to read file: {}", e),
})
.is_err()
{
eprintln!("[dampen-dev] Channel closed, stopping file watcher");
break;
}
continue;
}
};
let event = match parser::parse(&content) {
Ok(document) => {
FileEvent::Success {
path: path.clone(),
document: Box::new(document),
}
}
Err(error) => {
FileEvent::ParseError {
path: path.clone(),
error,
content,
}
}
};
if tx.blocking_send(event).is_err() {
eprintln!("[dampen-dev] Channel closed, stopping file watcher");
break;
}
}
Err(crossbeam_channel::RecvTimeoutError::Timeout) => {
if tx.is_closed() {
eprintln!("[dampen-dev] Channel closed, stopping file watcher");
break;
}
}
Err(crossbeam_channel::RecvTimeoutError::Disconnected) => {
eprintln!("[dampen-dev] File watcher disconnected");
break;
}
}
}
eprintln!("[dampen-dev] File watcher task exiting gracefully");
});
Box::pin(ReceiverStream::new(rx))
}
}
pub fn watch_files<P: AsRef<std::path::Path>>(
paths: Vec<P>,
debounce_ms: u64,
) -> Subscription<FileEvent> {
let path_bufs: Vec<PathBuf> = paths.iter().map(|p| p.as_ref().to_path_buf()).collect();
let recipe = FileWatcherRecipe::new(path_bufs, debounce_ms);
use iced::advanced::subscription::from_recipe;
from_recipe(recipe)
}
pub fn watch_system_theme() -> Subscription<String> {
dampen_iced::watch_system_theme()
}