spacedb_crdt/
reactive.rs

1//! Reactive queries — re-evaluate as the document changes.
2//!
3//! A [`CrdtDoc`] bumps a revision counter on every change (local mutation *or*
4//! applied remote update). [`Watcher`] observes that counter; [`ReactiveQuery`]
5//! turns it into a live query that re-evaluates when the document changes and
6//! **emits a result only when the result actually changes** — the "re-render as
7//! the mesh converges" behaviour the SDK exposes as a signal/stream.
8//!
9//! Polling, not callbacks: the yrs update observer fires *inside* the mutating
10//! transaction, where opening another transaction (to read fields) would panic.
11//! So changes are recorded as a counter bump and the query re-evaluates safely
12//! afterwards, when the caller polls.
13
14use std::cell::Cell;
15use std::sync::atomic::{AtomicU64, Ordering};
16use std::sync::Arc;
17
18use crate::CrdtDoc;
19
20/// Watches a document's revision counter for changes.
21pub struct Watcher {
22    revision: Arc<AtomicU64>,
23    last_seen: Cell<u64>,
24}
25
26impl Watcher {
27    pub(crate) fn new(revision: Arc<AtomicU64>) -> Self {
28        let current = revision.load(Ordering::Relaxed);
29        Self {
30            revision,
31            last_seen: Cell::new(current),
32        }
33    }
34
35    /// Return whether the document has changed since the last call, consuming the
36    /// pending change so a subsequent call returns `false` until the next change.
37    pub fn drain_changed(&self) -> bool {
38        let current = self.revision.load(Ordering::Relaxed);
39        if current != self.last_seen.get() {
40            self.last_seen.set(current);
41            true
42        } else {
43            false
44        }
45    }
46
47    /// The current revision of the watched document.
48    pub fn revision(&self) -> u64 {
49        self.revision.load(Ordering::Relaxed)
50    }
51}
52
53/// A live query over a document: re-evaluates `query` when the document changes
54/// and yields the new result only when it differs from the last emitted one.
55pub struct ReactiveQuery<R, F> {
56    watcher: Watcher,
57    query: F,
58    last: R,
59}
60
61impl<R, F> ReactiveQuery<R, F>
62where
63    R: Clone + PartialEq,
64    F: Fn(&CrdtDoc) -> R,
65{
66    /// Build a reactive query over `doc`, evaluating `query` once for the initial
67    /// value.
68    pub fn new(doc: &CrdtDoc, query: F) -> Self {
69        let watcher = doc.watch();
70        let last = query(doc);
71        Self {
72            watcher,
73            query,
74            last,
75        }
76    }
77
78    /// Poll for an update. Returns `Some(new_result)` if the document changed
79    /// *and* the query result changed since the last emission (the pushed delta);
80    /// otherwise `None`.
81    pub fn poll(&mut self, doc: &CrdtDoc) -> Option<R> {
82        if !self.watcher.drain_changed() {
83            return None;
84        }
85        let next = (self.query)(doc);
86        if next != self.last {
87            self.last = next.clone();
88            Some(next)
89        } else {
90            None
91        }
92    }
93
94    /// The last evaluated result.
95    pub fn current(&self) -> &R {
96        &self.last
97    }
98}
spacedb_crdt/reactive.rs

spacedb_crdt/
reactive.rs
