ferroui_dynamic_data/list/
source_list.rs

1//! `SourceList<T>`, `IObservableList<T>` and `ISourceList<T>` (`List/SourceList.cs`,
2//! `List/IObservableList.cs`, `List/ISourceList.cs`).
3
4use super::internal::{filter_static, ReaderWriter};
5use super::{Change, ChangeSet, IExtendedList, ListChangeReason};
6use ferroui_reactive::disposables::Disposable;
7use ferroui_reactive::subjects::{BehaviorSubject, Subject, SubjectBase};
8use ferroui_reactive::{Observable, ObservableExt};
9use ferroui_system::{IDisposable, IObservable, IObserver, ObservableError};
10use std::cell::{Cell, RefCell};
11use std::rc::{Rc, Weak};
12
13/// A list that can be observed (`IObservableList<T>`).
14pub trait IObservableList<T>: IDisposable {
15    /// The number of items (`Count`).
16    fn count(&self) -> usize;
17    /// The number of items, now and whenever it changes (`CountChanged`).
18    fn count_changed(&self) -> Rc<dyn IObservable<usize>>;
19    /// The items (`Items`).
20    fn items(&self) -> Vec<T>;
21    /// The items as one range added, then the changes, of the items that `predicate` accepts
22    /// (`Connect`).
23    fn connect_with(&self, predicate: Option<Rc<dyn Fn(&T) -> bool>>) -> Rc<dyn IObservable<ChangeSet<T>>>;
24    /// The changes before they are applied (`Preview`).
25    fn preview_with(&self, predicate: Option<Rc<dyn Fn(&T) -> bool>>) -> Rc<dyn IObservable<ChangeSet<T>>>;
26
27    /// The items and their changes (`Connect()`).
28    fn connect(&self) -> Rc<dyn IObservable<ChangeSet<T>>> {
29        self.connect_with(None)
30    }
31
32    /// The changes before they are applied (`Preview()`).
33    fn preview(&self) -> Rc<dyn IObservable<ChangeSet<T>>> {
34        self.preview_with(None)
35    }
36}
37
38/// A list whose items are edited by its owner (`ISourceList<T>`).
39pub trait ISourceList<T: Clone + PartialEq>: IObservableList<T> {
40    /// Runs an edit; its changes are delivered as one change set when it ends (`Edit`).
41    fn edit_dyn(&self, update: &mut dyn FnMut(&mut dyn IExtendedList<T>));
42}
43
44struct Inner<T> {
45    this: Weak<Inner<T>>,
46    changes: Rc<Subject<ChangeSet<T>>>,
47    changes_preview: Rc<Subject<ChangeSet<T>>>,
48    count_changed: RefCell<Option<Rc<Subject<usize>>>>,
49    reader_writer: ReaderWriter<T>,
50    is_edit_in_progress: RefCell<Option<Rc<BehaviorSubject<bool>>>>,
51    edit_level: Cell<usize>,
52    loader: RefCell<Option<Rc<dyn IDisposable>>>,
53    disposed: Cell<bool>,
54}
55
56/// The source of a reactive list (`SourceList<T>`): edits make change sets, which `connect`
57/// delivers to the operators.
58pub struct SourceList<T> {
59    inner: Rc<Inner<T>>,
60}
61
62impl<T> Clone for SourceList<T> {
63    fn clone(&self) -> Self {
64        Self { inner: self.inner.clone() }
65    }
66}
67
68impl<T: Clone + PartialEq + 'static> Default for SourceList<T> {
69    fn default() -> Self {
70        Self::new()
71    }
72}
73
74impl<T: Clone + PartialEq + 'static> SourceList<T> {
75    fn create() -> Self {
76        let inner = Rc::new_cyclic(|this| Inner {
77            this: this.clone(),
78            changes: Subject::new(),
79            changes_preview: Subject::new(),
80            count_changed: RefCell::new(None),
81            reader_writer: ReaderWriter::new(),
82            is_edit_in_progress: RefCell::new(None),
83            edit_level: Cell::new(0),
84            loader: RefCell::new(None),
85            disposed: Cell::new(false),
86        });
87        Self { inner }
88    }
89
90    /// An empty list (`SourceList()`).
91    pub fn new() -> Self {
92        Self::create()
93    }
94
95    /// A list that applies the change sets of `source` (`SourceList(IObservable<IChangeSet<T>>)`).
96    pub fn from_source<C: super::IChangeSet<T>>(source: Rc<dyn IObservable<C>>) -> Self {
97        let list = Self::create();
98        let (next, error, completed) = (Rc::downgrade(&list.inner), Rc::downgrade(&list.inner), Rc::downgrade(&list.inner));
99        let finally_target = completed.clone();
100        let loader = source
101            .finally(move || {
102                if let Some(inner) = finally_target.upgrade() {
103                    inner.on_completed();
104                }
105            })
106            .subscribe_all(
107                move |changes: C| {
108                    if let Some(inner) = next.upgrade() {
109                        let changes = inner.reader_writer.write_changes(changes.change_set());
110                        inner.invoke_next(&changes);
111                    }
112                },
113                move |e: ObservableError| {
114                    if let Some(inner) = error.upgrade() {
115                        inner.on_error(e);
116                    }
117                },
118                move || {
119                    if let Some(inner) = completed.upgrade() {
120                        inner.on_completed();
121                    }
122                },
123            );
124        *list.inner.loader.borrow_mut() = Some(loader);
125        list
126    }
127
128    /// Runs an edit (`Edit(Action<IExtendedList<T>>)`): the changes are delivered when the
129    /// outermost edit ends, before the edit is no longer in progress.
130    pub fn edit(&self, update: impl FnOnce(&mut dyn IExtendedList<T>)) {
131        let mut update = Some(update);
132        self.inner.edit(&mut |list| (update.take().expect("an edit runs once"))(list));
133    }
134}
135
136impl<T: Clone + PartialEq + 'static> Inner<T> {
137    fn edit_in_progress_subject(&self) -> Option<Rc<BehaviorSubject<bool>>> {
138        self.is_edit_in_progress.borrow().clone()
139    }
140
141    /// `Edit`.
142    fn edit(&self, update: &mut dyn FnMut(&mut dyn IExtendedList<T>)) {
143        let level = self.edit_level.get() + 1;
144        self.edit_level.set(level);
145        if level == 1 {
146            if let Some(subject) = self.edit_in_progress_subject() {
147                subject.on_next(true);
148            }
149        }
150        // An edit that panics still runs upstream's two `finally` blocks: the level comes down
151        // and, at the outermost edit, the edit is no longer in progress. The normal path below
152        // runs the same steps.
153        struct Level<'a, T: Clone + PartialEq + 'static>(&'a Inner<T>, bool);
154        impl<T: Clone + PartialEq + 'static> Drop for Level<'_, T> {
155            fn drop(&mut self) {
156                if self.1 {
157                    let level = self.0.edit_level.get() - 1;
158                    self.0.edit_level.set(level);
159                    if level == 0 {
160                        if let Some(subject) = self.0.edit_in_progress_subject() {
161                            subject.on_next(false);
162                        }
163                    }
164                }
165            }
166        }
167        let mut guard = Level(self, true);
168        let changes = if level == 1 {
169            if self.changes_preview.has_observers() {
170                let this = self.this.clone();
171                Some(self.reader_writer.write_with_preview(update, &move |changes| {
172                    if let Some(inner) = this.upgrade() {
173                        inner.invoke_next_preview(changes);
174                    }
175                }))
176            } else {
177                Some(self.reader_writer.write(update))
178            }
179        } else {
180            self.reader_writer.write_nested(update);
181            None
182        };
183        guard.1 = false;
184        drop(guard);
185        let level = self.edit_level.get() - 1;
186        self.edit_level.set(level);
187        if let Some(changes) = changes {
188            if level == 0 {
189                self.invoke_next(&changes);
190            }
191        }
192        // The outer `finally`.
193        if level == 0 {
194            if let Some(subject) = self.edit_in_progress_subject() {
195                subject.on_next(false);
196            }
197        }
198    }
199
200    fn invoke_next(&self, changes: &ChangeSet<T>) {
201        if changes.count() == 0 {
202            return;
203        }
204        self.changes.on_next(changes.clone());
205        let count = self.count_changed.borrow().clone();
206        if let Some(subject) = count {
207            subject.on_next(self.reader_writer.count());
208        }
209    }
210
211    fn invoke_next_preview(&self, changes: &ChangeSet<T>) {
212        if changes.count() == 0 {
213            return;
214        }
215        self.changes_preview.on_next(changes.clone());
216    }
217
218    fn on_completed(&self) {
219        self.changes_preview.on_completed();
220        self.changes.on_completed();
221        if let Some(subject) = self.edit_in_progress_subject() {
222            subject.on_completed();
223        }
224    }
225
226    fn on_error(&self, error: ObservableError) {
227        self.changes_preview.on_error(error.clone());
228        self.changes.on_error(error.clone());
229        if let Some(subject) = self.edit_in_progress_subject() {
230            subject.on_error(error);
231        }
232    }
233
234    /// `CreateConnectObservable`.
235    fn create_connect_observable(&self, predicate: Option<Rc<dyn Fn(&T) -> bool>>) -> Rc<dyn IObservable<ChangeSet<T>>> {
236        let this = self.this.clone();
237        let observable = Observable::create(move |observer: Rc<dyn IObserver<ChangeSet<T>>>| {
238            let Some(inner) = this.upgrade() else { return Disposable::empty() };
239            let items = inner.reader_writer.items();
240            if !items.is_empty() {
241                observer.on_next(ChangeSet::from(vec![Change::range_change(ListChangeReason::AddRange, items, Some(0))]));
242            }
243            let completed = observer.clone();
244            inner.changes.as_observable().finally(move || completed.on_completed()).subscribe(observer)
245        });
246        match predicate {
247            Some(predicate) => filter_static(observable, predicate),
248            None => observable,
249        }
250    }
251}
252
253impl<T: Clone + PartialEq + 'static> IObservableList<T> for SourceList<T> {
254    fn count(&self) -> usize {
255        self.inner.reader_writer.count()
256    }
257
258    fn count_changed(&self) -> Rc<dyn IObservable<usize>> {
259        let this = Rc::downgrade(&self.inner);
260        Observable::create(move |observer: Rc<dyn IObserver<usize>>| {
261            let Some(inner) = this.upgrade() else { return Disposable::empty() };
262            let subject = inner.count_changed.borrow_mut().get_or_insert_with(Subject::new).clone();
263            subject.as_observable().start_with([inner.reader_writer.count()]).distinct_until_changed().subscribe(observer)
264        })
265    }
266
267    fn items(&self) -> Vec<T> {
268        self.inner.reader_writer.items()
269    }
270
271    fn connect_with(&self, predicate: Option<Rc<dyn Fn(&T) -> bool>>) -> Rc<dyn IObservable<ChangeSet<T>>> {
272        let this = Rc::downgrade(&self.inner);
273        Observable::create(move |observer: Rc<dyn IObserver<ChangeSet<T>>>| {
274            let Some(inner) = this.upgrade() else { return Disposable::empty() };
275            let editing = inner.is_edit_in_progress.borrow().is_some() || inner.edit_level.get() != 0;
276            let observable = if editing {
277                // Defer the connection until no edit is in progress.
278                let subject = inner.is_edit_in_progress.borrow_mut().get_or_insert_with(|| BehaviorSubject::new(inner.edit_level.get() != 0)).clone();
279                let (target, predicate) = (this.clone(), predicate.clone());
280                subject.as_observable().where_(|editing| !*editing).take(1).select_many(move |_| match target.upgrade() {
281                    Some(inner) => inner.create_connect_observable(predicate.clone()),
282                    None => Observable::empty(),
283                })
284            } else {
285                inner.create_connect_observable(predicate.clone())
286            };
287            observable.subscribe(observer)
288        })
289    }
290
291    fn preview_with(&self, predicate: Option<Rc<dyn Fn(&T) -> bool>>) -> Rc<dyn IObservable<ChangeSet<T>>> {
292        let preview = self.inner.changes_preview.as_observable();
293        match predicate {
294            Some(predicate) => filter_static(preview, predicate),
295            None => preview,
296        }
297    }
298}
299
300impl<T: Clone + PartialEq + 'static> ISourceList<T> for SourceList<T> {
301    fn edit_dyn(&self, update: &mut dyn FnMut(&mut dyn IExtendedList<T>)) {
302        self.inner.edit(update);
303    }
304}
305
306impl<T: Clone + PartialEq + 'static> IDisposable for SourceList<T> {
307    /// Ends the subscription to the source and completes the observers (`Dispose`).
308    fn dispose(&self) {
309        if self.inner.disposed.replace(true) {
310            return;
311        }
312        let loader = self.inner.loader.borrow_mut().take();
313        if let Some(loader) = loader {
314            loader.dispose();
315        }
316        self.inner.on_completed();
317        let count = self.inner.count_changed.borrow().clone();
318        if let Some(subject) = count {
319            subject.on_completed();
320        }
321    }
322}
ferroui_dynamic_data/list/source_list.rs

ferroui_dynamic_data/list/
source_list.rs
