amethystate_core/primitives/
signal.rs

1use crate::ReactiveScope;
2use arc_swap::ArcSwap;
3use std::sync::atomic::{AtomicU64, Ordering};
4use std::sync::{Arc, Mutex};
5use uuid::Uuid;
6
7type SignalCallback<T> = Arc<dyn Fn(&T, Option<Uuid>) + Send + Sync + 'static>;
8type SubscriberEntry<T> = (u64, SignalCallback<T>, SubscriptionMeta);
9type SignalSubscribers<T> = Arc<Mutex<Vec<SubscriberEntry<T>>>>;
10
11#[derive(Clone, Copy)]
12pub struct SubscriptionMeta {
13    pub id: u64,
14    pub location: &'static std::panic::Location<'static>,
15    pub name: Option<&'static str>,
16}
17
18pub struct Signal<T> {
19    pub value: Arc<ArcSwap<T>>,
20    pub subscribers: SignalSubscribers<T>,
21    pub next_id: Arc<AtomicU64>,
22}
23
24impl<T> Clone for Signal<T> {
25    fn clone(&self) -> Self {
26        Self {
27            value: self.value.clone(),
28            next_id: self.next_id.clone(),
29            subscribers: self.subscribers.clone(),
30        }
31    }
32}
33
34#[derive(Clone)]
35pub struct SignalSubscription {
36    pub id: u64,
37    pub location: &'static std::panic::Location<'static>,
38    pub name: Option<&'static str>,
39    pub set_name: Arc<dyn Fn(&'static str) + Send + Sync + 'static>,
40    pub cleanup: Arc<dyn Fn(u64) + Send + Sync + 'static>,
41}
42
43impl SignalSubscription {
44    pub fn named(mut self, name: &'static str) -> Self {
45        self.name = Some(name);
46        (self.set_name)(name);
47        self
48    }
49
50    pub fn watch(self, scope: &mut ReactiveScope) {
51        scope.watch(self);
52    }
53}
54
55impl Drop for SignalSubscription {
56    fn drop(&mut self) {
57        (self.cleanup)(self.id);
58    }
59}
60
61impl<T: 'static> Signal<T> {
62    pub fn new(initial: T) -> Self {
63        Self {
64            value: Arc::new(ArcSwap::from_pointee(initial)),
65            subscribers: Arc::new(Mutex::new(Vec::new())),
66            next_id: Arc::new(AtomicU64::new(0)),
67        }
68    }
69
70    pub fn set(&self, new_value: T, source: Option<Uuid>) {
71        self.value.store(Arc::new(new_value));
72        self.emit(source);
73    }
74
75    fn emit(&self, _source: Option<Uuid>) {
76        let val = self.value.load_full();
77        let callbacks: Vec<_> = {
78            let subs = self.subscribers.lock().unwrap();
79            subs.iter()
80                .map(|(_, cb, meta)| (cb.clone(), *meta))
81                .collect()
82        };
83        for (cb, meta) in callbacks {
84            tracing::trace!(
85                target: "amethystate",
86                subscription_id = meta.id,
87                name = meta.name,
88                location = format!("{}:{}", meta.location.file(), meta.location.line()),
89                "signal emit → subscription fire",
90            );
91            cb(&val, _source);
92        }
93    }
94
95    #[track_caller]
96    pub fn subscribe<F>(&self, callback: F) -> SignalSubscription
97    where
98        F: Fn(&T) + Send + Sync + 'static,
99    {
100        self.subscribe_with_source(move |val, _src| callback(val))
101    }
102
103    #[track_caller]
104    pub fn subscribe_with_source<F>(&self, callback: F) -> SignalSubscription
105    where
106        F: Fn(&T, Option<Uuid>) + Send + Sync + 'static,
107    {
108        let location = std::panic::Location::caller();
109        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
110        let meta = SubscriptionMeta {
111            id,
112            location,
113            name: None,
114        };
115        {
116            let mut subs = self.subscribers.lock().unwrap();
117            subs.push((id, Arc::new(callback), meta));
118        }
119
120        let subscribers_for_name = self.subscribers.clone();
121        let set_name = Arc::new(move |name: &'static str| {
122            if let Ok(mut subs) = subscribers_for_name.lock()
123                && let Some(entry) = subs.iter_mut().find(|(sid, _, _)| *sid == id)
124            {
125                entry.2.name = Some(name);
126            }
127        });
128
129        let subscribers_for_cleanup = self.subscribers.clone();
130        SignalSubscription {
131            id,
132            location,
133            name: None,
134            set_name,
135            cleanup: Arc::new(move |id| {
136                if let Ok(mut subs) = subscribers_for_cleanup.lock() {
137                    subs.retain(|(sid, _, _)| *sid != id);
138                }
139            }),
140        }
141    }
142}
143
144impl<T: Clone + 'static> Signal<T> {
145    pub fn get(&self) -> T {
146        self.value.load().as_ref().clone()
147    }
148}
149
150#[cfg(test)]
151mod tests {
152    use super::*;
153    use std::sync::Mutex;
154
155    #[test]
156    fn signal_subscription_cleanup_on_drop() {
157        let signal = Signal::new("a".to_string());
158        let counter = Arc::new(Mutex::new(0usize));
159        {
160            let cap = counter.clone();
161            let _sub = signal.subscribe(move |_: &String| {
162                *cap.lock().unwrap() += 1;
163            });
164            signal.set("b".to_string(), None);
165            assert_eq!(*counter.lock().unwrap(), 1);
166        }
167        signal.set("c".to_string(), None);
168        assert_eq!(*counter.lock().unwrap(), 1);
169    }
170
171    #[test]
172    fn named_updates_meta_in_subscribers() {
173        let signal = Signal::new(0i32);
174        let _sub = signal.subscribe(|_| {}).named("MyWatcher");
175
176        let subs = signal.subscribers.lock().unwrap();
177        assert_eq!(subs[0].2.name, Some("MyWatcher"));
178    }
179
180    #[test]
181    fn subscription_location_captured() {
182        let signal = Signal::new(0i32);
183        let sub = signal.subscribe(|_| {});
184        assert!(!sub.location.file().is_empty());
185    }
186}
amethystate_core/primitives/signal.rs

amethystate_core/primitives/
signal.rs
