1use crate::SignalSubscription;
2use crate::change::MapChange;
3use crate::path::{PathRef, StorePath, Under};
4use crate::primitives::intercept::{InterceptDisposer, InterceptGuard};
5use crate::primitives::signal::{SubscriptionMeta, forget, held, label};
6use arc_swap::ArcSwap;
7use dashmap::DashMap;
8use rpds::RedBlackTreeMapSync;
9use serde::de::DeserializeOwned;
10use serde::{Deserialize, Serialize};
11use smol_str::{SmolStr, SmolStrBuilder};
12use std::any;
13use std::borrow::Borrow;
14use std::fmt::{self, Debug, Display, Write as _};
15use std::hash::Hash;
16use std::panic::Location;
17use std::str::FromStr;
18use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
19use std::sync::{Arc, Mutex};
20
21pub trait MapEntryPath {
23 fn entry(&self, name: &str) -> StorePath;
25}
26
27impl MapEntryPath for StorePath {
28 fn entry(&self, name: &str) -> StorePath {
29 self.push(name)
30 }
31}
32
33pub type InterceptorAny<K, V> =
34 Arc<dyn Fn(MapChange<K, V>) -> Option<MapChange<K, V>> + Send + Sync + 'static>;
35pub type InterceptorKey<K, V> =
36 Arc<dyn Fn(MapChange<K, V>) -> Option<MapChange<K, V>> + Send + Sync + 'static>;
37pub type SubscriberAny<K, V> = Arc<dyn Fn(&MapChange<K, V>) + Send + Sync + 'static>;
38pub type SubscriberKey<K, V> = Arc<dyn Fn(&MapChange<K, V>) + Send + Sync + 'static>;
39
40pub enum EntryOf<K> {
58 Took(K),
60
61 ThePathItself,
64
65 NameWillNotRead,
67
68 NotThisMaps,
72}
73
74pub fn entry_of<K>(prefix: &StorePath, stored: PathRef<'_>) -> EntryOf<K>
76where
77 K: ReactiveMapKey,
78{
79 match stored.level_under(prefix) {
80 Under::Prefix => EntryOf::ThePathItself,
81 Under::Deeper(_) | Under::Outside => EntryOf::NotThisMaps,
82 Under::Entry(name) => match K::read(name.as_str()) {
83 Some(key) => EntryOf::Took(key),
84 None => EntryOf::NameWillNotRead,
85 },
86 }
87}
88
89pub trait ReactiveMapKey: AsRef<str> + Clone + Hash + Eq + Send + Sync + 'static {
90 fn read(name: &str) -> Option<Self>;
96}
97
98impl ReactiveMapKey for String {
99 fn read(name: &str) -> Option<Self> {
100 Some(name.to_string())
101 }
102}
103
104impl ReactiveMapKey for SmolStr {
105 fn read(name: &str) -> Option<Self> {
106 Some(SmolStr::new(name))
107 }
108}
109
110impl<T> ReactiveMapKey for Id<T>
111where
112 T: Display + FromStr + Clone + Send + Sync + 'static,
113{
114 fn read(name: &str) -> Option<Self> {
115 T::from_str(name).ok().map(Id::new)
116 }
117}
118
119#[derive(Clone)]
132pub struct Id<T> {
133 spelled: SmolStr,
134 value: T,
135}
136
137impl<T: Display> Id<T> {
138 pub fn new(value: T) -> Self {
139 let mut out = SmolStrBuilder::new();
140 let _ = write!(out, "{value}");
141
142 Self {
143 spelled: out.finish(),
144 value,
145 }
146 }
147}
148
149impl<T> Id<T> {
150 pub fn get(&self) -> &T {
152 &self.value
153 }
154
155 pub fn into_inner(self) -> T {
156 self.value
157 }
158}
159
160impl<T> AsRef<str> for Id<T> {
161 fn as_ref(&self) -> &str {
162 &self.spelled
163 }
164}
165
166impl<T> Borrow<str> for Id<T> {
170 fn borrow(&self) -> &str {
171 &self.spelled
172 }
173}
174
175impl<T> PartialEq for Id<T> {
178 fn eq(&self, other: &Self) -> bool {
179 self.spelled == other.spelled
180 }
181}
182
183impl<T> Eq for Id<T> {}
184
185impl<T> Hash for Id<T> {
186 fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
187 self.spelled.hash(state);
188 }
189}
190
191impl<T> Display for Id<T> {
192 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
193 f.write_str(&self.spelled)
194 }
195}
196
197impl<T: Debug> Debug for Id<T> {
198 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
199 f.debug_tuple("Id").field(&self.value).finish()
200 }
201}
202
203impl<T> Serialize for Id<T> {
207 fn serialize<S: serde::Serializer>(&self, out: S) -> Result<S::Ok, S::Error> {
208 out.serialize_str(&self.spelled)
209 }
210}
211
212impl<'de, T: Display + FromStr> Deserialize<'de> for Id<T> {
213 fn deserialize<D: serde::Deserializer<'de>>(from: D) -> Result<Self, D::Error> {
214 let spelled = <std::borrow::Cow<'de, str>>::deserialize(from)?;
215
216 T::from_str(&spelled).map(Id::new).map_err(|_| {
217 serde::de::Error::custom(format!("{spelled:?} is not a {}", any::type_name::<T>()))
218 })
219 }
220}
221
222pub trait ReactiveMapValue:
223 Serialize + DeserializeOwned + Clone + Send + Sync + 'static + Default
224{
225}
226impl<T: Serialize + DeserializeOwned + Clone + Send + Sync + 'static + Default> ReactiveMapValue
227 for T
228{
229}
230
231pub struct MapCache<K, V> {
260 entries: ArcSwap<Snapshot<K, V>>,
261}
262
263impl<K: AsRef<str>, V> Default for MapCache<K, V> {
264 fn default() -> Self {
265 Self {
266 entries: ArcSwap::from_pointee(RedBlackTreeMapSync::new_sync()),
267 }
268 }
269}
270
271#[derive(Clone)]
279struct ByName<K>(K);
280
281impl<K: AsRef<str>> Borrow<str> for ByName<K> {
282 fn borrow(&self) -> &str {
283 self.0.as_ref()
284 }
285}
286
287impl<K: AsRef<str>> PartialEq for ByName<K> {
288 fn eq(&self, other: &Self) -> bool {
289 self.0.as_ref() == other.0.as_ref()
290 }
291}
292
293impl<K: AsRef<str>> Eq for ByName<K> {}
294
295impl<K: AsRef<str>> Ord for ByName<K> {
296 fn cmp(&self, other: &Self) -> std::cmp::Ordering {
297 self.0.as_ref().cmp(other.0.as_ref())
298 }
299}
300
301impl<K: AsRef<str>> PartialOrd for ByName<K> {
302 fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
303 Some(self.cmp(other))
304 }
305}
306
307impl<K: AsRef<str> + Clone, V: Clone> MapCache<K, V> {
308 pub fn new() -> Self {
309 Self::default()
310 }
311
312 pub fn get(&self, name: &str) -> Option<V> {
313 self.entries.load().get(name).cloned()
314 }
315
316 pub fn contains_key(&self, name: &str) -> bool {
317 self.entries.load().contains_key(name)
318 }
319
320 pub fn owned_key(&self, name: &str) -> Option<K> {
323 self.entries
324 .load()
325 .get_key_value(name)
326 .map(|(key, _)| key.0.clone())
327 }
328
329 pub fn len(&self) -> usize {
330 self.entries.load().size()
331 }
332
333 pub fn is_empty(&self) -> bool {
334 self.entries.load().is_empty()
335 }
336
337 pub fn clear(&self) {
338 self.entries
339 .store(Arc::new(RedBlackTreeMapSync::new_sync()));
340 }
341
342 pub fn view(&self) -> Entries<K, V> {
347 Entries {
348 held: self.entries.load_full(),
349 }
350 }
351
352 pub fn keys(&self) -> Walk<K, V, K> {
354 Walk::new(self.entries.load_full(), |key, _| key.0.clone())
355 }
356
357 pub fn entries(&self) -> Walk<K, V, (K, V)> {
359 Walk::new(self.entries.load_full(), |key, value| {
360 (key.0.clone(), value.clone())
361 })
362 }
363}
364
365type Snapshot<K, V> = RedBlackTreeMapSync<ByName<K>, V>;
366type Held<K, V> = Arc<Snapshot<K, V>>;
367type Pairs<'a, K, V> = <&'a Snapshot<K, V> as IntoIterator>::IntoIter;
368pub struct Values<'a, K: AsRef<str>, V> {
373 pairs: Pairs<'a, K, V>,
374}
375
376impl<'a, K: AsRef<str>, V> Iterator for Values<'a, K, V> {
377 type Item = (&'a K, &'a V);
378
379 fn next(&mut self) -> Option<Self::Item> {
380 self.pairs.next().map(|(key, value)| (&key.0, value))
381 }
382
383 fn size_hint(&self) -> (usize, Option<usize>) {
384 self.pairs.size_hint()
385 }
386}
387
388impl<K: AsRef<str>, V> DoubleEndedIterator for Values<'_, K, V> {
389 fn next_back(&mut self) -> Option<Self::Item> {
390 self.pairs.next_back().map(|(key, value)| (&key.0, value))
391 }
392}
393
394pub struct Walk<K: AsRef<str> + 'static, V: 'static, T> {
403 pairs: Pairs<'static, K, V>,
404 take: fn(&ByName<K>, &V) -> T,
405 _held: Held<K, V>,
406}
407
408impl<K: AsRef<str>, V, T> Walk<K, V, T> {
409 fn new(held: Held<K, V>, take: fn(&ByName<K>, &V) -> T) -> Self {
410 let pairs =
421 unsafe { std::mem::transmute::<Pairs<'_, K, V>, Pairs<'static, K, V>>(held.iter()) };
422
423 Self {
424 pairs,
425 take,
426 _held: held,
427 }
428 }
429}
430
431impl<K: AsRef<str>, V, T> Iterator for Walk<K, V, T> {
432 type Item = T;
433
434 fn next(&mut self) -> Option<T> {
435 self.pairs
436 .next()
437 .map(|(key, value)| (self.take)(key, value))
438 }
439
440 fn size_hint(&self) -> (usize, Option<usize>) {
441 self.pairs.size_hint()
442 }
443}
444
445impl<K: AsRef<str>, V, T> DoubleEndedIterator for Walk<K, V, T> {
446 fn next_back(&mut self) -> Option<T> {
447 self.pairs
448 .next_back()
449 .map(|(key, value)| (self.take)(key, value))
450 }
451}
452
453impl<K: AsRef<str>, V, T> ExactSizeIterator for Walk<K, V, T> {}
454
455pub struct Entries<K, V> {
457 held: Held<K, V>,
458}
459
460impl<'e, K: AsRef<str>, V> IntoIterator for &'e Entries<K, V> {
461 type Item = (&'e K, &'e V);
462 type IntoIter = Values<'e, K, V>;
463
464 fn into_iter(self) -> Self::IntoIter {
465 Values {
466 pairs: self.held.iter(),
467 }
468 }
469}
470
471impl<K: AsRef<str>, V> Entries<K, V> {
472 pub fn iter(&self) -> impl DoubleEndedIterator<Item = (&K, &V)> {
473 self.held.iter().map(|(key, value)| (&key.0, value))
474 }
475
476 pub fn len(&self) -> usize {
477 self.held.size()
478 }
479
480 pub fn is_empty(&self) -> bool {
481 self.held.is_empty()
482 }
483}
484
485impl<K: AsRef<str> + Clone, V: Clone> MapCache<K, V> {
486 pub fn insert(&self, key: K, value: V) -> Option<V> {
487 let replaced = self
488 .entries
489 .rcu(|current| current.insert(ByName(key.clone()), value.clone()));
490
491 replaced.get(key.as_ref()).cloned()
492 }
493
494 pub fn remove(&self, name: &str) -> Option<(K, V)> {
495 let previous = self.entries.rcu(|current| current.remove(name));
496
497 previous
498 .get_key_value(name)
499 .map(|(key, value)| (key.0.clone(), value.clone()))
500 }
501}
502
503impl<K: AsRef<str> + Debug, V: Debug> Debug for MapCache<K, V> {
504 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
505 let held = self.entries.load();
506
507 f.debug_map()
508 .entries(held.iter().map(|(key, value)| (&key.0, value)))
509 .finish()
510 }
511}
512
513pub struct ReactiveMapCore<K, V> {
514 pub interceptors_any: Arc<Mutex<Vec<(u64, InterceptorAny<K, V>)>>>,
515 pub interceptors_key: Arc<DashMap<K, Vec<(u64, InterceptorKey<K, V>)>>>,
516 pub subscribers_any: Arc<Mutex<Vec<(u64, SubscriberAny<K, V>, SubscriptionMeta)>>>,
517 pub subscribers_key: Arc<DashMap<K, Vec<(u64, SubscriberKey<K, V>, SubscriptionMeta)>>>,
518 pub next_id: Arc<AtomicU64>,
519 pub intercept_depth: Arc<AtomicUsize>,
520 pub cache: Arc<MapCache<K, V>>,
521}
522
523impl<K, V> Clone for ReactiveMapCore<K, V> {
524 fn clone(&self) -> Self {
525 Self {
526 interceptors_any: self.interceptors_any.clone(),
527 interceptors_key: self.interceptors_key.clone(),
528 subscribers_any: self.subscribers_any.clone(),
529 subscribers_key: self.subscribers_key.clone(),
530 next_id: self.next_id.clone(),
531 intercept_depth: self.intercept_depth.clone(),
532 cache: self.cache.clone(),
533 }
534 }
535}
536
537struct Counted<'a, T>(&'a Mutex<Vec<T>>);
538
539impl<T> Debug for Counted<'_, T> {
540 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
541 match self.0.try_lock() {
542 Ok(list) => write!(f, "{}", list.len()),
543 Err(_) => f.write_str("<locked>"),
544 }
545 }
546}
547
548impl<K: AsRef<str> + Debug + Hash + Eq, V: Debug> Debug for ReactiveMapCore<K, V> {
549 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
550 f.debug_struct("ReactiveMapCore")
551 .field("cache", &self.cache)
552 .field("interceptors_any", &Counted(&self.interceptors_any))
553 .field("subscribers_any", &Counted(&self.subscribers_any))
554 .finish()
555 }
556}
557
558impl<K: ReactiveMapKey, V: ReactiveMapValue> Default for ReactiveMapCore<K, V> {
559 fn default() -> Self {
560 Self::new()
561 }
562}
563
564impl<K: ReactiveMapKey, V: ReactiveMapValue> ReactiveMapCore<K, V> {
565 pub fn new() -> Self {
566 Self::with_capacity(0)
567 }
568
569 pub fn with_capacity(_entries: usize) -> Self {
573 Self {
574 interceptors_any: Arc::new(Mutex::new(Vec::new())),
575 interceptors_key: Arc::new(DashMap::new()),
576 subscribers_any: Arc::new(Mutex::new(Vec::new())),
577 subscribers_key: Arc::new(DashMap::new()),
578 next_id: Arc::new(AtomicU64::new(0)),
579 intercept_depth: Arc::new(AtomicUsize::new(0)),
580 cache: Arc::new(MapCache::new()),
581 }
582 }
583
584 #[track_caller]
585 pub fn subscribe_any<F>(&self, callback: F) -> SignalSubscription
586 where
587 F: Fn(&MapChange<K, V>) + Send + Sync + 'static,
588 {
589 let location = Location::caller();
590 let id = self.next_id.fetch_add(1, Ordering::Relaxed);
591 let meta = SubscriptionMeta {
592 id,
593 location,
594 name: None,
595 };
596 held(&self.subscribers_any).push((id, Arc::new(callback), meta));
597
598 let subs_for_name = self.subscribers_any.clone();
599 let set_name = Arc::new(move |name: &'static str| {
600 label(&mut held(&subs_for_name), id, name);
601 });
602 let subs_for_cleanup = self.subscribers_any.clone();
603 SignalSubscription::new(
604 id,
605 location,
606 set_name,
607 Arc::new(move |id| {
608 forget(&mut held(&subs_for_cleanup), id);
609 }),
610 )
611 }
612
613 #[track_caller]
614 pub fn subscribe_key<F>(&self, key: K, callback: F) -> SignalSubscription
615 where
616 F: Fn(&MapChange<K, V>) + Send + Sync + 'static,
617 {
618 let location = Location::caller();
619 let id = self.next_id.fetch_add(1, Ordering::Relaxed);
620 let meta = SubscriptionMeta {
621 id,
622 location,
623 name: None,
624 };
625 self.subscribers_key
626 .entry(key.clone())
627 .or_default()
628 .push((id, Arc::new(callback), meta));
629
630 let subs_for_name = self.subscribers_key.clone();
631 let key_for_name = key.clone();
632 let set_name = Arc::new(move |name: &'static str| {
633 if let Some(mut list) = subs_for_name.get_mut(&key_for_name) {
634 label(&mut list, id, name);
635 }
636 });
637 let subs_for_cleanup = self.subscribers_key.clone();
638 SignalSubscription::new(
639 id,
640 location,
641 set_name,
642 Arc::new(move |id| {
643 if let Some(mut list) = subs_for_cleanup.get_mut(&key) {
644 forget(&mut list, id);
645 }
646 subs_for_cleanup.remove_if(&key, |_, list| list.is_empty());
647 }),
648 )
649 }
650
651 pub fn intercept<F>(&self, path: StorePath, callback: F) -> InterceptDisposer
652 where
653 F: Fn(MapChange<K, V>) -> Option<MapChange<K, V>> + Send + Sync + 'static,
654 {
655 let id = self.next_id.fetch_add(1, Ordering::Relaxed);
656 held(&self.interceptors_any).push((id, Arc::new(callback)));
657 let subs = self.interceptors_any.clone();
658 InterceptDisposer {
659 id,
660 path,
661 cleanup: Arc::new(move |id| {
662 held(&subs).retain(|(i, _)| *i != id);
663 }),
664 }
665 }
666
667 pub fn intercept_key<F>(&self, key: K, callback: F) -> InterceptDisposer
668 where
669 F: Fn(MapChange<K, V>) -> Option<MapChange<K, V>> + Send + Sync + 'static,
670 {
671 let id = self.next_id.fetch_add(1, Ordering::Relaxed);
672 self.interceptors_key
673 .entry(key.clone())
674 .or_default()
675 .push((id, Arc::new(callback)));
676 let subs = self.interceptors_key.clone();
677 InterceptDisposer {
678 id,
679 path: StorePath::root(),
680 cleanup: Arc::new(move |id| {
681 if let Some(mut list) = subs.get_mut(&key) {
682 list.retain(|(i, _)| *i != id);
683 }
684 subs.remove_if(&key, |_, list| list.is_empty());
685 }),
686 }
687 }
688
689 pub fn run_interceptors(
690 &self,
691 path: StorePath,
692 mut change: MapChange<K, V>,
693 ) -> Result<MapChange<K, V>, crate::primitives::intercept::Refusal> {
694 use crate::primitives::intercept::Refusal;
695
696 let Some(_guard) = InterceptGuard::enter(&self.intercept_depth, path) else {
697 return Err(Refusal::Recursed);
698 };
699
700 if let Some(key) = change.key().cloned() {
701 let interceptors = self
702 .interceptors_key
703 .get(&key)
704 .map(|entry| entry.clone())
705 .unwrap_or_default();
706 for (_, interceptor) in interceptors {
707 if let Some(new_change) = interceptor(change.clone()) {
708 change = new_change;
709 } else {
710 return Err(Refusal::Said(
711 "refused by an interceptor on that key".to_string(),
712 ));
713 }
714 }
715 }
716
717 let interceptors_any = held(&self.interceptors_any).clone();
718 for (_, interceptor) in interceptors_any {
719 if let Some(new_change) = interceptor(change.clone()) {
720 change = new_change;
721 } else {
722 return Err(Refusal::Said(
723 "refused by an interceptor on the map".to_string(),
724 ));
725 }
726 }
727
728 Ok(change)
729 }
730
731 pub fn notify(&self, change: &MapChange<K, V>) {
738 let keyed: Vec<_> = match change.key() {
739 Some(k) => self
740 .subscribers_key
741 .get(k)
742 .map(|entries| {
743 entries
744 .iter()
745 .map(|(_, cb, meta)| (cb.clone(), *meta))
746 .collect()
747 })
748 .unwrap_or_default(),
749
750 None => self
755 .subscribers_key
756 .iter()
757 .flat_map(|entries| {
758 entries
759 .value()
760 .iter()
761 .map(|(_, cb, meta)| (cb.clone(), *meta))
762 .collect::<Vec<_>>()
763 })
764 .collect(),
765 };
766
767 let any: Vec<_> = held(&self.subscribers_any)
768 .iter()
769 .map(|(_, cb, meta)| (cb.clone(), *meta))
770 .collect();
771
772 for (cb, meta) in keyed {
773 tracing::trace!(
774 target: "amethystate",
775 subscription_id = meta.id,
776 name = meta.name,
777 location = format!("{}:{}", meta.location.file(), meta.location.line()),
778 "map signal emit → key subscription fire",
779 );
780 cb(change);
781 }
782
783 for (cb, meta) in any {
784 tracing::trace!(
785 target: "amethystate",
786 subscription_id = meta.id,
787 name = meta.name,
788 location = format!("{}:{}", meta.location.file(), meta.location.line()),
789 "map signal emit → any subscription fire",
790 );
791 cb(change);
792 }
793 }
794}
795
796#[cfg(test)]
797mod tests {
798 use super::*;
799
800 #[test]
801 fn the_cache_lists_entries_in_the_order_the_store_keys_them() {
802 let cache = MapCache::<String, u8>::new();
803 let names = ["a1b", "a.b", "10", "9", "100", "a\\b", "cpu", "Cpu"];
804
805 for (n, name) in names.iter().enumerate() {
806 cache.insert((*name).to_string(), n as u8);
807 }
808
809 let under = StorePath::segment("m");
810 let mut by_store = names.to_vec();
811 by_store.sort_by_key(|name| under.push(name).key().as_bytes().to_vec());
812
813 assert_eq!(cache.keys().collect::<Vec<_>>(), by_store);
814 }
815
816 #[test]
817 fn a_name_holding_the_separator_is_one_entry() {
818 let cache = MapCache::<String, u8>::new();
819
820 cache.insert("dark.mode".to_string(), 1);
821
822 assert_eq!(cache.len(), 1);
823 assert_eq!(cache.get("dark.mode"), Some(1));
824 assert_eq!(cache.get("dark"), None);
825 assert_eq!(cache.get("dark\\.mode"), None);
826 assert_eq!(cache.keys().collect::<Vec<_>>(), ["dark.mode"]);
827 }
828
829 #[test]
830 fn a_dropped_key_subscription_leaves_no_entry_behind() {
831 let core = ReactiveMapCore::<String, u64>::new();
832
833 for i in 0..64u64 {
834 let key = format!("col{i}");
835 let sub = core.subscribe_key(key.clone(), |_| {});
836 let held = core.intercept_key(key, Some);
837 drop(sub);
838 held.remove();
839 }
840
841 assert_eq!(core.subscribers_key.len(), 0);
842 assert_eq!(core.interceptors_key.len(), 0);
843 }
844
845 #[test]
846 fn a_key_still_watched_by_someone_keeps_its_entry() {
847 let core = ReactiveMapCore::<String, u64>::new();
848
849 let first = core.subscribe_key("cpu".to_string(), |_| {});
850 let second = core.subscribe_key("cpu".to_string(), |_| {});
851
852 drop(first);
853 assert_eq!(core.subscribers_key.len(), 1);
854
855 drop(second);
856 assert_eq!(core.subscribers_key.len(), 0);
857 }
858}