command_stream/command_packages/
watch.rs

1//! In-process manifest watching; successful reloads replace commands atomically.
2use super::CommandEnvironment;
3use std::io;
4use std::path::Path;
5use std::sync::Arc;
6use std::time::Duration;
7use tokio::sync::{mpsc, watch, RwLock};
8use tokio::task::JoinHandle;
9
10pub struct PackageWatcher {
11    stop: watch::Sender<bool>,
12    task: Option<JoinHandle<()>>,
13    events: mpsc::Receiver<Result<String, String>>,
14}
15
16impl PackageWatcher {
17    pub async fn next_event(&mut self) -> Option<Result<String, String>> {
18        self.events.recv().await
19    }
20
21    pub async fn close(mut self) -> io::Result<()> {
22        let _ = self.stop.send(true);
23        self.task.take().unwrap().await.map_err(io::Error::other)
24    }
25}
26
27impl Drop for PackageWatcher {
28    fn drop(&mut self) {
29        let _ = self.stop.send(true);
30    }
31}
32
33/// Install a manifest and start watching its canonical path before returning.
34/// Atomic editor saves are recommended. Invalid edits retain the last working
35/// package and produce an error event. Close waits for outstanding reload work.
36pub async fn watch_package(
37    environment: Arc<RwLock<CommandEnvironment>>,
38    path: impl AsRef<Path>,
39    interval: Duration,
40) -> io::Result<PackageWatcher> {
41    if interval < Duration::from_millis(10) {
42        return Err(io::Error::new(
43            io::ErrorKind::InvalidInput,
44            "Watch interval must be at least 10ms",
45        ));
46    }
47    let path = tokio::fs::canonicalize(path).await?;
48    let mut previous = tokio::fs::read(&path).await?;
49    environment
50        .write()
51        .await
52        .install_file(&path)
53        .await
54        .map_err(io::Error::other)?;
55    let (stop, mut receiver) = watch::channel(false);
56    let (tx, events) = mpsc::channel(32);
57    let task = tokio::spawn(async move {
58        let mut tick = tokio::time::interval(interval);
59        tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
60        loop {
61            tokio::select! {
62                _ = receiver.changed() => break,
63                _ = tick.tick() => {
64                    match tokio::fs::read(&path).await {
65                        Ok(next) if next != previous => {
66                            previous = next;
67                            let result = environment.write().await.reload_file(&path).await;
68                            let _ = tx.try_send(result);
69                        }
70                        Ok(_) => {},
71                        Err(error) => { let _ = tx.try_send(Err(error.to_string())); },
72                    }
73                }
74            }
75        }
76    });
77    Ok(PackageWatcher {
78        stop,
79        task: Some(task),
80        events,
81    })
82}
command_stream/command_packages/watch.rs

command_stream/command_packages/
watch.rs
