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        if let Some(_guard) = InterceptGuard::enter(&self.intercept_depth, path) {
236            let mut keys_to_intercept = Vec::new();
237            if let Some(k) = change.key() {
238                keys_to_intercept.push(k.clone());
239            } else {
240                let lock = self.interceptors_key.lock().unwrap();
241                keys_to_intercept = lock.keys().cloned().collect();
242            }
243            for key in keys_to_intercept {
244                let interceptors = {
245                    let lock = self.interceptors_key.lock().unwrap();
246                    lock.get(&key).cloned().unwrap_or_default()
247                };
248                for (_, interceptor) in interceptors {
249                    if let Some(new_change) = interceptor(change.clone()) {
250                        change = new_change;
251                    } else {
252                        return Err("Map change intercepted by key filter".to_string());
253                    }
254                }
255            }
256            let interceptors_any = self.interceptors_any.lock().unwrap().clone();
257            for (_, interceptor) in interceptors_any {
258                if let Some(new_change) = interceptor(change.clone()) {
259                    change = new_change;
260                } else {
261                    return Err("Map change intercepted by global filter".to_string());
262                }
263            }
264        }
265        Ok(change)
266    }
267
268    pub fn notify(&self, change: &MapChange<K, V>) {
269        if let Some(k) = change.key()
270            && let Ok(lock) = self.subscribers_key.lock()
271            && let Some(entries) = lock.get(k)
272        {
273            for (_, cb, meta) in entries {
274                tracing::trace!(
275                    target: "amethystate",
276                    subscription_id = meta.id,
277                    name = meta.name,
278                    location = format!("{}:{}", meta.location.file(), meta.location.line()),
279                    "map signal emit → key subscription fire",
280                );
281                cb(change);
282            }
283        }
284
285        if let Ok(lock) = self.subscribers_any.lock() {
286            for (_, cb, meta) in lock.iter() {
287                tracing::trace!(
288                    target: "amethystate",
289                    subscription_id = meta.id,
290                    name = meta.name,
291                    location = format!("{}:{}", meta.location.file(), meta.location.line()),
292                    "map signal emit → any subscription fire",
293                );
294                cb(change);
295            }
296        }
297    }
298}
amethystate_core/primitives/map_core.rs

amethystate_core/primitives/
map_core.rs
