ferroui_dynamic_data/cache/
testing.rs

1//! The aids for tests that DynamicData ships with the library (`Cache/Tests`, namespace
2//! `DynamicData.Tests`): `ChangeSetAggregator<TObject, TKey>`, and the aggregators of the sorted,
3//! paged, virtual and contextual change sets ([`TypedChangeSetAggregator`]).
4
5use super::{ChangeSet, IObservableCache, ObservableCache, ObservableCacheEx};
6use crate::diagnostics::ChangeSummary;
7use ferroui_reactive::disposables::Disposable;
8use ferroui_reactive::ObservableExt;
9use ferroui_system::{IDisposable, IObservable, ObservableError};
10use std::cell::{Cell, RefCell};
11use std::hash::Hash;
12use std::rc::Rc;
13
14struct State<T, K> {
15    messages: RefCell<Vec<ChangeSet<T, K>>>,
16    error: RefCell<Option<ObservableError>>,
17    is_completed: Cell<bool>,
18    summary: RefCell<ChangeSummary>,
19}
20
21/// Records what a sequence of change sets of a cache produces (`ChangeSetAggregator<TObject,
22/// TKey>`): the change sets, the items they make, their statistics, the error and the end.
23pub struct ChangeSetAggregator<T, K> {
24    state: Rc<State<T, K>>,
25    data: ObservableCache<T, K>,
26    disposer: Rc<dyn IDisposable>,
27}
28
29impl<T: Clone + 'static, K: Eq + Hash + Clone + 'static> ChangeSetAggregator<T, K> {
30    /// Subscribes to `source` (`ChangeSetAggregator(IObservable<IChangeSet>)`).
31    pub fn new(source: Rc<dyn IObservable<ChangeSet<T, K>>>) -> Self {
32        let state = Rc::new(State { messages: RefCell::new(Vec::new()), error: RefCell::new(None), is_completed: Cell::new(false), summary: RefCell::new(ChangeSummary::empty()) });
33        let published = source.publish();
34        let published_source: Rc<dyn IObservable<ChangeSet<T, K>>> = published.clone();
35        let data = published_source.as_observable_cache();
36        let (on_next, on_error, on_completed) = (state.clone(), state.clone(), state.clone());
37        let results = published_source.subscribe_all(
38            move |changes| on_next.messages.borrow_mut().push(changes),
39            move |error| *on_error.error.borrow_mut() = Some(error),
40            move || on_completed.is_completed.set(true),
41        );
42        let summary_target = state.clone();
43        let summariser = published_source.collect_update_stats().subscribe_all(move |summary| *summary_target.summary.borrow_mut() = summary, |_| {}, || {});
44        let connected = published.connect();
45        let cache = data.clone();
46        let disposer = Disposable::create(move || {
47            cache.dispose();
48            connected.dispose();
49            summariser.dispose();
50            results.dispose();
51        });
52        Self { state, data, disposer }
53    }
54
55    /// The items the change sets make (`Data`).
56    pub fn data(&self) -> &ObservableCache<T, K> {
57        &self.data
58    }
59
60    /// The change sets so far (`Messages`).
61    pub fn messages(&self) -> Vec<ChangeSet<T, K>> {
62        self.state.messages.borrow().clone()
63    }
64
65    /// The error of the sequence (`Error`).
66    pub fn error(&self) -> Option<ObservableError> {
67        self.state.error.borrow().clone()
68    }
69
70    /// Whether the sequence has ended (`IsCompleted`).
71    pub fn is_completed(&self) -> bool {
72        self.state.is_completed.get()
73    }
74
75    /// The statistics of the change sets (`Summary`).
76    pub fn summary(&self) -> ChangeSummary {
77        self.state.summary.borrow().clone()
78    }
79
80    /// The number of items the change sets make.
81    pub fn count(&self) -> usize {
82        self.data.count()
83    }
84}
85
86impl<T, K> IDisposable for ChangeSetAggregator<T, K> {
87    fn dispose(&self) {
88        self.disposer.dispose();
89    }
90}
91
92impl<T, K> Drop for ChangeSetAggregator<T, K> {
93    /// `using var results = ...AsAggregator()`: the subscriptions end with the aggregator.
94    fn drop(&mut self) {
95        self.disposer.dispose();
96    }
97}
98
99struct TypedState<C> {
100    messages: RefCell<Vec<C>>,
101    error: RefCell<Option<ObservableError>>,
102    is_completed: Cell<bool>,
103    summary: RefCell<ChangeSummary>,
104}
105
106/// Records what a sequence of change sets of a cache of any form produces: the change sets as
107/// they are, the items they make, their statistics, the error and the end.
108///
109/// It is upstream's `SortedChangeSetAggregator<TObject, TKey>`,
110/// `PagedChangeSetAggregator<TObject, TKey>`, `VirtualChangeSetAggregator<TObject, TKey>` and
111/// `ChangeSetAggregator<TObject, TKey, TContext>`, which differ only by the type of their change
112/// sets (the aliases [`SortedChangeSetAggregator`], [`PagedChangeSetAggregator`],
113/// [`VirtualChangeSetAggregator`] and [`ContextChangeSetAggregator`]).
114pub struct TypedChangeSetAggregator<T, K, C> {
115    state: Rc<TypedState<C>>,
116    data: ObservableCache<T, K>,
117    disposer: Rc<dyn IDisposable>,
118}
119
120/// The aggregator of sorted change sets (`SortedChangeSetAggregator<TObject, TKey>`).
121pub type SortedChangeSetAggregator<T, K> = TypedChangeSetAggregator<T, K, super::SortedChangeSet<T, K>>;
122
123/// The aggregator of paged change sets (`PagedChangeSetAggregator<TObject, TKey>`).
124pub type PagedChangeSetAggregator<T, K> = TypedChangeSetAggregator<T, K, super::PagedChangeSet<T, K>>;
125
126/// The aggregator of virtual change sets (`VirtualChangeSetAggregator<TObject, TKey>`).
127pub type VirtualChangeSetAggregator<T, K> = TypedChangeSetAggregator<T, K, super::VirtualChangeSet<T, K>>;
128
129/// The aggregator of change sets with a context (`ChangeSetAggregator<TObject, TKey, TContext>`).
130pub type ContextChangeSetAggregator<T, K, X> = TypedChangeSetAggregator<T, K, super::ChangeSetWithContext<T, K, X>>;
131
132impl<T: Clone + 'static, K: Eq + Hash + Clone + 'static, C: super::IChangeSet<T, K>> TypedChangeSetAggregator<T, K, C> {
133    /// Subscribes to `source` (`SortedChangeSetAggregator(IObservable<ISortedChangeSet>)` and its
134    /// siblings).
135    pub fn new(source: Rc<dyn IObservable<C>>) -> Self {
136        let state = Rc::new(TypedState { messages: RefCell::new(Vec::new()), error: RefCell::new(None), is_completed: Cell::new(false), summary: RefCell::new(ChangeSummary::empty()) });
137        let published = source.publish();
138        let published_source: Rc<dyn IObservable<C>> = published.clone();
139        let data = ObservableCacheEx::<T, K>::as_observable_cache(&published_source);
140        let (on_next, on_error, on_completed) = (state.clone(), state.clone(), state.clone());
141        let results = published_source.subscribe_all(
142            move |changes| on_next.messages.borrow_mut().push(changes),
143            move |error| *on_error.error.borrow_mut() = Some(error),
144            move || on_completed.is_completed.set(true),
145        );
146        let summary_target = state.clone();
147        let summariser = ObservableCacheEx::<T, K>::collect_update_stats(&published_source).subscribe_all(move |summary| *summary_target.summary.borrow_mut() = summary, |_| {}, || {});
148        let connected = published.connect();
149        let cache = data.clone();
150        let disposer = Disposable::create(move || {
151            cache.dispose();
152            connected.dispose();
153            summariser.dispose();
154            results.dispose();
155        });
156        Self { state, data, disposer }
157    }
158
159    /// The items the change sets make (`Data`).
160    pub fn data(&self) -> &ObservableCache<T, K> {
161        &self.data
162    }
163
164    /// The change sets so far (`Messages`).
165    pub fn messages(&self) -> Vec<C> {
166        self.state.messages.borrow().clone()
167    }
168
169    /// The error of the sequence (`Error`).
170    pub fn error(&self) -> Option<ObservableError> {
171        self.state.error.borrow().clone()
172    }
173
174    /// Whether the sequence has ended (`IsCompleted` of the aggregator of contextual change sets).
175    pub fn is_completed(&self) -> bool {
176        self.state.is_completed.get()
177    }
178
179    /// The statistics of the change sets (`Summary`).
180    pub fn summary(&self) -> ChangeSummary {
181        self.state.summary.borrow().clone()
182    }
183
184    /// The number of items the change sets make.
185    pub fn count(&self) -> usize {
186        self.data.count()
187    }
188}
189
190impl<T, K, C> IDisposable for TypedChangeSetAggregator<T, K, C> {
191    fn dispose(&self) {
192        self.disposer.dispose();
193    }
194}
195
196impl<T, K, C> Drop for TypedChangeSetAggregator<T, K, C> {
197    /// `using var results = ...AsAggregator()`: the subscriptions end with the aggregator.
198    fn drop(&mut self) {
199        self.disposer.dispose();
200    }
201}
202
203struct GroupAggregatorState<T, K, G> {
204    messages: std::cell::RefCell<Vec<super::GroupChangeSet<T, K, G>>>,
205    error: std::cell::RefCell<Option<ferroui_system::ObservableError>>,
206    is_completed: std::cell::Cell<bool>,
207    summary: std::cell::RefCell<crate::diagnostics::ChangeSummary>,
208}
209
210/// Records what a sequence of change sets of groups produces (`GroupChangeSetAggregator<TObject,
211/// TKey, TGroupKey>`): the change sets, the groups they make, an aggregator of the changes of each
212/// group, their statistics, the error and the end.
213pub struct GroupChangeSetAggregator<T, K, G> {
214    state: std::rc::Rc<GroupAggregatorState<T, K, G>>,
215    data: super::ObservableCache<super::Group<T, K, G>, G>,
216    groups: super::ObservableCache<std::rc::Rc<ChangeSetAggregator<T, K>>, G>,
217    disposer: std::rc::Rc<dyn ferroui_system::IDisposable>,
218}
219
220impl<T: Clone + 'static, K: Eq + std::hash::Hash + Clone + 'static, G: Eq + std::hash::Hash + Clone + 'static> GroupChangeSetAggregator<T, K, G> {
221    /// Subscribes to `source` (`GroupChangeSetAggregator(IObservable<IGroupChangeSet>)`).
222    pub fn new(source: std::rc::Rc<dyn ferroui_system::IObservable<super::GroupChangeSet<T, K, G>>>) -> Self {
223        let state = std::rc::Rc::new(GroupAggregatorState {
224            messages: std::cell::RefCell::new(Vec::new()),
225            error: std::cell::RefCell::new(None),
226            is_completed: std::cell::Cell::new(false),
227            summary: std::cell::RefCell::new(crate::diagnostics::ChangeSummary::empty()),
228        });
229        let published = ferroui_reactive::ObservableExt::publish(&source);
230        let published_source: std::rc::Rc<dyn ferroui_system::IObservable<super::GroupChangeSet<T, K, G>>> = published.clone();
231
232        let data = super::ObservableCache::from_source(published_source.clone());
233        // `Transform(grp => grp.Cache.Connect().AsAggregator()).DisposeMany().AsObservableCache()`.
234        let groups = super::ObservableCache::from_source(group_aggregators(published_source.clone()));
235        let (on_next, on_error, on_completed) = (state.clone(), state.clone(), state.clone());
236        let results = ferroui_reactive::ObservableExt::subscribe_all(
237            &published_source,
238            move |changes| on_next.messages.borrow_mut().push(changes),
239            move |error| *on_error.error.borrow_mut() = Some(error),
240            move || on_completed.is_completed.set(true),
241        );
242        let summary_target = state.clone();
243        let summariser = ferroui_reactive::ObservableExt::subscribe_all(
244            &super::ObservableCacheEx::collect_update_stats(&published_source),
245            move |summary| *summary_target.summary.borrow_mut() = summary,
246            |_| {},
247            || {},
248        );
249        let connected = published.connect();
250
251        let (data_cache, groups_cache) = (data.clone(), groups.clone());
252        let disposer = ferroui_reactive::disposables::Disposable::create(move || {
253            ferroui_system::IDisposable::dispose(&data_cache);
254            ferroui_system::IDisposable::dispose(&groups_cache);
255            results.dispose();
256            summariser.dispose();
257            connected.dispose();
258        });
259        Self { state, data, groups, disposer }
260    }
261
262    /// The groups the change sets make (`Data`).
263    pub fn data(&self) -> &super::ObservableCache<super::Group<T, K, G>, G> {
264        &self.data
265    }
266
267    /// An aggregator of the changes of the cache of each group, by group key (`Groups`).
268    pub fn groups(&self) -> &super::ObservableCache<std::rc::Rc<ChangeSetAggregator<T, K>>, G> {
269        &self.groups
270    }
271
272    /// The change sets so far (`Messages`).
273    pub fn messages(&self) -> Vec<super::GroupChangeSet<T, K, G>> {
274        self.state.messages.borrow().clone()
275    }
276
277    /// The error of the sequence (`Error`).
278    pub fn error(&self) -> Option<ferroui_system::ObservableError> {
279        self.state.error.borrow().clone()
280    }
281
282    /// Whether the sequence has ended (`IsCompleted`).
283    pub fn is_completed(&self) -> bool {
284        self.state.is_completed.get()
285    }
286
287    /// The statistics of the change sets (`Summary`).
288    pub fn summary(&self) -> crate::diagnostics::ChangeSummary {
289        self.state.summary.borrow().clone()
290    }
291}
292
293impl<T, K, G> ferroui_system::IDisposable for GroupChangeSetAggregator<T, K, G> {
294    fn dispose(&self) {
295        self.disposer.dispose();
296    }
297}
298
299impl<T, K, G> Drop for GroupChangeSetAggregator<T, K, G> {
300    /// `using var results = ...AsAggregator()`: the subscriptions end with the aggregator.
301    fn drop(&mut self) {
302        self.disposer.dispose();
303    }
304}
305
306/// An aggregator of the changes of the cache of each group, disposed when its group is removed or
307/// replaced, and when the subscription ends (the `Groups` of `GroupChangeSetAggregator`).
308fn group_aggregators<T: Clone + 'static, K: Eq + std::hash::Hash + Clone + 'static, G: Eq + std::hash::Hash + Clone + 'static>(
309    source: std::rc::Rc<dyn ferroui_system::IObservable<super::GroupChangeSet<T, K, G>>>,
310) -> std::rc::Rc<dyn ferroui_system::IObservable<super::ChangeSet<std::rc::Rc<ChangeSetAggregator<T, K>>, G>>> {
311    ferroui_reactive::Observable::create(move |observer: std::rc::Rc<dyn ferroui_system::IObserver<super::ChangeSet<std::rc::Rc<ChangeSetAggregator<T, K>>, G>>>| {
312        let aggregators = std::rc::Rc::new(std::cell::RefCell::new(super::ChangeAwareCache::<std::rc::Rc<ChangeSetAggregator<T, K>>, G>::new()));
313        let (target, on_error, on_completed) = (aggregators.clone(), observer.clone(), observer.clone());
314        let subscription = ferroui_reactive::ObservableExt::subscribe_all(
315            &source,
316            move |changes: super::GroupChangeSet<T, K, G>| {
317                let mut replaced = Vec::new();
318                let output = {
319                    let mut cache = target.borrow_mut();
320                    for change in &changes {
321                        match change.reason() {
322                            super::ChangeReason::Add | super::ChangeReason::Update => {
323                                let connected = super::IConnectableCache::connect(change.current().observable_cache());
324                                let aggregator = std::rc::Rc::new(ChangeSetAggregator::new(connected));
325                                replaced.extend(super::IQuery::lookup(&*cache, change.key()));
326                                super::ICache::add_or_update(&mut *cache, aggregator, change.key().clone());
327                            }
328                            super::ChangeReason::Remove => {
329                                replaced.extend(super::IQuery::lookup(&*cache, change.key()));
330                                super::ICache::remove(&mut *cache, change.key());
331                            }
332                            super::ChangeReason::Refresh => super::ICache::refresh(&mut *cache, change.key()),
333                            super::ChangeReason::Moved => {}
334                        }
335                    }
336                    cache.capture_changes()
337                };
338                observer.on_next(output);
339                for aggregator in replaced {
340                    ferroui_system::IDisposable::dispose(&*aggregator);
341                }
342            },
343            move |error| on_error.on_error(error),
344            move || on_completed.on_completed(),
345        );
346        ferroui_reactive::disposables::Disposable::create(move || {
347            subscription.dispose();
348            let remaining = super::IQuery::items(&*aggregators.borrow());
349            for aggregator in remaining {
350                ferroui_system::IDisposable::dispose(&*aggregator);
351            }
352        })
353    })
354}
355
356/// `AsAggregator` of a sequence of change sets of groups (`TestEx.AsAggregator(IObservable<IGroupChangeSet>)`).
357///
358/// Deviation (DEVIATIONS.md, DynamicData): an overload of `AsAggregator` by the type of the change
359/// set; here it is `as_group_aggregator`, of a trait of its own, as `as_aggregator` of
360/// `ObservableCacheEx` applies to sequences of groups too.
361pub trait GroupChangeSetAggregatorEx<T, K, G> {
362    /// Records the change sets of groups, for tests.
363    fn as_group_aggregator(&self) -> GroupChangeSetAggregator<T, K, G>;
364}
365
366impl<T: Clone + 'static, K: Eq + std::hash::Hash + Clone + 'static, G: Eq + std::hash::Hash + Clone + 'static> GroupChangeSetAggregatorEx<T, K, G>
367    for std::rc::Rc<dyn ferroui_system::IObservable<super::GroupChangeSet<T, K, G>>>
368{
369    fn as_group_aggregator(&self) -> GroupChangeSetAggregator<T, K, G> {
370        GroupChangeSetAggregator::new(self.clone())
371    }
372}
ferroui_dynamic_data/cache/testing.rs

ferroui_dynamic_data/cache/
testing.rs
