zerodds_ast_core/
observer.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 ZeroDDS Contributors
3//! Change observers. Spec §5.3.
4//!
5//! zerodds-lint: allow no_dyn_in_safe — this file is the plugin
6//! boundary `docs/architecture/04_safety_by_architecture.md` §2.1
7//! carves out for `dyn`: an observer is by definition a caller-supplied
8//! implementation behind a stable trait, and a generic parameter here
9//! would monomorphise the registry per observer set and leak every
10//! observer type into `Ast`'s own signature.
11
12use crate::change::Changeset;
13use alloc::sync::Arc;
14use alloc::vec::Vec;
15
16/// Receives the changeset of every successful commit.
17///
18/// Invoked after the commit, never during it, so an observer always sees
19/// a model that satisfies the structural invariants. Spec §5.3, §8.1.
20pub trait AstObserver: Send + Sync {
21    /// React to a committed changeset.
22    fn on_commit(&self, changeset: &Changeset);
23}
24
25/// Handle for removing a registered observer.
26#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
27pub struct ObserverToken(u64);
28
29/// The set of registered observers.
30///
31/// The layout cache and the model indices are ordinary members of this
32/// set; they hold no privileged path into the model.
33#[derive(Clone, Default)]
34pub struct ObserverRegistry {
35    entries: Vec<(ObserverToken, Arc<dyn AstObserver>)>,
36    next_token: u64,
37}
38
39impl ObserverRegistry {
40    /// Empty registry.
41    #[must_use]
42    pub const fn new() -> Self {
43        Self {
44            entries: Vec::new(),
45            next_token: 0,
46        }
47    }
48
49    /// Register an observer, returning the token that removes it again.
50    pub fn register(&mut self, observer: Arc<dyn AstObserver>) -> ObserverToken {
51        let token = ObserverToken(self.next_token);
52        self.next_token = self.next_token.wrapping_add(1);
53        self.entries.push((token, observer));
54        token
55    }
56
57    /// Remove an observer. Returns whether it was registered.
58    pub fn unregister(&mut self, token: ObserverToken) -> bool {
59        let before = self.entries.len();
60        self.entries.retain(|(candidate, _)| *candidate != token);
61        self.entries.len() != before
62    }
63
64    /// Number of registered observers.
65    #[must_use]
66    pub fn len(&self) -> usize {
67        self.entries.len()
68    }
69
70    /// Whether no observer is registered.
71    #[must_use]
72    pub fn is_empty(&self) -> bool {
73        self.entries.is_empty()
74    }
75
76    /// Notify every observer, in registration order.
77    ///
78    /// Called only after a successful commit. A commit that changed
79    /// nothing still notifies with an empty changeset, so a consumer
80    /// tracking sequence numbers sees no gaps.
81    pub fn notify(&self, changeset: &Changeset) {
82        for (_, observer) in &self.entries {
83            observer.on_commit(changeset);
84        }
85    }
86}
87
88impl core::fmt::Debug for ObserverRegistry {
89    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
90        f.debug_struct("ObserverRegistry")
91            .field("len", &self.entries.len())
92            .finish()
93    }
94}
95
96#[cfg(test)]
97mod tests {
98    // Tests assert on Results directly; a failed expect IS the test failure.
99    #![allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)]
100    use super::*;
101    use crate::change::{Change, NodeKindTag};
102    use crate::id::{IdSpace, NodeId};
103    use core::sync::atomic::{AtomicUsize, Ordering};
104
105    struct Counter {
106        changes: AtomicUsize,
107        calls: AtomicUsize,
108    }
109
110    impl Counter {
111        fn new() -> Self {
112            Self {
113                changes: AtomicUsize::new(0),
114                calls: AtomicUsize::new(0),
115            }
116        }
117    }
118
119    impl AstObserver for Counter {
120        fn on_commit(&self, changeset: &Changeset) {
121            self.changes.fetch_add(changeset.len(), Ordering::SeqCst);
122            self.calls.fetch_add(1, Ordering::SeqCst);
123        }
124    }
125
126    fn one_change_set() -> Changeset {
127        let mut cs = Changeset::new(1);
128        cs.push(Change::Created {
129            node: NodeId::from_parts(IdSpace::MODEL, 1),
130            kind: NodeKindTag(1),
131        });
132        cs
133    }
134
135    #[test]
136    fn every_registered_observer_is_notified() {
137        let mut registry = ObserverRegistry::new();
138        let a = Arc::new(Counter::new());
139        let b = Arc::new(Counter::new());
140        let _ = registry.register(a.clone());
141        let _ = registry.register(b.clone());
142
143        registry.notify(&one_change_set());
144
145        assert_eq!(a.changes.load(Ordering::SeqCst), 1);
146        assert_eq!(b.changes.load(Ordering::SeqCst), 1);
147        assert_eq!(registry.len(), 2);
148    }
149
150    #[test]
151    fn an_unregistered_observer_stops_receiving() {
152        let mut registry = ObserverRegistry::new();
153        let observer = Arc::new(Counter::new());
154        let token = registry.register(observer.clone());
155
156        registry.notify(&one_change_set());
157        assert!(registry.unregister(token));
158        registry.notify(&one_change_set());
159
160        assert_eq!(observer.calls.load(Ordering::SeqCst), 1);
161        assert_eq!(registry.len(), 0);
162        assert!(registry.is_empty());
163    }
164
165    #[test]
166    fn unregistering_twice_reports_the_second_call_as_a_no_op() {
167        let mut registry = ObserverRegistry::new();
168        let token = registry.register(Arc::new(Counter::new()));
169        assert!(registry.unregister(token));
170        assert!(!registry.unregister(token));
171    }
172}
zerodds_ast_core/observer.rs

zerodds_ast_core/
observer.rs
