command_stream/development/
watch.rs

1use super::child::ManagedChild;
2use crate::trace::trace_lazy;
3use glob::{MatchOptions, Pattern};
4use std::collections::{BTreeMap, BTreeSet};
5use std::io;
6use std::path::{Path, PathBuf};
7use std::process::Stdio;
8use std::time::{Duration, Instant, SystemTime};
9use tokio::process::Command;
10use tokio::sync::{mpsc, watch};
11use tokio::task::JoinHandle;
12
13/// Watch glob patterns and restart Cargo, or another exact-argv command.
14#[derive(Debug, Clone)]
15pub struct DevOptions {
16    pub cwd: PathBuf,
17    pub watch: Vec<String>,
18    /// Exact argv. Defaults to `cargo run`; use `--manifest-path` for another crate.
19    pub command: Vec<String>,
20    pub initial: bool,
21    pub interval: Duration,
22    pub debounce: Duration,
23}
24
25impl Default for DevOptions {
26    fn default() -> Self {
27        Self {
28            cwd: std::env::current_dir().unwrap_or_else(|_| ".".into()),
29            watch: vec![
30                "**/*.rs".into(),
31                "**/Cargo.toml".into(),
32                "**/Cargo.lock".into(),
33            ],
34            command: vec!["cargo".into(), "run".into()],
35            initial: true,
36            interval: Duration::from_millis(100),
37            debounce: Duration::from_millis(50),
38        }
39    }
40}
41
42#[derive(Debug, Clone)]
43pub enum DevEvent {
44    Restarted {
45        /// Absolute paths under the canonical watch root; empty for the initial run.
46        files: Vec<PathBuf>,
47        pid: Option<u32>,
48    },
49    Error(String),
50}
51
52/// Explicit shutdown waits for the current child and stops its descendants.
53/// Dropping the controller also requests shutdown.
54pub struct DevController {
55    stop: watch::Sender<bool>,
56    task: Option<JoinHandle<io::Result<()>>>,
57    events: mpsc::Receiver<DevEvent>,
58}
59
60impl DevController {
61    pub async fn next_event(&mut self) -> Option<DevEvent> {
62        self.events.recv().await
63    }
64    pub async fn close(mut self) -> io::Result<()> {
65        let _ = self.stop.send(true);
66        self.task.take().unwrap().await.map_err(io::Error::other)?
67    }
68}
69impl Drop for DevController {
70    fn drop(&mut self) {
71        let _ = self.stop.send(true);
72    }
73}
74
75type Snapshot = BTreeMap<PathBuf, (SystemTime, u64)>;
76fn snapshot(root: &Path, patterns: &[Pattern]) -> io::Result<Snapshot> {
77    let mut files = BTreeMap::new();
78    let mut directories = vec![root.to_owned()];
79    let options = MatchOptions {
80        require_literal_separator: true,
81        ..MatchOptions::new()
82    };
83    while let Some(dir) = directories.pop() {
84        for entry in std::fs::read_dir(dir)? {
85            let entry = entry?;
86            let name = entry.file_name();
87            if ["target", ".git", "node_modules"]
88                .iter()
89                .any(|ignored| name == *ignored)
90                || name.to_string_lossy().starts_with(".command-stream-")
91            {
92                continue;
93            }
94            let path = entry.path();
95            let file_type = entry.file_type()?;
96            if file_type.is_dir() {
97                directories.push(path);
98            } else if file_type.is_file()
99                && patterns.iter().any(|pattern| {
100                    pattern.matches_path_with(path.strip_prefix(root).unwrap(), options)
101                })
102            {
103                match entry.metadata() {
104                    Ok(info) => {
105                        files.insert(path, (info.modified()?, info.len()));
106                    }
107                    Err(error) if error.kind() == io::ErrorKind::NotFound => {}
108                    Err(error) => return Err(error),
109                }
110            }
111        }
112    }
113    Ok(files)
114}
115
116fn spawn(options: &DevOptions) -> io::Result<ManagedChild> {
117    let mut command = Command::new(&options.command[0]);
118    command
119        .args(&options.command[1..])
120        .current_dir(&options.cwd)
121        .env("COMMAND_STREAM_DEV_CHILD", "1")
122        .stdin(Stdio::null())
123        .stdout(Stdio::inherit())
124        .stderr(Stdio::inherit());
125    ManagedChild::spawn(&mut command, false)
126}
127
128/// Start watching after validating the root, patterns and executable.
129/// Creation, deletion and edits all trigger restarts; build output is excluded.
130pub async fn dev(mut options: DevOptions) -> io::Result<DevController> {
131    if options.interval < Duration::from_millis(10) || options.command.is_empty() {
132        return Err(io::Error::new(
133            io::ErrorKind::InvalidInput,
134            "Invalid watch interval or empty command",
135        ));
136    }
137    options.cwd = options.cwd.canonicalize()?;
138    let patterns = options
139        .watch
140        .iter()
141        .map(|pattern| Pattern::new(pattern.trim_start_matches("./")))
142        .collect::<std::result::Result<Vec<_>, _>>()
143        .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error))?;
144    let previous = snapshot(&options.cwd, &patterns)?;
145    let (stop, receiver) = watch::channel(false);
146    let (tx, events) = mpsc::channel(32);
147    let child_mode = std::env::var_os("COMMAND_STREAM_DEV_CHILD").is_some();
148    let child = if options.initial && !child_mode {
149        Some(spawn(&options)?)
150    } else {
151        None
152    };
153    if let Some(child) = &child {
154        let _ = tx.try_send(DevEvent::Restarted {
155            files: vec![],
156            pid: child.child.id(),
157        });
158    }
159    let task = tokio::spawn(async move {
160        if child_mode {
161            return Ok(());
162        }
163        watch_loop(options, patterns, previous, child, receiver, tx).await
164    });
165    Ok(DevController {
166        stop,
167        task: Some(task),
168        events,
169    })
170}
171
172async fn watch_loop(
173    options: DevOptions,
174    patterns: Vec<Pattern>,
175    mut previous: Snapshot,
176    mut child: Option<ManagedChild>,
177    mut stop: watch::Receiver<bool>,
178    events: mpsc::Sender<DevEvent>,
179) -> io::Result<()> {
180    let mut tick = tokio::time::interval(options.interval);
181    tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
182    let mut changes = BTreeSet::new();
183    let mut changed_at = Instant::now();
184    loop {
185        tokio::select! {
186            _ = stop.changed() => break,
187            _ = tick.tick() => {
188                let next = match snapshot(&options.cwd, &patterns) {
189                    Ok(next) => next,
190                    Err(error) => { let _ = events.try_send(DevEvent::Error(error.to_string())); continue; }
191                };
192                for file in previous.keys().chain(next.keys()) {
193                    if previous.get(file) != next.get(file) {
194                        changes.insert(file.clone());
195                        changed_at = Instant::now();
196                    }
197                }
198                previous = next;
199                if !changes.is_empty() && changed_at.elapsed() >= options.debounce {
200                    if let Some(mut current) = child.take() { current.stop().await?; }
201                    if *stop.borrow() { break; }
202                    let files = std::mem::take(&mut changes).into_iter().collect::<Vec<_>>();
203                    trace_lazy("Development", || format!("Restarting after changes: {files:?}"));
204                    match spawn(&options) {
205                        Ok(next) => {
206                            let _ = events.try_send(DevEvent::Restarted { files, pid: next.child.id() });
207                            child = Some(next);
208                        }
209                        Err(error) => { let _ = events.try_send(DevEvent::Error(error.to_string())); }
210                    }
211                }
212            }
213        }
214    }
215    if let Some(mut child) = child {
216        child.stop().await?;
217    }
218    Ok(())
219}
command_stream/development/watch.rs

command_stream/development/
watch.rs
