live 0.3.1

A modular configuration framework with live reloading and atomic, format-agnostic updates.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114

/* src/signal/watcher.rs */

use notify::{RecommendedWatcher, RecursiveMode, Watcher as NotifyWatcher};
use tokio::sync::{broadcast, mpsc};
use tokio::task::JoinHandle;

use super::target::CompiledTarget;
use super::worker::process_events;
use super::{Config, Event, Result, SignalError, Target};

/// The main watcher struct.
pub struct Watcher {
	_internal_watcher: RecommendedWatcher,
	task_handle: JoinHandle<()>,
	event_tx: broadcast::Sender<Event>,
}

impl Watcher {
	/// Creates a new Watcher and starts monitoring immediately.
	#[must_use = "Watcher must be kept alive"]
	pub fn new(target: Target, config: Config) -> Result<Self> {
		let (raw_tx, raw_rx) = mpsc::channel(100);

		let mut internal_watcher =
			notify::recommended_watcher(move |res: notify::Result<notify::Event>| {
				let _ = raw_tx.blocking_send(res);
			})?;

		let (watch_path, mode, root_path) = match &target {
			Target::File(p) => {
				if let Some(parent) = p.parent() {
					let parent = if parent.as_os_str().is_empty() {
						std::path::PathBuf::from(".")
					} else {
						parent.to_path_buf()
					};
					(parent.clone(), RecursiveMode::NonRecursive, parent)
				} else {
					(p.clone(), RecursiveMode::NonRecursive, p.clone())
				}
			}
			Target::Directory(p) | Target::Filtered { path: p, .. } => {
				(p.clone(), RecursiveMode::Recursive, p.clone())
			}
		};

		if !watch_path.exists() {
			return Err(SignalError::Config(format!(
				"Path does not exist: {:?}",
				watch_path
			)));
		}

		let compiled_target = CompiledTarget::new(target)?;

		internal_watcher.watch(&watch_path, mode)?;

		// Canonicalize root path to ensure strip_prefix works with absolute paths from notify
		let root_path = std::fs::canonicalize(&root_path).unwrap_or(root_path);

		let (user_tx, _) = broadcast::channel(100);
		let tx_clone = user_tx.clone();

		let task_handle = tokio::spawn(async move {
			process_events(raw_rx, tx_clone, compiled_target, config, root_path).await;
		});

		Ok(Self {
			_internal_watcher: internal_watcher,
			task_handle,
			event_tx: user_tx,
		})
	}

	pub fn subscribe(&self) -> broadcast::Receiver<Event> {
		self.event_tx.subscribe()
	}

	pub fn stop(&self) {
		self.task_handle.abort();
	}
}

impl Drop for Watcher {
	fn drop(&mut self) {
		self.task_handle.abort();
	}
}

#[cfg(feature = "stream")]
pub struct EventStream {
	inner: tokio_stream::wrappers::BroadcastStream<Event>,
}

#[cfg(feature = "stream")]
impl futures_util::Stream for EventStream {
	type Item = std::result::Result<Event, tokio_stream::wrappers::errors::BroadcastStreamRecvError>;

	fn poll_next(
		mut self: std::pin::Pin<&mut Self>,
		cx: &mut std::task::Context<'_>,
	) -> std::task::Poll<Option<Self::Item>> {
		std::pin::Pin::new(&mut self.inner).poll_next(cx)
	}
}

#[cfg(feature = "stream")]
impl Watcher {
	pub fn stream(&self) -> EventStream {
		EventStream {
			inner: tokio_stream::wrappers::BroadcastStream::new(self.subscribe()),
		}
	}
}