ferroui_base/reactive/
lightweight_observable_base.rs

1use super::{Disposable, IDisposable, IObserver, ObservableError};
2use std::cell::RefCell;
3use std::rc::Rc;
4
5/// The overridable members of an observable built on
6/// [`LightweightObservableBase`]: the observable embeds the base, returns
7/// it from [`observable_base`](Self::observable_base) and implements
8/// `IObservable` with [`LightweightObservableBase::subscribe`].
9pub trait LightweightObservable<T>: 'static {
10    /// The embedded state.
11    fn observable_base(&self) -> &LightweightObservableBase<T>;
12
13    /// Called when the first observer subscribes.
14    fn initialize(&self);
15
16    /// Called when the last observer unsubscribes, and after the
17    /// observable completed or failed.
18    fn deinitialize(&self);
19
20    /// Called after an observer subscribed; `first` tells whether it is
21    /// the only one.
22    fn subscribed(&self, _observer: &Rc<dyn IObserver<T>>, _first: bool) {}
23}
24
25/// The state of a lightweight observable: its observers, notified in
26/// subscription order, and the error it failed with. Once it completes or
27/// fails, the list is gone and later subscribers are told so at once.
28pub struct LightweightObservableBase<T> {
29    error: RefCell<Option<ObservableError>>,
30    observers: RefCell<Option<Vec<Rc<dyn IObserver<T>>>>>,
31}
32
33impl<T> Default for LightweightObservableBase<T> {
34    fn default() -> Self {
35        Self::new()
36    }
37}
38
39impl<T> LightweightObservableBase<T> {
40    pub fn new() -> Self {
41        Self { error: RefCell::new(None), observers: RefCell::new(Some(Vec::new())) }
42    }
43
44    /// Whether any observer is subscribed.
45    pub fn has_observers(&self) -> bool {
46        self.observers.borrow().as_ref().is_some_and(|observers| !observers.is_empty())
47    }
48
49    /// Subscribes `observer` to `owner`. The first subscription initializes
50    /// the owner; disposing the returned handle removes the observer.
51    pub fn subscribe<O: LightweightObservable<T>>(owner: &Rc<O>, observer: Rc<dyn IObserver<T>>) -> Rc<dyn IDisposable>
52    where
53        T: 'static,
54    {
55        let base = owner.observable_base();
56        let first = {
57            let mut observers = base.observers.borrow_mut();
58            match observers.as_mut() {
59                Some(observers) => {
60                    let first = observers.is_empty();
61                    observers.push(observer.clone());
62                    first
63                }
64                None => {
65                    drop(observers);
66                    let error = base.error.borrow().clone();
67                    match error {
68                        Some(error) => observer.on_error(error),
69                        None => observer.on_completed(),
70                    }
71                    return Disposable::empty();
72                }
73            }
74        };
75
76        if first {
77            owner.initialize();
78        }
79
80        owner.subscribed(&observer, first);
81
82        Rc::new(RemoveObserver { parent: RefCell::new(Some((owner.clone(), observer))) })
83    }
84
85    fn remove<O: LightweightObservable<T>>(owner: &O, observer: &Rc<dyn IObserver<T>>) {
86        let base = owner.observable_base();
87        let now_empty = {
88            let mut observers = base.observers.borrow_mut();
89            let Some(observers) = observers.as_mut() else { return };
90            if let Some(index) = observers.iter().position(|o| std::ptr::addr_eq(Rc::as_ptr(o), Rc::as_ptr(observer))) {
91                observers.remove(index);
92            }
93            if observers.is_empty() {
94                observers.shrink_to_fit();
95                true
96            } else {
97                false
98            }
99        };
100        if now_empty {
101            owner.deinitialize();
102        }
103    }
104
105    /// The observers to notify, copied so that they can subscribe and
106    /// unsubscribe while they are notified.
107    fn snapshot(&self) -> Option<Vec<Rc<dyn IObserver<T>>>> {
108        self.observers.borrow().clone()
109    }
110
111    /// Notifies every observer of a value.
112    pub fn publish_next(&self, value: T)
113    where
114        T: Clone,
115    {
116        let Some(observers) = self.snapshot() else { return };
117        for observer in &observers {
118            observer.on_next(value.clone());
119        }
120    }
121
122    /// Completes the observable: notifies every observer and deinitializes
123    /// the owner.
124    pub fn publish_completed<O: LightweightObservable<T>>(owner: &O) {
125        let Some(observers) = owner.observable_base().observers.borrow_mut().take() else { return };
126        for observer in &observers {
127            observer.on_completed();
128        }
129        owner.deinitialize();
130    }
131
132    /// Fails the observable: notifies every observer and deinitializes the
133    /// owner.
134    pub fn publish_error<O: LightweightObservable<T>>(owner: &O, error: ObservableError) {
135        let base = owner.observable_base();
136        let Some(observers) = base.observers.borrow_mut().take() else { return };
137        *base.error.borrow_mut() = Some(error.clone());
138        for observer in &observers {
139            observer.on_error(error.clone());
140        }
141        owner.deinitialize();
142    }
143}
144
145/// The subscription returned by [`LightweightObservableBase::subscribe`].
146struct RemoveObserver<O, T> {
147    parent: RefCell<Option<(Rc<O>, Rc<dyn IObserver<T>>)>>,
148}
149
150impl<O: LightweightObservable<T>, T: 'static> IDisposable for RemoveObserver<O, T> {
151    fn dispose(&self) {
152        let parent = self.parent.borrow_mut().take();
153        if let Some((parent, observer)) = parent {
154            LightweightObservableBase::remove(&*parent, &observer);
155        }
156    }
157}
ferroui_base/reactive/lightweight_observable_base.rs

ferroui_base/reactive/
lightweight_observable_base.rs
