ferroui_reactive/
observer_base.rs

1//! `ObserverBase<T>` (`ObserverBase.cs`).
2//!
3//! Upstream is an abstract class whose subclasses override `OnNextCore`, `OnErrorCore` and
4//! `OnCompletedCore`. Rust has no inheritance: the port is a struct that wraps the three handlers
5//! of the user, given as an [`ObserverCore`] or as functions, and adds what the base class does
6//! (the grammar of the notifications, and `Dispose`).
7
8use ferroui_system::{IDisposable, IObserver, ObservableError};
9use std::cell::Cell;
10
11/// The handlers of an [`ObserverBase`] (the abstract members `OnNextCore`, `OnErrorCore` and
12/// `OnCompletedCore` of `ObserverBase<T>`).
13pub trait ObserverCore<T> {
14    /// Handles a value (`OnNextCore`).
15    fn on_next_core(&self, value: T);
16
17    /// Handles the error that ends the sequence (`OnErrorCore`).
18    fn on_error_core(&self, error: ObservableError);
19
20    /// Handles the end of the sequence (`OnCompletedCore`).
21    fn on_completed_core(&self);
22}
23
24/// The core of an observer made of functions.
25struct FnCore<T> {
26    on_next: Box<dyn Fn(T)>,
27    on_error: Box<dyn Fn(ObservableError)>,
28    on_completed: Box<dyn Fn()>,
29}
30
31impl<T> ObserverCore<T> for FnCore<T> {
32    fn on_next_core(&self, value: T) {
33        (self.on_next)(value);
34    }
35
36    fn on_error_core(&self, error: ObservableError) {
37        (self.on_error)(error);
38    }
39
40    fn on_completed_core(&self) {
41        (self.on_completed)();
42    }
43}
44
45/// An observer that enforces the grammar of the notifications on its core (`ObserverBase<T>`):
46/// values reach the core until a terminal notification; the first terminal notification reaches
47/// it, and every notification after it is ignored. Disposing the observer stops it the same way.
48pub struct ObserverBase<T> {
49    core: Box<dyn ObserverCore<T>>,
50    is_stopped: Cell<bool>,
51}
52
53impl<T: 'static> ObserverBase<T> {
54    /// An observer of the handlers of `core` (a subclass of `ObserverBase<T>` upstream).
55    pub fn new(core: impl ObserverCore<T> + 'static) -> Self {
56        Self { core: Box::new(core), is_stopped: Cell::new(false) }
57    }
58
59    /// An observer whose handlers are the three functions.
60    pub fn from_fns(
61        on_next_core: impl Fn(T) + 'static,
62        on_error_core: impl Fn(ObservableError) + 'static,
63        on_completed_core: impl Fn() + 'static,
64    ) -> Self {
65        Self::new(FnCore { on_next: Box::new(on_next_core), on_error: Box::new(on_error_core), on_completed: Box::new(on_completed_core) })
66    }
67
68    /// Stops the observer and hands `error` to the core, unless it is stopped already (`Fail`);
69    /// true when the core received it.
70    pub(crate) fn fail(&self, error: ObservableError) -> bool {
71        if !self.is_stopped.replace(true) {
72            self.core.on_error_core(error);
73            return true;
74        }
75        false
76    }
77}
78
79impl<T: 'static> IObserver<T> for ObserverBase<T> {
80    /// Hands the value to the core unless the observer is stopped (`OnNext`).
81    fn on_next(&self, value: T) {
82        if !self.is_stopped.get() {
83            self.core.on_next_core(value);
84        }
85    }
86
87    /// Stops the observer and hands the error to the core, the first time (`OnError`).
88    fn on_error(&self, error: ObservableError) {
89        self.fail(error);
90    }
91
92    /// Stops the observer and tells the core, the first time (`OnCompleted`).
93    fn on_completed(&self) {
94        if !self.is_stopped.replace(true) {
95            self.core.on_completed_core();
96        }
97    }
98}
99
100impl<T> IDisposable for ObserverBase<T> {
101    /// Stops the observer: later notifications are ignored (`Dispose`).
102    fn dispose(&self) {
103        self.is_stopped.set(true);
104    }
105}
ferroui_reactive/observer_base.rs

ferroui_reactive/
observer_base.rs
