amethystate_core/primitives/
map_core.rs

1use crate::SignalSubscription;
2use crate::change::MapChange;
3use crate::primitives::intercept::{InterceptDisposer, InterceptGuard};
4use crate::primitives::signal::SubscriptionMeta;
5use serde::Serialize;
6use serde::de::DeserializeOwned;
7use std::collections::HashMap;
8use std::fmt::Display;
9use std::hash::Hash;
10use std::str::FromStr;
11use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
12use std::sync::{Arc, Mutex};
13
14pub type InterceptorAny<K, V> =
15    Arc<dyn Fn(MapChange<K, V>) -> Option<MapChange<K, V>> + Send + Sync + 'static>;
16pub type InterceptorKey<K, V> =
17    Arc<dyn Fn(MapChange<K, V>) -> Option<MapChange<K, V>> + Send + Sync + 'static>;
18pub type SubscriberAny<K, V> = Arc<dyn Fn(&MapChange<K, V>) + Send + Sync + 'static>;
19pub type SubscriberKey<K, V> = Arc<dyn Fn(&MapChange<K, V>) + Send + Sync + 'static>;
20
21pub trait ReactiveMapKey: FromStr + Display + Clone + Hash + Eq + Send + Sync + 'static {}
22impl<T: FromStr + Display + Clone + Hash + Eq + Send + Sync + 'static> ReactiveMapKey for T {}
23
24pub trait ReactiveMapValue:
25    Serialize + DeserializeOwned + Clone + Send + Sync + 'static + Default
26{
27}
28impl<T: Serialize + DeserializeOwned + Clone + Send + Sync + 'static + Default> ReactiveMapValue
29    for T
30{
31}
32
33pub struct ReactiveMapCore<K, V> {
34    pub interceptors_any: Arc<Mutex<Vec<(u64, InterceptorAny<K, V>)>>>,
35    pub interceptors_key: Arc<Mutex<HashMap<K, Vec<(u64, InterceptorKey<K, V>)>>>>,
36    pub subscribers_any: Arc<Mutex<Vec<(u64, SubscriberAny<K, V>, SubscriptionMeta)>>>,
37    pub subscribers_key: Arc<Mutex<HashMap<K, Vec<(u64, SubscriberKey<K, V>, SubscriptionMeta)>>>>,
38    pub next_id: Arc<AtomicU64>,
39    pub intercept_depth: Arc<AtomicUsize>,
40    pub cache: Arc<Mutex<HashMap<K, V>>>,
41}
42
43impl<K, V> Clone for ReactiveMapCore<K, V> {
44    fn clone(&self) -> Self {
45        Self {
46            interceptors_any: self.interceptors_any.clone(),
47            interceptors_key: self.interceptors_key.clone(),
48            subscribers_any: self.subscribers_any.clone(),
49            subscribers_key: self.subscribers_key.clone(),
50            next_id: self.next_id.clone(),
51            intercept_depth: self.intercept_depth.clone(),
52            cache: self.cache.clone(),
53        }
54    }
55}
56
57impl<K: std::fmt::Debug, V: std::fmt::Debug> std::fmt::Debug for ReactiveMapCore<K, V> {
58    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
59        let mut d = f.debug_struct("ReactiveMapCore");
60        if let Ok(cache) = self.cache.try_lock() {
61            d.field("cache", &*cache);
62        } else {
63            d.field("cache", &"<locked>");
64        }
65        let interceptors_count = self
66            .interceptors_any
67            .try_lock()
68            .map(|l| l.len())
69            .unwrap_or(0);
70        let subscribers_count = self
71            .subscribers_any
72            .try_lock()
73            .map(|l| l.len())
74            .unwrap_or(0);
75        d.field("interceptors_any_count", &interceptors_count)
76            .field("subscribers_any_count", &subscribers_count)
77            .finish()
78    }
79}
80
81impl<K: ReactiveMapKey, V: ReactiveMapValue> Default for ReactiveMapCore<K, V> {
82    fn default() -> Self {
83        Self::new()
84    }
85}
86
87impl<K: ReactiveMapKey, V: ReactiveMapValue> ReactiveMapCore<K, V> {
88    pub fn new() -> Self {
89        Self {
90            interceptors_any: Arc::new(Mutex::new(Vec::new())),
91            interceptors_key: Arc::new(Mutex::new(HashMap::new())),
92            subscribers_any: Arc::new(Mutex::new(Vec::new())),
93            subscribers_key: Arc::new(Mutex::new(HashMap::new())),
94            next_id: Arc::new(AtomicU64::new(0)),
95            intercept_depth: Arc::new(AtomicUsize::new(0)),
96            cache: Arc::new(Mutex::new(HashMap::new())),
97        }
98    }
99
100    #[track_caller]
101    pub fn subscribe_any<F>(&self, callback: F) -> SignalSubscription
102    where
103        F: Fn(&MapChange<K, V>) + Send + Sync + 'static,
104    {
105        let location = std::panic::Location::caller();
106        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
107        let meta = SubscriptionMeta {
108            id,
109            location,
110            name: None,
111        };
112        self.subscribers_any
113            .lock()
114            .unwrap()
115            .push((id, Arc::new(callback), meta));
116
117        let subs_for_name = self.subscribers_any.clone();
118        let set_name = Arc::new(move |name: &'static str| {
119            if let Ok(mut lock) = subs_for_name.lock()
120                && let Some(entry) = lock.iter_mut().find(|(i, _, _)| *i == id)
121            {
122                entry.2.name = Some(name);
123            }
124        });
125        let subs_for_cleanup = self.subscribers_any.clone();
126        SignalSubscription {
127            id,
128            location,
129            name: None,
130            set_name,
131            cleanup: Arc::new(move |id| {
132                if let Ok(mut lock) = subs_for_cleanup.lock() {
133                    lock.retain(|(i, _, _)| *i != id);
134                }
135            }),
136        }
137    }
138
139    #[track_caller]
140    pub fn subscribe_key<F>(&self, key: K, callback: F) -> SignalSubscription
141    where
142        F: Fn(&MapChange<K, V>) + Send + Sync + 'static,
143    {
144        let location = std::panic::Location::caller();
145        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
146        let meta = SubscriptionMeta {
147            id,
148            location,
149            name: None,
150        };
151        self.subscribers_key
152            .lock()
153            .unwrap()
154            .entry(key.clone())
155            .or_default()
156            .push((id, Arc::new(callback), meta));
157
158        let subs_for_name = self.subscribers_key.clone();
159        let key_for_name = key.clone();
160        let set_name = Arc::new(move |name: &'static str| {
161            if let Ok(mut lock) = subs_for_name.lock()
162                && let Some(list) = lock.get_mut(&key_for_name)
163                && let Some(entry) = list.iter_mut().find(|(i, _, _)| *i == id)
164            {
165                entry.2.name = Some(name);
166            }
167        });
168        let subs_for_cleanup = self.subscribers_key.clone();
169        SignalSubscription {
170            id,
171            location,
172            name: None,
173            set_name,
174            cleanup: Arc::new(move |id| {
175                if let Ok(mut lock) = subs_for_cleanup.lock()
176                    && let Some(list) = lock.get_mut(&key)
177                {
178                    list.retain(|(i, _, _)| *i != id);
179                }
180            }),
181        }
182    }
183
184    pub fn intercept<F>(&self, path: Arc<str>, callback: F) -> InterceptDisposer
185    where
186        F: Fn(MapChange<K, V>) -> Option<MapChange<K, V>> + Send + Sync + 'static,
187    {
188        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
189        self.interceptors_any
190            .lock()
191            .unwrap()
192            .push((id, Arc::new(callback)));
193        let subs = self.interceptors_any.clone();
194        InterceptDisposer {
195            id,
196            path,
197            cleanup: Arc::new(move |id| {
198                if let Ok(mut lock) = subs.lock() {
199                    lock.retain(|(i, _)| *i != id);
200                }
201            }),
202        }
203    }
204
205    pub fn intercept_key<F>(&self, key: K, callback: F) -> InterceptDisposer
206    where
207        F: Fn(MapChange<K, V>) -> Option<MapChange<K, V>> + Send + Sync + 'static,
208    {
209        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
210        self.interceptors_key
211            .lock()
212            .unwrap()
213            .entry(key.clone())
214            .or_default()
215            .push((id, Arc::new(callback)));
216        let subs = self.interceptors_key.clone();
217        InterceptDisposer {
218            id,
219            path: Arc::from(""),
220            cleanup: Arc::new(move |id| {
221                if let Ok(mut lock) = subs.lock()
222                    && let Some(list) = lock.get_mut(&key)
223                {
224                    list.retain(|(i, _)| *i != id);
225                }
226            }),
227        }
228    }
229
230    pub fn run_interceptors(
231        &self,
232        path: Arc<str>,
233        mut change: MapChange<K, V>,
234    ) -> Result<MapChange<K, V>, String> {
235        let Some(_guard) = InterceptGuard::enter(&self.intercept_depth, path) else {
236            // Letting the change through unchecked would turn a validating
237            // interceptor off exactly where recursion is deepest, and the value
238            // it exists to reject would reach the backend.
239            return Err("Maximum intercept depth reached".to_string());
240        };
241
242        // A change with no key is not about any one key, so key interceptors do
243        // not apply. Running all of them used to hand each the same `Clear`,
244        // accumulating their rewrites, in `HashMap` order.
245        if let Some(key) = change.key().cloned() {
246            let interceptors = {
247                let lock = self.interceptors_key.lock().unwrap();
248                lock.get(&key).cloned().unwrap_or_default()
249            };
250            for (_, interceptor) in interceptors {
251                if let Some(new_change) = interceptor(change.clone()) {
252                    change = new_change;
253                } else {
254                    return Err("Map change intercepted by key filter".to_string());
255                }
256            }
257        }
258
259        let interceptors_any = self.interceptors_any.lock().unwrap().clone();
260        for (_, interceptor) in interceptors_any {
261            if let Some(new_change) = interceptor(change.clone()) {
262                change = new_change;
263            } else {
264                return Err("Map change intercepted by global filter".to_string());
265            }
266        }
267
268        Ok(change)
269    }
270
271    /// Fires every subscriber interested in `change`.
272    ///
273    /// The callbacks are collected before any of them runs, and the locks are
274    /// released first. A subscriber reacting to a change by writing to the same
275    /// map is ordinary, and `std::sync::Mutex` is not reentrant, so holding the
276    /// lock across the calls deadlocks the thread. It also means a panicking
277    /// callback cannot poison the subscriber lists.
278    pub fn notify(&self, change: &MapChange<K, V>) {
279        let keyed: Vec<_> = match change.key() {
280            Some(k) => self
281                .subscribers_key
282                .lock()
283                .ok()
284                .and_then(|lock| {
285                    lock.get(k).map(|entries| {
286                        entries
287                            .iter()
288                            .map(|(_, cb, meta)| (cb.clone(), *meta))
289                            .collect()
290                    })
291                })
292                .unwrap_or_default(),
293            None => Vec::new(),
294        };
295
296        let any: Vec<_> = self
297            .subscribers_any
298            .lock()
299            .map(|lock| {
300                lock.iter()
301                    .map(|(_, cb, meta)| (cb.clone(), *meta))
302                    .collect()
303            })
304            .unwrap_or_default();
305
306        for (cb, meta) in keyed {
307            tracing::trace!(
308                target: "amethystate",
309                subscription_id = meta.id,
310                name = meta.name,
311                location = format!("{}:{}", meta.location.file(), meta.location.line()),
312                "map signal emit → key subscription fire",
313            );
314            cb(change);
315        }
316
317        for (cb, meta) in any {
318            tracing::trace!(
319                target: "amethystate",
320                subscription_id = meta.id,
321                name = meta.name,
322                location = format!("{}:{}", meta.location.file(), meta.location.line()),
323                "map signal emit → any subscription fire",
324            );
325            cb(change);
326        }
327    }
328}
amethystate_core/primitives/map_core.rs

amethystate_core/primitives/
map_core.rs
