ferroui_dynamic_data/
observable_change_set.rs

1//! `ObservableChangeSet` (`ObservableChangeSet.cs`): sequences of change sets made by a function
2//! that fills a cache or a list of their own, as `Observable.Create` makes a sequence of a
3//! function that notifies an observer.
4//!
5//! Each subscription creates a `SourceCache` (keyed by `key_selector`) or a `SourceList`, passes it
6//! to `subscribe`, and delivers its change sets. Disposing the subscription completes the
7//! sequence, disposes the cache or list, and releases what `subscribe` returned: a disposable or
8//! an action.
9//!
10//! The overloads of upstream map to the names of the Rx port's `Observable::create` forms:
11//! - `Func<ISourceCache, IDisposable>` is `create_cache`, `Func<ISourceCache, Action>` is
12//!   `create_cache_with_action`; their `try_` forms take a function that can fail, whose error ends
13//!   the sequence (the exception that upstream catches).
14//! - `Func<ISourceCache, Task>` is `create_cache_async`, `Func<ISourceCache, Task<IDisposable>>`
15//!   is `create_cache_async_with_disposable`, `Func<ISourceCache, Task<Action>>` is
16//!   `create_cache_async_with_action`; the `CancellationToken` forms take the suffix `_with_token`
17//!   (the token is cancelled when the subscription is disposed). The function returns a future,
18//!   which runs as a task (docs/porting/reactive.md, section 5.8); its error ends the sequence.
19//! - The same with `list` for `ISourceList<T>`.
20//!
21//! The function receives the `SourceCache` or `SourceList` itself, a handle to the collection,
22//! where upstream passes the interface `ISourceCache<TObject, TKey>` or `ISourceList<T>`.
23
24use crate::cache::{ChangeSet, IConnectableCache, SourceCache};
25use crate::list::{IObservableList, SourceList};
26use ferroui_reactive::disposables::Disposable;
27use ferroui_reactive::{Observable, ObservableExt};
28use ferroui_system::threading::CancellationToken;
29use ferroui_system::{IDisposable, IObservable, IObserver, ObservableError};
30use std::future::Future;
31use std::hash::Hash;
32use std::rc::Rc;
33
34/// The factories of sequences of change sets (`ObservableChangeSet`).
35pub struct ObservableChangeSet;
36
37/// The release of a subscription of an asynchronous form: the action or the disposable that the
38/// task of `subscribe` gave, if any.
39fn release(resource: Option<Rc<dyn IDisposable>>) {
40    if let Some(resource) = resource {
41        resource.dispose();
42    }
43}
44
45impl ObservableChangeSet {
46    // The forms of caches.
47
48    /// A sequence of the change sets of a cache that `subscribe` fills; the disposable it returns
49    /// is disposed with the subscription
50    /// (`Create<TObject, TKey>(Func<ISourceCache<TObject, TKey>, IDisposable>, Func<TObject, TKey>)`).
51    pub fn create_cache<T: Clone + 'static, K: Eq + Hash + Clone + 'static>(
52        subscribe: impl Fn(SourceCache<T, K>) -> Rc<dyn IDisposable> + 'static,
53        key_selector: impl Fn(&T) -> K + 'static,
54    ) -> Rc<dyn IObservable<ChangeSet<T, K>>> {
55        Self::try_create_cache(move |cache| Ok(subscribe(cache)), key_selector)
56    }
57
58    /// [`create_cache`](Self::create_cache) with a function that can fail: its error ends the
59    /// sequence (the exception that upstream catches and delivers with `OnError`).
60    pub fn try_create_cache<T: Clone + 'static, K: Eq + Hash + Clone + 'static>(
61        subscribe: impl Fn(SourceCache<T, K>) -> Result<Rc<dyn IDisposable>, ObservableError> + 'static,
62        key_selector: impl Fn(&T) -> K + 'static,
63    ) -> Rc<dyn IObservable<ChangeSet<T, K>>> {
64        let key_selector: Rc<dyn Fn(&T) -> K> = Rc::new(key_selector);
65        Observable::create(move |observer: Rc<dyn IObserver<ChangeSet<T, K>>>| {
66            let key_selector = key_selector.clone();
67            let cache = SourceCache::new(move |item: &T| key_selector(item));
68
69            let disposable = match subscribe(cache.clone()) {
70                Ok(disposable) => Some(disposable),
71                Err(error) => {
72                    observer.on_error(error);
73                    None
74                }
75            };
76
77            let completion = observer.clone();
78            let connection = cache.connect().subscribe_safe(observer);
79            Disposable::create(move || {
80                release(disposable);
81                completion.on_completed();
82                connection.dispose();
83                cache.dispose();
84            })
85        })
86    }
87
88    /// A sequence of the change sets of a cache that `subscribe` fills; the action it returns runs
89    /// when the subscription is disposed
90    /// (`Create<TObject, TKey>(Func<ISourceCache<TObject, TKey>, Action>, Func<TObject, TKey>)`).
91    pub fn create_cache_with_action<T: Clone + 'static, K: Eq + Hash + Clone + 'static, A: FnOnce() + 'static>(
92        subscribe: impl Fn(SourceCache<T, K>) -> A + 'static,
93        key_selector: impl Fn(&T) -> K + 'static,
94    ) -> Rc<dyn IObservable<ChangeSet<T, K>>> {
95        Self::try_create_cache(move |cache| Ok(Disposable::create(subscribe(cache))), key_selector)
96    }
97
98    /// [`create_cache_with_action`](Self::create_cache_with_action) with a function that can fail:
99    /// its error ends the sequence.
100    pub fn try_create_cache_with_action<T: Clone + 'static, K: Eq + Hash + Clone + 'static, A: FnOnce() + 'static>(
101        subscribe: impl Fn(SourceCache<T, K>) -> Result<A, ObservableError> + 'static,
102        key_selector: impl Fn(&T) -> K + 'static,
103    ) -> Rc<dyn IObservable<ChangeSet<T, K>>> {
104        Self::try_create_cache(move |cache| subscribe(cache).map(|action| Disposable::create(action)), key_selector)
105    }
106
107    /// The asynchronous form whose task gives the disposable released with the subscription
108    /// (`Create<TObject, TKey>(Func<ISourceCache<TObject, TKey>, CancellationToken,
109    /// Task<IDisposable>>, Func<TObject, TKey>)`): the change sets are delivered from the start,
110    /// and an error of the task ends the sequence.
111    pub fn create_cache_async_with_disposable_with_token<
112        T: Clone + 'static,
113        K: Eq + Hash + Clone + 'static,
114        Fut: Future<Output = Result<Rc<dyn IDisposable>, ObservableError>> + 'static,
115    >(
116        subscribe: impl Fn(SourceCache<T, K>, CancellationToken) -> Fut + 'static,
117        key_selector: impl Fn(&T) -> K + 'static,
118    ) -> Rc<dyn IObservable<ChangeSet<T, K>>> {
119        let key_selector: Rc<dyn Fn(&T) -> K> = Rc::new(key_selector);
120        Observable::create_async_with_disposable_with_token(move |observer: Rc<dyn IObserver<ChangeSet<T, K>>>, token: CancellationToken| {
121            let key_selector = key_selector.clone();
122            let cache = SourceCache::new(move |item: &T| key_selector(item));
123            let responder = cache.connect().subscribe_safe(observer.clone());
124            let future = subscribe(cache.clone(), token);
125            async move {
126                let disposable = match future.await {
127                    Ok(disposable) => Some(disposable),
128                    Err(error) => {
129                        observer.on_error(error);
130                        None
131                    }
132                };
133
134                Ok::<Rc<dyn IDisposable>, ObservableError>(Disposable::create(move || {
135                    responder.dispose();
136                    cache.dispose();
137                    release(disposable);
138                    observer.on_completed();
139                }))
140            }
141        })
142    }
143
144    /// [`create_cache_async_with_disposable_with_token`](Self::create_cache_async_with_disposable_with_token)
145    /// without a token (`Create<TObject, TKey>(Func<ISourceCache<TObject, TKey>, Task<IDisposable>>,
146    /// Func<TObject, TKey>)`).
147    pub fn create_cache_async_with_disposable<
148        T: Clone + 'static,
149        K: Eq + Hash + Clone + 'static,
150        Fut: Future<Output = Result<Rc<dyn IDisposable>, ObservableError>> + 'static,
151    >(
152        subscribe: impl Fn(SourceCache<T, K>) -> Fut + 'static,
153        key_selector: impl Fn(&T) -> K + 'static,
154    ) -> Rc<dyn IObservable<ChangeSet<T, K>>> {
155        Self::create_cache_async_with_disposable_with_token(move |cache, _token| subscribe(cache), key_selector)
156    }
157
158    /// The asynchronous form whose task gives the action run with the release of the subscription
159    /// (`Create<TObject, TKey>(Func<ISourceCache<TObject, TKey>, CancellationToken, Task<Action>>,
160    /// Func<TObject, TKey>)`): disposing the subscription completes the sequence, then runs the
161    /// action.
162    pub fn create_cache_async_with_action_with_token<
163        T: Clone + 'static,
164        K: Eq + Hash + Clone + 'static,
165        A: FnOnce() + 'static,
166        Fut: Future<Output = Result<A, ObservableError>> + 'static,
167    >(
168        subscribe: impl Fn(SourceCache<T, K>, CancellationToken) -> Fut + 'static,
169        key_selector: impl Fn(&T) -> K + 'static,
170    ) -> Rc<dyn IObservable<ChangeSet<T, K>>> {
171        let key_selector: Rc<dyn Fn(&T) -> K> = Rc::new(key_selector);
172        Observable::create_async_with_disposable_with_token(move |observer: Rc<dyn IObserver<ChangeSet<T, K>>>, token: CancellationToken| {
173            let key_selector = key_selector.clone();
174            let cache = SourceCache::new(move |item: &T| key_selector(item));
175            let responder = cache.connect().subscribe_safe(observer.clone());
176            let future = subscribe(cache.clone(), token);
177            async move {
178                let dispose_action = match future.await {
179                    Ok(action) => Some(action),
180                    Err(error) => {
181                        observer.on_error(error);
182                        None
183                    }
184                };
185
186                Ok::<Rc<dyn IDisposable>, ObservableError>(Disposable::create(move || {
187                    responder.dispose();
188                    cache.dispose();
189                    observer.on_completed();
190                    if let Some(dispose_action) = dispose_action {
191                        dispose_action();
192                    }
193                }))
194            }
195        })
196    }
197
198    /// [`create_cache_async_with_action_with_token`](Self::create_cache_async_with_action_with_token)
199    /// without a token (`Create<TObject, TKey>(Func<ISourceCache<TObject, TKey>, Task<Action>>,
200    /// Func<TObject, TKey>)`).
201    pub fn create_cache_async_with_action<
202        T: Clone + 'static,
203        K: Eq + Hash + Clone + 'static,
204        A: FnOnce() + 'static,
205        Fut: Future<Output = Result<A, ObservableError>> + 'static,
206    >(
207        subscribe: impl Fn(SourceCache<T, K>) -> Fut + 'static,
208        key_selector: impl Fn(&T) -> K + 'static,
209    ) -> Rc<dyn IObservable<ChangeSet<T, K>>> {
210        Self::create_cache_async_with_action_with_token(move |cache, _token| subscribe(cache), key_selector)
211    }
212
213    /// The asynchronous form whose task only fills the cache
214    /// (`Create<TObject, TKey>(Func<ISourceCache<TObject, TKey>, CancellationToken, Task>,
215    /// Func<TObject, TKey>)`): disposing the subscription completes the sequence.
216    pub fn create_cache_async_with_token<T: Clone + 'static, K: Eq + Hash + Clone + 'static, Fut: Future<Output = Result<(), ObservableError>> + 'static>(
217        subscribe: impl Fn(SourceCache<T, K>, CancellationToken) -> Fut + 'static,
218        key_selector: impl Fn(&T) -> K + 'static,
219    ) -> Rc<dyn IObservable<ChangeSet<T, K>>> {
220        let key_selector: Rc<dyn Fn(&T) -> K> = Rc::new(key_selector);
221        Observable::create_async_with_disposable_with_token(move |observer: Rc<dyn IObserver<ChangeSet<T, K>>>, token: CancellationToken| {
222            let key_selector = key_selector.clone();
223            let cache = SourceCache::new(move |item: &T| key_selector(item));
224            let responder = cache.connect().subscribe_safe(observer.clone());
225            let future = subscribe(cache.clone(), token);
226            async move {
227                if let Err(error) = future.await {
228                    observer.on_error(error);
229                }
230
231                Ok::<Rc<dyn IDisposable>, ObservableError>(Disposable::create(move || {
232                    responder.dispose();
233                    cache.dispose();
234                    observer.on_completed();
235                }))
236            }
237        })
238    }
239
240    /// [`create_cache_async_with_token`](Self::create_cache_async_with_token) without a token
241    /// (`Create<TObject, TKey>(Func<ISourceCache<TObject, TKey>, Task>, Func<TObject, TKey>)`).
242    pub fn create_cache_async<T: Clone + 'static, K: Eq + Hash + Clone + 'static, Fut: Future<Output = Result<(), ObservableError>> + 'static>(
243        subscribe: impl Fn(SourceCache<T, K>) -> Fut + 'static,
244        key_selector: impl Fn(&T) -> K + 'static,
245    ) -> Rc<dyn IObservable<ChangeSet<T, K>>> {
246        Self::create_cache_async_with_token(move |cache, _token| subscribe(cache), key_selector)
247    }
248
249    // The forms of lists.
250
251    /// A sequence of the change sets of a list that `subscribe` fills; the disposable it returns is
252    /// disposed with the subscription, after the sequence completes
253    /// (`Create<T>(Func<ISourceList<T>, IDisposable>)`).
254    pub fn create_list<T: Clone + PartialEq + 'static>(
255        subscribe: impl Fn(SourceList<T>) -> Rc<dyn IDisposable> + 'static,
256    ) -> Rc<dyn IObservable<crate::list::ChangeSet<T>>> {
257        Self::try_create_list(move |list| Ok(subscribe(list)))
258    }
259
260    /// [`create_list`](Self::create_list) with a function that can fail: its error ends the
261    /// sequence.
262    pub fn try_create_list<T: Clone + PartialEq + 'static>(
263        subscribe: impl Fn(SourceList<T>) -> Result<Rc<dyn IDisposable>, ObservableError> + 'static,
264    ) -> Rc<dyn IObservable<crate::list::ChangeSet<T>>> {
265        Observable::create(move |observer: Rc<dyn IObserver<crate::list::ChangeSet<T>>>| {
266            let list = SourceList::new();
267
268            let dispose_action = match subscribe(list.clone()) {
269                Ok(disposable) => Some(disposable),
270                Err(error) => {
271                    observer.on_error(error);
272                    None
273                }
274            };
275
276            let completion = observer.clone();
277            let connection = list.connect().subscribe_safe(observer);
278            Disposable::create(move || {
279                connection.dispose();
280                list.dispose();
281                completion.on_completed();
282                release(dispose_action);
283            })
284        })
285    }
286
287    /// A sequence of the change sets of a list that `subscribe` fills; the action it returns runs
288    /// when the subscription is disposed (`Create<T>(Func<ISourceList<T>, Action>)`).
289    pub fn create_list_with_action<T: Clone + PartialEq + 'static, A: FnOnce() + 'static>(
290        subscribe: impl Fn(SourceList<T>) -> A + 'static,
291    ) -> Rc<dyn IObservable<crate::list::ChangeSet<T>>> {
292        Self::try_create_list(move |list| Ok(Disposable::create(subscribe(list))))
293    }
294
295    /// [`create_list_with_action`](Self::create_list_with_action) with a function that can fail:
296    /// its error ends the sequence.
297    pub fn try_create_list_with_action<T: Clone + PartialEq + 'static, A: FnOnce() + 'static>(
298        subscribe: impl Fn(SourceList<T>) -> Result<A, ObservableError> + 'static,
299    ) -> Rc<dyn IObservable<crate::list::ChangeSet<T>>> {
300        Self::try_create_list(move |list| subscribe(list).map(|action| Disposable::create(action)))
301    }
302
303    /// The asynchronous form whose task gives the disposable released with the subscription
304    /// (`Create<T>(Func<ISourceList<T>, CancellationToken, Task<IDisposable>>)`).
305    pub fn create_list_async_with_disposable_with_token<
306        T: Clone + PartialEq + 'static,
307        Fut: Future<Output = Result<Rc<dyn IDisposable>, ObservableError>> + 'static,
308    >(
309        subscribe: impl Fn(SourceList<T>, CancellationToken) -> Fut + 'static,
310    ) -> Rc<dyn IObservable<crate::list::ChangeSet<T>>> {
311        Observable::create_async_with_disposable_with_token(move |observer: Rc<dyn IObserver<crate::list::ChangeSet<T>>>, token: CancellationToken| {
312            let list = SourceList::new();
313            let responder = list.connect().subscribe_safe(observer.clone());
314            let future = subscribe(list.clone(), token);
315            async move {
316                let dispose_action = match future.await {
317                    Ok(disposable) => Some(disposable),
318                    Err(error) => {
319                        observer.on_error(error);
320                        None
321                    }
322                };
323
324                Ok::<Rc<dyn IDisposable>, ObservableError>(Disposable::create(move || {
325                    responder.dispose();
326                    list.dispose();
327                    observer.on_completed();
328                    release(dispose_action);
329                }))
330            }
331        })
332    }
333
334    /// [`create_list_async_with_disposable_with_token`](Self::create_list_async_with_disposable_with_token)
335    /// without a token (`Create<T>(Func<ISourceList<T>, Task<IDisposable>>)`).
336    pub fn create_list_async_with_disposable<T: Clone + PartialEq + 'static, Fut: Future<Output = Result<Rc<dyn IDisposable>, ObservableError>> + 'static>(
337        subscribe: impl Fn(SourceList<T>) -> Fut + 'static,
338    ) -> Rc<dyn IObservable<crate::list::ChangeSet<T>>> {
339        Self::create_list_async_with_disposable_with_token(move |list, _token| subscribe(list))
340    }
341
342    /// The asynchronous form whose task gives the action run with the release of the subscription
343    /// (`Create<T>(Func<ISourceList<T>, CancellationToken, Task<Action>>)`).
344    pub fn create_list_async_with_action_with_token<
345        T: Clone + PartialEq + 'static,
346        A: FnOnce() + 'static,
347        Fut: Future<Output = Result<A, ObservableError>> + 'static,
348    >(
349        subscribe: impl Fn(SourceList<T>, CancellationToken) -> Fut + 'static,
350    ) -> Rc<dyn IObservable<crate::list::ChangeSet<T>>> {
351        Observable::create_async_with_disposable_with_token(move |observer: Rc<dyn IObserver<crate::list::ChangeSet<T>>>, token: CancellationToken| {
352            let list = SourceList::new();
353            let responder = list.connect().subscribe_safe(observer.clone());
354            let future = subscribe(list.clone(), token);
355            async move {
356                let dispose_action = match future.await {
357                    Ok(action) => Some(action),
358                    Err(error) => {
359                        observer.on_error(error);
360                        None
361                    }
362                };
363
364                Ok::<Rc<dyn IDisposable>, ObservableError>(Disposable::create(move || {
365                    responder.dispose();
366                    list.dispose();
367                    observer.on_completed();
368                    if let Some(dispose_action) = dispose_action {
369                        dispose_action();
370                    }
371                }))
372            }
373        })
374    }
375
376    /// [`create_list_async_with_action_with_token`](Self::create_list_async_with_action_with_token)
377    /// without a token (`Create<T>(Func<ISourceList<T>, Task<Action>>)`).
378    pub fn create_list_async_with_action<T: Clone + PartialEq + 'static, A: FnOnce() + 'static, Fut: Future<Output = Result<A, ObservableError>> + 'static>(
379        subscribe: impl Fn(SourceList<T>) -> Fut + 'static,
380    ) -> Rc<dyn IObservable<crate::list::ChangeSet<T>>> {
381        Self::create_list_async_with_action_with_token(move |list, _token| subscribe(list))
382    }
383
384    /// The asynchronous form whose task only fills the list
385    /// (`Create<T>(Func<ISourceList<T>, Task>)`): disposing the subscription completes the
386    /// sequence.
387    pub fn create_list_async<T: Clone + PartialEq + 'static, Fut: Future<Output = Result<(), ObservableError>> + 'static>(
388        subscribe: impl Fn(SourceList<T>) -> Fut + 'static,
389    ) -> Rc<dyn IObservable<crate::list::ChangeSet<T>>> {
390        Observable::create_async_with_disposable(move |observer: Rc<dyn IObserver<crate::list::ChangeSet<T>>>| {
391            let list = SourceList::new();
392            let responder = list.connect().subscribe_safe(observer.clone());
393            let future = subscribe(list.clone());
394            async move {
395                if let Err(error) = future.await {
396                    observer.on_error(error);
397                }
398
399                Ok::<Rc<dyn IDisposable>, ObservableError>(Disposable::create(move || {
400                    responder.dispose();
401                    list.dispose();
402                    observer.on_completed();
403                }))
404            }
405        })
406    }
407
408    /// The asynchronous form whose task only fills the list, with a token
409    /// (`Create<T>(Func<ISourceList<T>, CancellationToken, Task>)`): unlike the other forms, the
410    /// change sets are delivered once the task has ended, as upstream subscribes to the list after
411    /// awaiting it.
412    pub fn create_list_async_with_token<T: Clone + PartialEq + 'static, Fut: Future<Output = Result<(), ObservableError>> + 'static>(
413        subscribe: impl Fn(SourceList<T>, CancellationToken) -> Fut + 'static,
414    ) -> Rc<dyn IObservable<crate::list::ChangeSet<T>>> {
415        Observable::create_async_with_disposable_with_token(move |observer: Rc<dyn IObserver<crate::list::ChangeSet<T>>>, token: CancellationToken| {
416            let list = SourceList::new();
417            let future = subscribe(list.clone(), token);
418            async move {
419                if let Err(error) = future.await {
420                    observer.on_error(error);
421                }
422
423                let connection = list.connect().subscribe_safe(observer.clone());
424                Ok::<Rc<dyn IDisposable>, ObservableError>(Disposable::create(move || {
425                    connection.dispose();
426                    list.dispose();
427                    observer.on_completed();
428                }))
429            }
430        })
431    }
432}
ferroui_dynamic_data/observable_change_set.rs

ferroui_dynamic_data/
observable_change_set.rs
