olai_stack_state/
watch.rs

1//! [`Watcher`]: notice writes other processes made to the store.
2
3use std::time::Duration;
4
5use futures::Stream;
6use rusqlite::Connection;
7
8use crate::error::Result;
9use crate::store::{Store, connect};
10
11/// Notices commits to the store made through any other connection, in this process or
12/// another (the CLI writing while the app is open, say). It holds its own connection and
13/// compares SQLite's `PRAGMA data_version`, which changes exactly when someone else commits.
14/// It says *that* something changed, not what: a consumer re-reads what it shows.
15pub struct Watcher {
16    conn: Connection,
17    version: i64,
18}
19
20impl Store {
21    /// A [`Watcher`] for this store's file.
22    pub fn watcher(&self) -> Result<Watcher> {
23        let conn = connect(self.path())?;
24        let version = data_version(&conn)?;
25        Ok(Watcher { conn, version })
26    }
27}
28
29fn data_version(conn: &Connection) -> Result<i64> {
30    Ok(conn.query_row("PRAGMA data_version", [], |r| r.get(0))?)
31}
32
33impl Watcher {
34    /// Whether another connection committed since the last call (or since the watcher was
35    /// made).
36    pub fn changed(&mut self) -> Result<bool> {
37        let version = data_version(&self.conn)?;
38        let changed = version != self.version;
39        self.version = version;
40        Ok(changed)
41    }
42
43    /// A stream yielding once per poll (every `interval`) that saw a change. Needs a Tokio
44    /// runtime with the timer enabled.
45    pub fn changes(self, interval: Duration) -> impl Stream<Item = Result<()>> {
46        futures::stream::unfold(self, move |mut watcher| async move {
47            loop {
48                tokio::time::sleep(interval).await;
49                match watcher.changed() {
50                    Ok(true) => return Some((Ok(()), watcher)),
51                    Ok(false) => continue,
52                    Err(e) => return Some((Err(e), watcher)),
53                }
54            }
55        })
56    }
57}
olai_stack_state/watch.rs

olai_stack_state/
watch.rs
