ferroui_base/reactive/
observable.rs

1use super::operators::{CombineLatest, CombineLatestEnumerable, Switch};
2use super::{AnonymousObserver, Disposable, IDisposable, IObservable, IObserver, ObservableError};
3use std::cell::{Cell, RefCell};
4use std::marker::PhantomData;
5use std::rc::Rc;
6
7/// Factory functions for observables.
8pub struct Observable;
9
10struct CreateObservable<T, F> {
11    subscribe: F,
12    _marker: PhantomData<fn(T)>,
13}
14
15impl<T, F> IObservable<T> for CreateObservable<T, F>
16where
17    F: Fn(Rc<dyn IObserver<T>>) -> Rc<dyn IDisposable>,
18{
19    fn subscribe(&self, observer: Rc<dyn IObserver<T>>) -> Rc<dyn IDisposable> {
20        (self.subscribe)(observer)
21    }
22}
23
24impl Observable {
25    /// Creates an observable from a subscribe function.
26    pub fn create<T: 'static>(
27        subscribe: impl Fn(Rc<dyn IObserver<T>>) -> Rc<dyn IDisposable> + 'static,
28    ) -> Rc<dyn IObservable<T>> {
29        Rc::new(CreateObservable { subscribe, _marker: PhantomData })
30    }
31
32    /// An observable that produces a single value and completes.
33    pub fn return_<T: Clone + 'static>(value: T) -> Rc<dyn IObservable<T>> {
34        Self::create(move |observer| {
35            observer.on_next(value.clone());
36            observer.on_completed();
37            Disposable::empty()
38        })
39    }
40
41    /// An observable that produces a single value and never completes.
42    pub fn single_value<T: Clone + 'static>(value: T) -> Rc<dyn IObservable<T>> {
43        Self::create(move |observer| {
44            observer.on_next(value.clone());
45            Disposable::empty()
46        })
47    }
48
49    /// An observable that never produces a value.
50    pub fn never<T: 'static>() -> Rc<dyn IObservable<T>> {
51        Self::create(|_| Disposable::empty())
52    }
53
54    /// An observable that completes immediately.
55    pub fn empty<T: 'static>() -> Rc<dyn IObservable<T>> {
56        Self::create(|observer| {
57            observer.on_completed();
58            Disposable::empty()
59        })
60    }
61
62    /// Forwards the values of the most recent of the observables that
63    /// `sources` produces.
64    pub fn switch<TSource: 'static>(
65        sources: Rc<dyn IObservable<Rc<dyn IObservable<TSource>>>>,
66    ) -> Rc<dyn IObservable<TSource>> {
67        Rc::new(Switch::new(sources))
68    }
69
70    /// Combines the latest values of a sequence of observables: once every
71    /// one of them has produced a value, each new value produces the list of
72    /// the latest ones.
73    pub fn combine_latest_all<TInput: Clone + 'static>(
74        inputs: impl IntoIterator<Item = Rc<dyn IObservable<TInput>>>,
75    ) -> Rc<dyn IObservable<Vec<TInput>>> {
76        Rc::new(CombineLatestEnumerable::<TInput, Vec<TInput>>::new(inputs, Rc::new(|items: &[TInput]| items.to_vec())))
77    }
78
79    /// An observable of the arguments of an event.
80    ///
81    /// `add_handler` adds the handler it is given to the event and returns
82    /// the handle that removes it, as the events of this crate do (the
83    /// original takes one function to add and one to remove a handler, for
84    /// events with and without typed arguments).
85    pub fn from_event_pattern<T: Clone + 'static>(
86        add_handler: impl Fn(Rc<dyn Fn(&T)>) -> Rc<dyn IDisposable> + 'static,
87    ) -> Rc<dyn IObservable<T>> {
88        Self::create(move |observer: Rc<dyn IObserver<T>>| {
89            let converted: Rc<dyn Fn(&T)> = Rc::new(move |args: &T| observer.on_next(args.clone()));
90            add_handler(converted)
91        })
92    }
93}
94
95struct MapObserver<T, U, F> {
96    target: Rc<dyn IObserver<U>>,
97    f: Rc<F>,
98    _marker: PhantomData<fn(T)>,
99}
100
101impl<T, U, F: Fn(T) -> U> IObserver<T> for MapObserver<T, U, F> {
102    fn on_next(&self, value: T) {
103        self.target.on_next((self.f)(value))
104    }
105    fn on_error(&self, error: ObservableError) {
106        self.target.on_error(error)
107    }
108    fn on_completed(&self) {
109        self.target.on_completed()
110    }
111}
112
113struct FilterObserver<T, F> {
114    target: Rc<dyn IObserver<T>>,
115    f: Rc<F>,
116}
117
118impl<T, F: Fn(&T) -> bool> IObserver<T> for FilterObserver<T, F> {
119    fn on_next(&self, value: T) {
120        if (self.f)(&value) {
121            self.target.on_next(value)
122        }
123    }
124    fn on_error(&self, error: ObservableError) {
125        self.target.on_error(error)
126    }
127    fn on_completed(&self) {
128        self.target.on_completed()
129    }
130}
131
132/// Operators available on every shared observable.
133pub trait ObservableExt<T: 'static> {
134    /// Subscribes with a closure invoked for every value.
135    fn subscribe_fn(&self, on_next: impl Fn(T) + 'static) -> Rc<dyn IDisposable>;
136
137    /// Projects each value.
138    fn select<U: 'static>(&self, f: impl Fn(T) -> U + 'static) -> Rc<dyn IObservable<U>>;
139
140    /// Filters values.
141    fn where_(&self, f: impl Fn(&T) -> bool + 'static) -> Rc<dyn IObservable<T>>;
142
143    /// Produces `value` to each observer before the values of the source.
144    fn start_with(&self, value: T) -> Rc<dyn IObservable<T>>
145    where
146        T: Clone;
147
148    /// Combines the latest values of this observable and `second` with
149    /// `result_selector`, once both have produced a value.
150    fn combine_latest<TSecond: Clone + 'static, TResult: 'static>(
151        &self,
152        second: &Rc<dyn IObservable<TSecond>>,
153        result_selector: impl Fn(T, TSecond) -> TResult + 'static,
154    ) -> Rc<dyn IObservable<TResult>>
155    where
156        T: Clone;
157
158    /// Leaves out the first `skip_count` values.
159    ///
160    /// Panics when `skip_count` is not bigger than zero.
161    fn skip(&self, skip_count: i32) -> Rc<dyn IObservable<T>>;
162
163    /// Produces the first `take_count` values and completes.
164    fn take(&self, take_count: i32) -> Rc<dyn IObservable<T>>;
165}
166
167impl<T: 'static> ObservableExt<T> for Rc<dyn IObservable<T>> {
168    fn subscribe_fn(&self, on_next: impl Fn(T) + 'static) -> Rc<dyn IDisposable> {
169        self.subscribe(Rc::new(AnonymousObserver::new(on_next)))
170    }
171
172    fn select<U: 'static>(&self, f: impl Fn(T) -> U + 'static) -> Rc<dyn IObservable<U>> {
173        let source = self.clone();
174        let f = Rc::new(f);
175        Observable::create(move |observer| {
176            source.subscribe(Rc::new(MapObserver { target: observer, f: f.clone(), _marker: PhantomData }))
177        })
178    }
179
180    fn where_(&self, f: impl Fn(&T) -> bool + 'static) -> Rc<dyn IObservable<T>> {
181        let source = self.clone();
182        let f = Rc::new(f);
183        Observable::create(move |observer| {
184            source.subscribe(Rc::new(FilterObserver { target: observer, f: f.clone() }))
185        })
186    }
187
188    fn start_with(&self, value: T) -> Rc<dyn IObservable<T>>
189    where
190        T: Clone,
191    {
192        let source = self.clone();
193        Observable::create(move |obs: Rc<dyn IObserver<T>>| {
194            obs.on_next(value.clone());
195            source.subscribe(obs)
196        })
197    }
198
199    fn combine_latest<TSecond: Clone + 'static, TResult: 'static>(
200        &self,
201        second: &Rc<dyn IObservable<TSecond>>,
202        result_selector: impl Fn(T, TSecond) -> TResult + 'static,
203    ) -> Rc<dyn IObservable<TResult>>
204    where
205        T: Clone,
206    {
207        let result_selector: Rc<dyn Fn(T, TSecond) -> TResult> = Rc::new(result_selector);
208        Rc::new(CombineLatest::new(self.clone(), second.clone(), result_selector))
209    }
210
211    fn skip(&self, skip_count: i32) -> Rc<dyn IObservable<T>> {
212        if skip_count <= 0 {
213            panic!("Skip count must be bigger than zero (Parameter 'skipCount')");
214        }
215
216        let source = self.clone();
217        Observable::create(move |obs: Rc<dyn IObserver<T>>| {
218            let remaining = Cell::new(skip_count);
219            let (next, error, completed) = (obs.clone(), obs.clone(), obs);
220            source.subscribe(Rc::new(AnonymousObserver::new_with_error_and_completed(
221                move |input: T| {
222                    if remaining.get() <= 0 {
223                        next.on_next(input);
224                    } else {
225                        remaining.set(remaining.get() - 1);
226                    }
227                },
228                move |e| error.on_error(e),
229                move || completed.on_completed(),
230            )))
231        })
232    }
233
234    fn take(&self, take_count: i32) -> Rc<dyn IObservable<T>> {
235        if take_count <= 0 {
236            return Observable::empty();
237        }
238
239        let source = self.clone();
240        Observable::create(move |obs: Rc<dyn IObserver<T>>| {
241            let remaining = Cell::new(take_count);
242            let sub: Rc<RefCell<Option<Rc<dyn IDisposable>>>> = Rc::new(RefCell::new(None));
243            let observer_sub = sub.clone();
244            let (next, error, completed) = (obs.clone(), obs.clone(), obs);
245            let subscription = source.subscribe(Rc::new(AnonymousObserver::new_with_error_and_completed(
246                move |input: T| {
247                    if remaining.get() > 0 {
248                        remaining.set(remaining.get() - 1);
249                        next.on_next(input);
250
251                        if remaining.get() == 0 {
252                            let sub = observer_sub.borrow_mut().take();
253                            if let Some(sub) = sub {
254                                sub.dispose();
255                            }
256                            next.on_completed();
257                        }
258                    }
259                },
260                move |e| error.on_error(e),
261                move || completed.on_completed(),
262            )));
263            *sub.borrow_mut() = Some(subscription.clone());
264            subscription
265        })
266    }
267}
268
269#[cfg(test)]
270mod tests {
271    // Not from upstream: the original has no tests of the operators.
272    use super::*;
273    use crate::reactive::LightweightSubject;
274
275    type Collected<T> = (Rc<RefCell<Vec<T>>>, Rc<Cell<bool>>, Rc<dyn IDisposable>);
276
277    fn collect<T: 'static>(source: &Rc<dyn IObservable<T>>) -> Collected<T> {
278        let received = Rc::new(RefCell::new(Vec::new()));
279        let completed = Rc::new(Cell::new(false));
280        let (r, c) = (received.clone(), completed.clone());
281        let subscription = source.subscribe(Rc::new(AnonymousObserver::new_with_completed(
282            move |value: T| r.borrow_mut().push(value),
283            move || c.set(true),
284        )));
285        (received, completed, subscription)
286    }
287
288    fn subject() -> (LightweightSubject<i32>, Rc<dyn IObservable<i32>>) {
289        let subject = LightweightSubject::<i32>::new();
290        let source: Rc<dyn IObservable<i32>> = Rc::new(subject.clone());
291        (subject, source)
292    }
293
294    #[test]
295    fn start_with_produces_the_value_first() {
296        let (subject, source) = subject();
297        let (received, _, _subscription) = collect(&source.start_with(0));
298
299        subject.on_next(1);
300        assert_eq!(vec![0, 1], *received.borrow());
301    }
302
303    #[test]
304    fn skip_leaves_out_the_first_values() {
305        let (subject, source) = subject();
306        let (received, completed, _subscription) = collect(&source.skip(2));
307
308        for value in 1..=4 {
309            subject.on_next(value);
310        }
311        subject.on_completed();
312        assert_eq!(vec![3, 4], *received.borrow());
313        assert!(completed.get());
314    }
315
316    #[test]
317    #[should_panic(expected = "Skip count must be bigger than zero")]
318    fn skip_throws_for_a_count_of_zero() {
319        let (_, source) = subject();
320        source.skip(0);
321    }
322
323    #[test]
324    fn take_produces_the_first_values_and_completes() {
325        let (subject, source) = subject();
326        let (received, completed, _subscription) = collect(&source.take(2));
327
328        subject.on_next(1);
329        assert!(!completed.get());
330        subject.on_next(2);
331        assert!(completed.get());
332        assert!(!subject.has_observers());
333        subject.on_next(3);
334        assert_eq!(vec![1, 2], *received.borrow());
335    }
336
337    #[test]
338    fn take_of_nothing_completes_at_once() {
339        let (subject, source) = subject();
340        let (received, completed, _subscription) = collect(&source.take(0));
341
342        assert!(completed.get());
343        assert!(!subject.has_observers());
344        assert!(received.borrow().is_empty());
345    }
346
347    #[test]
348    fn combine_latest_and_switch_are_reachable_as_operators() {
349        let (first, first_source) = subject();
350        let (second, second_source) = subject();
351        let (sums, _, _sum_subscription) = collect(&first_source.combine_latest(&second_source, |a: i32, b: i32| a + b));
352
353        let sources = LightweightSubject::<Rc<dyn IObservable<i32>>>::new();
354        let outer: Rc<dyn IObservable<Rc<dyn IObservable<i32>>>> = Rc::new(sources.clone());
355        let (switched, _, _switch_subscription) = collect(&Observable::switch(outer));
356
357        let all = Observable::combine_latest_all([first_source.clone(), second_source.clone()]);
358        let (lists, _, _list_subscription) = collect(&all);
359
360        sources.on_next(first_source.clone());
361        first.on_next(1);
362        second.on_next(10);
363        sources.on_next(second_source.clone());
364        first.on_next(2);
365        second.on_next(20);
366
367        assert_eq!(vec![11, 12, 22], *sums.borrow());
368        assert_eq!(vec![1, 20], *switched.borrow());
369        assert_eq!(vec![vec![1, 10], vec![2, 10], vec![2, 20]], *lists.borrow());
370    }
371
372    #[test]
373    fn from_event_pattern_adds_and_removes_a_handler() {
374        type Handler = Rc<dyn Fn(&i32)>;
375        let handlers: Rc<RefCell<Vec<Handler>>> = Rc::new(RefCell::new(Vec::new()));
376        let h = handlers.clone();
377        let source = Observable::from_event_pattern(move |handler: Handler| {
378            h.borrow_mut().push(handler);
379            let h = h.clone();
380            Disposable::create(move || h.borrow_mut().clear())
381        });
382        let (received, _, subscription) = collect(&source);
383
384        let handler = handlers.borrow()[0].clone();
385        handler(&5);
386        assert_eq!(vec![5], *received.borrow());
387
388        subscription.dispose();
389        assert!(handlers.borrow().is_empty());
390    }
391}
ferroui_base/reactive/observable.rs

ferroui_base/reactive/
observable.rs
