coven_database/
database_changes.rs

1//! Commit notifications shared by file work and synchronization.
2use crate::DbError;
3
4/// A live commit observation; coalesces changes and ends when the writer closes.
5pub struct DatabaseChanges {
6    commits: crate::observation::CommitSubscription,
7    reads: crate::observation::ReadSet,
8    first: bool,
9}
10impl DatabaseChanges {
11    /// What `commits`, waiting on `reads`, reports; `first` reports the
12    /// state now as their first change.
13    pub(crate) fn new(
14        commits: crate::observation::CommitSubscription,
15        reads: crate::observation::ReadSet,
16        first: bool,
17    ) -> Self {
18        Self { commits, reads, first }
19    }
20
21    /// Whether a relevant commit happened since the previous observation,
22    /// without waiting; the first call reports the initial state.
23    pub fn take_changed(&mut self) -> Result<bool, DbError> {
24        let state = self.commits.state();
25        if let Some(error) = state.error() {
26            return Err(error);
27        }
28        if self.first || state.changed {
29            self.first = false;
30            self.commits.begin();
31            self.commits.finish(self.reads.clone());
32            return Ok(true);
33        }
34        Ok(false)
35    }
36
37    /// Resolves once a relevant commit happened since the previous
38    /// observation, or the handle failed or closed. It owns its subscription,
39    /// so a worker can wait on it beside other wakes; `take_changed` then
40    /// consumes the change.
41    pub fn changed(&self) -> impl std::future::Future<Output = ()> + Send + 'static {
42        let mut commits = self.commits.clone();
43        let first = self.first;
44        async move {
45            loop {
46                let state = commits.state();
47                if first || state.changed || state.error().is_some() {
48                    return;
49                }
50                commits.changed().await;
51            }
52        }
53    }
54
55    /// The initial state, then each relevant commit. No timer or polling is used.
56    pub async fn next(&mut self) -> Result<(), DbError> {
57        loop {
58            let state = self.commits.state();
59            if let Some(error) = state.error() {
60                return Err(error);
61            }
62            if self.first || state.changed {
63                self.first = false;
64                self.commits.begin();
65                self.commits.finish(self.reads.clone());
66                return Ok(());
67            }
68            self.commits.changed().await;
69        }
70    }
71}
72
73#[cfg(test)]
74#[path = "database_changes_tests.rs"]
75pub(crate) mod tests;
coven_database/database_changes.rs

coven_database/
database_changes.rs
