ferroui_dynamic_data/cache/
lock_free_observable_cache.rs

1//! `LockFreeObservableCache<TObject, TKey>` (`Cache/Internal/LockFreeObservableCache.cs`, public
2//! upstream), with its edit convenience (`ObservableCacheEx.Clear(LockFreeObservableCache)`).
3
4use super::internal::{filter_static, CacheUpdater};
5use super::{Change, ChangeAwareCache, ChangeReason, ChangeSet, ICache, ICacheUpdater, IChangeSet, IConnectableCache, IObservableCache, IQuery, ObservableCacheEx};
6use crate::kernel::return_with;
7use ferroui_reactive::disposables::Disposable;
8use ferroui_reactive::subjects::Subject;
9use ferroui_reactive::{Observable, ObservableExt};
10use ferroui_system::collections::generic::Dictionary;
11use ferroui_system::{IDisposable, IObservable, IObserver};
12use std::cell::RefCell;
13use std::hash::Hash;
14use std::rc::Rc;
15
16struct Inner<T, K> {
17    changes: Rc<Subject<ChangeSet<T, K>>>,
18    changes_preview: Rc<Subject<ChangeSet<T, K>>>,
19    count_changed: Rc<Subject<usize>>,
20    inner_cache: Rc<RefCell<ChangeAwareCache<T, K>>>,
21    clean_up: RefCell<Option<Rc<dyn IDisposable>>>,
22}
23
24/// An observable cache for a single writer (`LockFreeObservableCache<TObject, TKey>`): an edit
25/// delivers its change set as soon as it ends. A clone is a handle to the same cache.
26///
27/// As upstream's, the count and the preview are never notified after the subscription: the
28/// count is the number of items when it is observed, and the preview never has a change set.
29pub struct LockFreeObservableCache<T, K> {
30    inner: Rc<Inner<T, K>>,
31}
32
33impl<T, K> Clone for LockFreeObservableCache<T, K> {
34    fn clone(&self) -> Self {
35        Self { inner: self.inner.clone() }
36    }
37}
38
39impl<T: Clone + 'static, K: Eq + Hash + Clone + 'static> Default for LockFreeObservableCache<T, K> {
40    fn default() -> Self {
41        Self::new()
42    }
43}
44
45impl<T: Clone + 'static, K: Eq + Hash + Clone + 'static> LockFreeObservableCache<T, K> {
46    fn create() -> Self {
47        Self {
48            inner: Rc::new(Inner {
49                changes: Subject::new(),
50                changes_preview: Subject::new(),
51                count_changed: Subject::new(),
52                inner_cache: Rc::new(RefCell::new(ChangeAwareCache::new())),
53                clean_up: RefCell::new(None),
54            }),
55        }
56    }
57
58    /// An empty cache (`LockFreeObservableCache()`).
59    pub fn new() -> Self {
60        let cache = Self::create();
61        let (changes, count_changed) = (cache.inner.changes.clone(), cache.inner.count_changed.clone());
62        *cache.inner.clean_up.borrow_mut() = Some(Disposable::create(move || {
63            changes.on_completed();
64            count_changed.on_completed();
65        }));
66        cache
67    }
68
69    /// A cache that applies the change sets of `source` (`LockFreeObservableCache(IObservable<IChangeSet>)`).
70    pub fn from_source<C: IChangeSet<T, K>>(source: Rc<dyn IObservable<C>>) -> Self {
71        let cache = Self::create();
72        let inner_cache = cache.inner.inner_cache.clone();
73        let loader = source
74            .change_sets()
75            .select(move |changes: ChangeSet<T, K>| {
76                let mut inner_cache = inner_cache.borrow_mut();
77                inner_cache.clone_changes(&changes);
78                inner_cache.capture_changes()
79            })
80            .subscribe_safe(cache.inner.changes.as_observer());
81        let (changes_preview, changes, count_changed) = (cache.inner.changes_preview.clone(), cache.inner.changes.clone(), cache.inner.count_changed.clone());
82        *cache.inner.clean_up.borrow_mut() = Some(Disposable::create(move || {
83            loader.dispose();
84            changes_preview.on_completed();
85            changes.on_completed();
86            count_changed.on_completed();
87        }));
88        cache
89    }
90
91    /// Runs an edit, then delivers its changes (`Edit(Action<ICacheUpdater>)`).
92    pub fn edit(&self, edit_action: impl FnOnce(&mut dyn ICacheUpdater<T, K>)) {
93        let changes = {
94            let cache: &RefCell<dyn ICache<T, K>> = &*self.inner.inner_cache;
95            let mut updater = CacheUpdater::new(cache, None);
96            edit_action(&mut updater);
97            self.inner.inner_cache.borrow_mut().capture_changes()
98        };
99        self.inner.changes.on_next(changes);
100    }
101
102    /// Removes every item (`Clear(LockFreeObservableCache)`).
103    pub fn clear(&self) {
104        self.edit(|updater| updater.clear());
105    }
106
107    fn get_initial_updates(inner_cache: &RefCell<ChangeAwareCache<T, K>>, predicate: Option<&dyn Fn(&T) -> bool>) -> ChangeSet<T, K> {
108        inner_cache
109            .borrow()
110            .key_values()
111            .into_iter()
112            .filter(|(_, item)| predicate.is_none_or(|predicate| predicate(item)))
113            .map(|(key, item)| Change::new(ChangeReason::Add, key, item))
114            .collect()
115    }
116}
117
118impl<T: Clone + 'static, K: Eq + Hash + Clone + 'static> IConnectableCache<T, K> for LockFreeObservableCache<T, K> {
119    fn count_changed(&self) -> Rc<dyn IObservable<usize>> {
120        let count = self.inner.inner_cache.borrow().count();
121        self.inner.count_changed.as_observable().start_with([count]).distinct_until_changed()
122    }
123
124    fn connect_with(&self, predicate: Option<Rc<dyn Fn(&T) -> bool>>, suppress_empty_change_sets: bool) -> Rc<dyn IObservable<ChangeSet<T, K>>> {
125        let (inner_cache, changes) = (self.inner.inner_cache.clone(), self.inner.changes.clone());
126        Observable::defer(move || {
127            let (inner_cache, filter) = (inner_cache.clone(), predicate.clone());
128            let initial = return_with(move || Self::get_initial_updates(&inner_cache, filter.as_deref()));
129            let changes = initial.concat(changes.as_observable());
130            match &predicate {
131                Some(predicate) => filter_static(changes, predicate.clone(), suppress_empty_change_sets),
132                None if suppress_empty_change_sets => changes.not_empty(),
133                None => changes,
134            }
135        })
136    }
137
138    fn preview_with(&self, predicate: Option<Rc<dyn Fn(&T) -> bool>>) -> Rc<dyn IObservable<ChangeSet<T, K>>> {
139        let preview = self.inner.changes_preview.as_observable();
140        match predicate {
141            None => preview,
142            Some(predicate) => filter_static(preview, predicate, true),
143        }
144    }
145
146    fn watch(&self, key: K) -> Rc<dyn IObservable<Change<T, K>>> {
147        let (inner_cache, changes) = (self.inner.inner_cache.clone(), self.inner.changes.clone());
148        Observable::create(move |observer: Rc<dyn IObserver<Change<T, K>>>| {
149            let initial = inner_cache.borrow().lookup(&key);
150            if let Some(initial) = initial {
151                observer.on_next(Change::new(ChangeReason::Add, key.clone(), initial));
152            }
153            let key = key.clone();
154            changes.as_observable().subscribe_next(move |changes: ChangeSet<T, K>| {
155                for change in &changes {
156                    if *change.key() == key {
157                        observer.on_next(change.clone());
158                    }
159                }
160            })
161        })
162    }
163}
164
165impl<T: Clone + 'static, K: Eq + Hash + Clone + 'static> IObservableCache<T, K> for LockFreeObservableCache<T, K> {
166    fn count(&self) -> usize {
167        self.inner.inner_cache.borrow().count()
168    }
169
170    fn items(&self) -> Vec<T> {
171        self.inner.inner_cache.borrow().items()
172    }
173
174    fn keys(&self) -> Vec<K> {
175        self.inner.inner_cache.borrow().keys()
176    }
177
178    fn key_values(&self) -> Dictionary<K, T> {
179        self.inner.inner_cache.borrow().dictionary().clone()
180    }
181
182    fn lookup(&self, key: &K) -> Option<T> {
183        self.inner.inner_cache.borrow().lookup(key)
184    }
185}
186
187impl<T: Clone + 'static, K: Eq + Hash + Clone + 'static> IDisposable for LockFreeObservableCache<T, K> {
188    fn dispose(&self) {
189        let clean_up = self.inner.clean_up.borrow_mut().take();
190        if let Some(clean_up) = clean_up {
191            clean_up.dispose();
192        }
193    }
194}
ferroui_dynamic_data/cache/lock_free_observable_cache.rs

ferroui_dynamic_data/cache/
lock_free_observable_cache.rs
