clankerdiff_watch/
repository_watcher.rs

1use crate::{
2    error::WatchError,
3    file_watcher::{FileWatchError, FileWatcher, NotifyFileWatcher},
4    filter::should_refresh,
5};
6use clankerdiff_core::DiffScope;
7use clankerdiff_git::{GitError, GitRepository, RepositorySnapshot};
8use std::{sync::Arc, time::Duration};
9use tokio::{
10    sync::{mpsc, oneshot, watch},
11    task::JoinHandle,
12};
13
14#[derive(Debug, Clone, Copy, PartialEq, Eq)]
15pub struct WatchOptions {
16    pub debounce: Duration,
17}
18
19impl Default for WatchOptions {
20    fn default() -> Self {
21        Self {
22            debounce: Duration::from_millis(150),
23        }
24    }
25}
26
27#[derive(Debug)]
28pub enum RepositoryRequest {
29    SetScope {
30        scope: DiffScope,
31        result_tx: oneshot::Sender<Result<(), Arc<GitError>>>,
32    },
33}
34
35#[derive(Debug)]
36pub struct RepositoryWatcher {
37    pub request_tx: mpsc::Sender<RepositoryRequest>,
38    pub snapshot_rx: watch::Receiver<Result<Arc<RepositorySnapshot>, Arc<GitError>>>,
39    task: JoinHandle<()>,
40}
41
42impl RepositoryWatcher {
43    pub async fn spawn(
44        repository: GitRepository,
45        scope: DiffScope,
46        options: WatchOptions,
47    ) -> Result<Self, WatchError> {
48        let watcher = Self::create_file_watcher(&repository, options.debounce).await?;
49        let snapshot = repository.snapshot_with_sources(scope).await?;
50        let (request_tx, request_rx) = mpsc::channel(64);
51        let (state_tx, state_rx) = watch::channel(Ok(Arc::new(snapshot)));
52        let actor = RepositoryActor {
53            repository,
54            watcher,
55            scope,
56            request_rx,
57            state_tx,
58        };
59        Ok(Self {
60            request_tx,
61            snapshot_rx: state_rx,
62            task: tokio::spawn(actor.run()),
63        })
64    }
65
66    async fn create_file_watcher(
67        repository: &GitRepository,
68        debounce: Duration,
69    ) -> Result<NotifyFileWatcher, WatchError> {
70        let root = repository.root().to_path_buf();
71        let directories = repository.metadata_directories().await?;
72        let mut roots = vec![root.clone()];
73        roots.extend(directories.iter().cloned());
74        let filter_repository = repository.clone();
75        let watcher = NotifyFileWatcher::new(roots, debounce, move |paths| {
76            let repository = filter_repository.clone();
77            let directories = directories.clone();
78            async move { should_refresh(&repository, &directories, paths).await }
79        })
80        .map_err(|error| match error {
81            FileWatchError::Create(source) => WatchError::Watch { path: root, source },
82            FileWatchError::Watch { path, source } => WatchError::Watch { path, source },
83        })?;
84
85        Ok(watcher)
86    }
87}
88
89impl Drop for RepositoryWatcher {
90    fn drop(&mut self) {
91        self.task.abort();
92    }
93}
94
95struct RepositoryActor {
96    repository: GitRepository,
97    watcher: NotifyFileWatcher,
98    scope: DiffScope,
99    request_rx: mpsc::Receiver<RepositoryRequest>,
100    state_tx: watch::Sender<Result<Arc<RepositorySnapshot>, Arc<GitError>>>,
101}
102
103impl RepositoryActor {
104    async fn run(mut self) {
105        loop {
106            let result_tx = tokio::select! {
107                biased;
108                request = self.request_rx.recv() => match request {
109                    Some(RepositoryRequest::SetScope { scope, result_tx }) => {
110                        self.scope = scope;
111                        Some(result_tx)
112                    }
113                    None => return,
114                },
115                Some(()) = self.watcher.recv() => None,
116            };
117
118            let result = self
119                .repository
120                .snapshot_with_sources(self.scope)
121                .await
122                .map(Arc::new)
123                .map_err(Arc::new);
124
125            self.state_tx.send_if_modified(|state| {
126                let unchanged = match (&*state, &result) {
127                    (Ok(previous), Ok(snapshot)) => previous == snapshot,
128                    (Err(previous), Err(error)) => previous.to_string() == error.to_string(),
129                    _ => false,
130                };
131
132                if unchanged {
133                    return false;
134                }
135
136                state.clone_from(&result);
137                true
138            });
139
140            if let Some(tx) = result_tx {
141                let _ = tx.send(result.map(|_| ()));
142            }
143        }
144    }
145}
clankerdiff_watch/repository_watcher.rs

clankerdiff_watch/
repository_watcher.rs
