a2ui_base/observable/
signal.rs

1//! Stateful reactive value (signal / BehaviorSubject pattern).
2//!
3//! Holds a current value and notifies listeners when it changes.
4
5use std::sync::Arc;
6
7use super::event_stream::{EventSubscription, EventStream};
8
9/// A reactive container that holds a value and notifies subscribers on change.
10pub struct Signal<T: 'static> {
11    value: Arc<std::sync::Mutex<T>>,
12    stream: EventStream<T>,
13}
14
15impl<T: Clone + 'static> Signal<T> {
16    pub fn new(initial: T) -> Self {
17        Self {
18            value: Arc::new(std::sync::Mutex::new(initial)),
19            stream: EventStream::new(),
20        }
21    }
22
23    /// Get the current value.
24    pub fn get(&self) -> T {
25        self.value.lock().unwrap().clone()
26    }
27
28    /// Set a new value and notify all subscribers.
29    pub fn set(&self, new_value: T) {
30        {
31            let mut guard = self.value.lock().unwrap();
32            *guard = new_value.clone();
33        }
34        self.stream.emit(&new_value);
35    }
36
37    /// Subscribe to value changes. Fires on every `set()`.
38    pub fn subscribe<F>(&self, callback: F) -> EventSubscription
39    where
40        F: Fn(&T) + Send + Sync + 'static,
41    {
42        self.stream.on(callback)
43    }
44
45    /// Subscribe and immediately fire with the current value.
46    #[allow(dead_code)]
47    pub fn subscribe_with_initial<F>(&self, callback: F) -> EventSubscription
48    where
49        F: Fn(&T) + Send + Sync + 'static,
50    {
51        let current = self.get();
52        callback(&current);
53        self.stream.on(callback)
54    }
55}
56
57impl<T: Clone + 'static> Clone for Signal<T> {
58    fn clone(&self) -> Self {
59        Self {
60            value: Arc::clone(&self.value),
61            stream: self.stream.clone(),
62        }
63    }
64}
65
66#[cfg(test)]
67mod tests {
68    use super::*;
69
70    #[test]
71    fn test_set_and_get() {
72        let signal = Signal::new(0i32);
73        assert_eq!(signal.get(), 0);
74        signal.set(42);
75        assert_eq!(signal.get(), 42);
76    }
77
78    #[test]
79    fn test_notifies_on_change() {
80        let signal = Signal::new(0i32);
81        let received = Arc::new(std::sync::Mutex::new(Vec::new()));
82        let r = Arc::clone(&received);
83
84        let _sub = signal.subscribe(move |v: &i32| {
85            r.lock().unwrap().push(*v);
86        });
87
88        signal.set(1);
89        signal.set(2);
90        signal.set(3);
91        assert_eq!(*received.lock().unwrap(), vec![1, 2, 3]);
92    }
93}
a2ui_base/observable/signal.rs

a2ui_base/observable/
signal.rs
