ferroui_reactive/
observer.rs

1//! `AnonymousObserver<T>` and `Observer` (`AnonymousObserver.cs`, `Observer.Extensions.cs`), the
2//! observers of `Synchronize` (`Internal/SynchronizedObserver.cs`,
3//! `Internal/AsyncLockObserver.cs`), the observers of `NotifyOn`, and the conversions between
4//! observers and progress objects (`ToProgress`, `ToObserver(IProgress<T>)`,
5//! `AnonymousProgress<T>`).
6
7use crate::concurrency::AsyncLock;
8use crate::notification::Notification;
9use crate::observer_base::{ObserverBase, ObserverCore};
10use ferroui_system::{IObserver, IProgress, ObservableError};
11use std::cell::Cell;
12use std::rc::Rc;
13
14type OnError = Box<dyn Fn(ObservableError)>;
15
16/// An observer made of functions (`AnonymousObserver<T>`). It ignores everything after a terminal
17/// notification. Without a function for errors, an error panics, as upstream rethrows it.
18pub struct AnonymousObserver<T> {
19    on_next: Box<dyn Fn(T)>,
20    on_error: OnError,
21    on_completed: Box<dyn Fn()>,
22    is_stopped: Cell<bool>,
23}
24
25impl<T> AnonymousObserver<T> {
26    /// An observer of the three functions.
27    pub fn new(
28        on_next: impl Fn(T) + 'static,
29        on_error: impl Fn(ObservableError) + 'static,
30        on_completed: impl Fn() + 'static,
31    ) -> Self {
32        Self { on_next: Box::new(on_next), on_error: Box::new(on_error), on_completed: Box::new(on_completed), is_stopped: Cell::new(false) }
33    }
34}
35
36/// The error of an observer that has no function for errors (`Stubs.Throw`): it panics with the
37/// error, which is how an unhandled error of a sequence surfaces.
38pub(crate) fn throw(error: ObservableError) {
39    panic!("unhandled error of an observable sequence: {error}");
40}
41
42impl<T> IObserver<T> for AnonymousObserver<T> {
43    fn on_next(&self, value: T) {
44        if !self.is_stopped.get() {
45            (self.on_next)(value);
46        }
47    }
48
49    fn on_error(&self, error: ObservableError) {
50        if !self.is_stopped.replace(true) {
51            (self.on_error)(error);
52        }
53    }
54
55    fn on_completed(&self) {
56        if !self.is_stopped.replace(true) {
57            (self.on_completed)();
58        }
59    }
60}
61
62/// The factory of observers (`Observer`).
63pub struct Observer;
64
65impl Observer {
66    /// An observer of values; an error panics (`Create(Action<T>)`).
67    pub fn create<T: 'static>(on_next: impl Fn(T) + 'static) -> Rc<dyn IObserver<T>> {
68        Rc::new(AnonymousObserver::new(on_next, throw, || {}))
69    }
70
71    /// An observer of values and errors (`Create(Action<T>, Action<Exception>)`).
72    pub fn create_with_error<T: 'static>(
73        on_next: impl Fn(T) + 'static,
74        on_error: impl Fn(ObservableError) + 'static,
75    ) -> Rc<dyn IObserver<T>> {
76        Rc::new(AnonymousObserver::new(on_next, on_error, || {}))
77    }
78
79    /// An observer of values and the end; an error panics (`Create(Action<T>, Action)`).
80    pub fn create_with_completed<T: 'static>(on_next: impl Fn(T) + 'static, on_completed: impl Fn() + 'static) -> Rc<dyn IObserver<T>> {
81        Rc::new(AnonymousObserver::new(on_next, throw, on_completed))
82    }
83
84    /// An observer of all three notifications (`Create(Action<T>, Action<Exception>, Action)`).
85    pub fn create_all<T: 'static>(
86        on_next: impl Fn(T) + 'static,
87        on_error: impl Fn(ObservableError) + 'static,
88        on_completed: impl Fn() + 'static,
89    ) -> Rc<dyn IObserver<T>> {
90        Rc::new(AnonymousObserver::new(on_next, on_error, on_completed))
91    }
92
93    /// An observer that hands each notification, as a value, to `handler` (`ToObserver(Action<Notification<T>>)`).
94    pub fn to_observer<T: 'static>(handler: impl Fn(Notification<T>) + 'static) -> Rc<dyn IObserver<T>> {
95        let handler = Rc::new(handler);
96        let (on_error, on_completed) = (handler.clone(), handler.clone());
97        Rc::new(AnonymousObserver::new(
98            move |value| handler(Notification::OnNext(value)),
99            move |error| on_error(Notification::OnError(error)),
100            move || on_completed(Notification::OnCompleted),
101        ))
102    }
103}
104
105/// The extension methods of `Observer` on an observer (`Observer.Extensions.cs`).
106pub trait ObserverExt<T> {
107    /// The observer as a function of notifications (`ToNotifier`).
108    fn to_notifier(&self) -> Box<dyn Fn(Notification<T>)>;
109
110    /// An observer that hides the identity of this one (`AsObserver`).
111    fn as_observer(&self) -> Rc<dyn IObserver<T>>;
112
113    /// An observer that checks the grammar of the notifications it receives (`Checked`): it panics,
114    /// where upstream throws `InvalidOperationException`, on a notification after a terminal one
115    /// and on a notification while another is being handled.
116    fn checked(&self) -> Rc<dyn IObserver<T>>;
117
118    /// An observer whose notifications reach this one under a lock of its own
119    /// (`Synchronize(IObserver<T>)`, a `SynchronizedObserver<T>`): it ignores everything after a
120    /// terminal notification. On one thread the notifications are serialized already, and the lock
121    /// of upstream (a monitor) is re-entrant: a notification that re-enters the observer from
122    /// inside another reaches this one at once.
123    fn synchronize(&self) -> Rc<dyn IObserver<T>>;
124
125    /// An observer whose notifications reach this one one at a time
126    /// (`Synchronize(IObserver<T>, bool preventReentrancy)`): with `prevent_reentrancy`, through an
127    /// [`AsyncLock`] of its own, so that a notification that re-enters the observer from inside
128    /// another is queued and reaches this one when that one has returned (`AsyncLockObserver<T>`);
129    /// without, as [`synchronize`](Self::synchronize).
130    fn synchronize_prevent_reentrancy(&self, prevent_reentrancy: bool) -> Rc<dyn IObserver<T>>;
131
132    /// An observer whose notifications reach this one under the lock `gate`
133    /// (`Synchronize(IObserver<T>, object gate)`), which observers may share; as
134    /// [`synchronize`](Self::synchronize), the lock is re-entrant and has nothing to serialize on
135    /// one thread. The observer keeps the gate.
136    fn synchronize_with_gate(&self, gate: Rc<dyn std::any::Any>) -> Rc<dyn IObserver<T>>;
137
138    /// An observer whose notifications reach this one through `async_lock`
139    /// (`Synchronize(IObserver<T>, AsyncLock)`, an `AsyncLockObserver<T>`), which observers may
140    /// share: a notification that arrives while an action of the lock runs (a re-entrant one) is
141    /// queued and reaches this one after it.
142    fn synchronize_with_lock(&self, async_lock: Rc<AsyncLock>) -> Rc<dyn IObserver<T>>;
143
144    /// An observer that delivers the notifications to this one on `scheduler`, in order
145    /// (`NotifyOn(IScheduler)`).
146    fn notify_on(&self, scheduler: Rc<dyn crate::concurrency::IScheduler>) -> Rc<dyn IObserver<T>>;
147
148    /// An observer that delivers the notifications to this one through `context`, in order
149    /// (`NotifyOn(SynchronizationContext)`: the observer of `NotifyOn(IScheduler)` on a
150    /// [`SynchronizationContextScheduler`](crate::concurrency::SynchronizationContextScheduler) of
151    /// the context, which posts with
152    /// [`post_local`](ferroui_system::threading::SynchronizationContext::post_local): the observer
153    /// is notified on the thread of the context).
154    fn notify_on_context(&self, context: std::sync::Arc<dyn ferroui_system::threading::SynchronizationContext>) -> Rc<dyn IObserver<T>>;
155
156    /// A progress object whose reports are the values of this observer (`ToProgress(IObserver<T>)`).
157    fn to_progress(&self) -> Rc<dyn IProgress<T>>;
158
159    /// A progress object whose reports reach this observer as values on `scheduler`, in order
160    /// (`ToProgress(IObserver<T>, IScheduler)`, through the observer of `ObserveOn`).
161    fn to_progress_on(&self, scheduler: Rc<dyn crate::concurrency::IScheduler>) -> Rc<dyn IProgress<T>>;
162}
163
164impl<T: 'static> ObserverExt<T> for Rc<dyn IObserver<T>> {
165    fn to_notifier(&self) -> Box<dyn Fn(Notification<T>)> {
166        let observer = self.clone();
167        Box::new(move |notification| notification.accept(&*observer))
168    }
169
170    fn as_observer(&self) -> Rc<dyn IObserver<T>> {
171        let (next, error, completed) = (self.clone(), self.clone(), self.clone());
172        Rc::new(AnonymousObserver::new(move |v| next.on_next(v), move |e| error.on_error(e), move || completed.on_completed()))
173    }
174
175    fn checked(&self) -> Rc<dyn IObserver<T>> {
176        Rc::new(CheckedObserver { observer: self.clone(), state: Cell::new(CheckedState::Idle) })
177    }
178
179    fn synchronize(&self) -> Rc<dyn IObserver<T>> {
180        synchronized(self.clone(), Rc::new(()))
181    }
182
183    fn synchronize_prevent_reentrancy(&self, prevent_reentrancy: bool) -> Rc<dyn IObserver<T>> {
184        if prevent_reentrancy {
185            return Rc::new(ObserverBase::new(AsyncLockCore { observer: self.clone(), gate: Rc::new(AsyncLock::new()) }));
186        }
187        synchronized(self.clone(), Rc::new(()))
188    }
189
190    fn synchronize_with_gate(&self, gate: Rc<dyn std::any::Any>) -> Rc<dyn IObserver<T>> {
191        synchronized(self.clone(), gate)
192    }
193
194    fn synchronize_with_lock(&self, async_lock: Rc<AsyncLock>) -> Rc<dyn IObserver<T>> {
195        Rc::new(ObserverBase::new(AsyncLockCore { observer: self.clone(), gate: async_lock }))
196    }
197
198    fn notify_on(&self, scheduler: Rc<dyn crate::concurrency::IScheduler>) -> Rc<dyn IObserver<T>> {
199        Rc::new(NotifyOnObserver { inner: crate::internal::ScheduledObserver::new(scheduler, self.clone()) })
200    }
201
202    fn notify_on_context(&self, context: std::sync::Arc<dyn ferroui_system::threading::SynchronizationContext>) -> Rc<dyn IObserver<T>> {
203        let scheduler: Rc<dyn crate::concurrency::IScheduler> = crate::concurrency::SynchronizationContextScheduler::new(context);
204        self.notify_on(scheduler)
205    }
206
207    fn to_progress(&self) -> Rc<dyn IProgress<T>> {
208        let observer = self.clone();
209        Rc::new(AnonymousProgress { progress: Box::new(move |value| observer.on_next(value)) })
210    }
211
212    fn to_progress_on(&self, scheduler: Rc<dyn crate::concurrency::IScheduler>) -> Rc<dyn IProgress<T>> {
213        let observer = self.notify_on(scheduler);
214        Rc::new(AnonymousProgress { progress: Box::new(move |value| observer.on_next(value)) })
215    }
216}
217
218/// A progress object made of a function (`AnonymousProgress<T>` of `Observer.Extensions.cs`).
219struct AnonymousProgress<T> {
220    progress: Box<dyn Fn(T)>,
221}
222
223impl<T> IProgress<T> for AnonymousProgress<T> {
224    fn report(&self, value: T) {
225        (self.progress)(value);
226    }
227}
228
229/// The extension methods of `Observer` on a progress object (`Observer.Extensions.cs`).
230pub trait ProgressExt<T> {
231    /// An observer whose values are reports of this progress object; an error panics, as the
232    /// observer of upstream rethrows it (`ToObserver(IProgress<T>)`).
233    fn to_observer(&self) -> Rc<dyn IObserver<T>>;
234}
235
236impl<T: 'static> ProgressExt<T> for Rc<dyn IProgress<T>> {
237    fn to_observer(&self) -> Rc<dyn IObserver<T>> {
238        let progress = self.clone();
239        Observer::create(move |value| progress.report(value))
240    }
241}
242
243/// `SynchronizedObserver<T>` on `observer` with the lock `gate`.
244fn synchronized<T: 'static>(observer: Rc<dyn IObserver<T>>, gate: Rc<dyn std::any::Any>) -> Rc<dyn IObserver<T>> {
245    Rc::new(ObserverBase::new(SynchronizedCore { observer, _gate: gate }))
246}
247
248/// The core of `SynchronizedObserver<T>` (an `ObserverBase<T>`): each notification reaches the
249/// observer inside `lock (gate)`. A monitor is re-entrant and the notifications of one thread
250/// cannot overlap, so the notification reaches it at once; the gate is kept, as upstream keeps it.
251struct SynchronizedCore<T> {
252    observer: Rc<dyn IObserver<T>>,
253    _gate: Rc<dyn std::any::Any>,
254}
255
256impl<T> ObserverCore<T> for SynchronizedCore<T> {
257    fn on_next_core(&self, value: T) {
258        self.observer.on_next(value);
259    }
260
261    fn on_error_core(&self, error: ObservableError) {
262        self.observer.on_error(error);
263    }
264
265    fn on_completed_core(&self) {
266        self.observer.on_completed();
267    }
268}
269
270/// The core of `AsyncLockObserver<T>` (an `ObserverBase<T>`): each notification reaches the
271/// observer through `gate.Wait`, so that one that arrives while another runs is queued after it.
272struct AsyncLockCore<T> {
273    observer: Rc<dyn IObserver<T>>,
274    gate: Rc<AsyncLock>,
275}
276
277impl<T: 'static> ObserverCore<T> for AsyncLockCore<T> {
278    fn on_next_core(&self, value: T) {
279        let observer = self.observer.clone();
280        self.gate.wait(move || observer.on_next(value));
281    }
282
283    fn on_error_core(&self, error: ObservableError) {
284        let observer = self.observer.clone();
285        self.gate.wait(move || observer.on_error(error));
286    }
287
288    fn on_completed_core(&self) {
289        let observer = self.observer.clone();
290        self.gate.wait(move || observer.on_completed());
291    }
292}
293
294/// The observer of `NotifyOn` (`ObserveOnObserver<T>` as an observer): each notification is
295/// queued and the queue is delivered on the scheduler.
296struct NotifyOnObserver<T: 'static> {
297    inner: Rc<crate::internal::ScheduledObserver<T>>,
298}
299
300impl<T: 'static> IObserver<T> for NotifyOnObserver<T> {
301    fn on_next(&self, value: T) {
302        use crate::internal::IScheduledObserver;
303        self.inner.on_next(value);
304        self.inner.ensure_active();
305    }
306
307    fn on_error(&self, error: ObservableError) {
308        use crate::internal::IScheduledObserver;
309        self.inner.on_error(error);
310        self.inner.ensure_active();
311    }
312
313    fn on_completed(&self) {
314        use crate::internal::IScheduledObserver;
315        self.inner.on_completed();
316        self.inner.ensure_active();
317    }
318}
319
320#[derive(Clone, Copy, PartialEq, Eq)]
321enum CheckedState {
322    Idle,
323    Busy,
324    Done,
325}
326
327/// `CheckedObserver<T>`.
328struct CheckedObserver<T> {
329    observer: Rc<dyn IObserver<T>>,
330    state: Cell<CheckedState>,
331}
332
333impl<T> CheckedObserver<T> {
334    /// Moves from idle to busy; any other state is left as it is and panics (`CheckAccess`, which
335    /// only moves from `Idle`).
336    fn check_access(&self) {
337        match self.state.get() {
338            CheckedState::Busy => panic!("This observer is currently busy with a notification."),
339            CheckedState::Done => panic!("This observer has already terminated."),
340            CheckedState::Idle => self.state.set(CheckedState::Busy),
341        }
342    }
343}
344
345impl<T> IObserver<T> for CheckedObserver<T> {
346    fn on_next(&self, value: T) {
347        self.check_access();
348        self.observer.on_next(value);
349        self.state.set(CheckedState::Idle);
350    }
351
352    fn on_error(&self, error: ObservableError) {
353        self.check_access();
354        self.observer.on_error(error);
355        self.state.set(CheckedState::Done);
356    }
357
358    fn on_completed(&self) {
359        self.check_access();
360        self.observer.on_completed();
361        self.state.set(CheckedState::Done);
362    }
363}
ferroui_reactive/observer.rs

ferroui_reactive/
observer.rs
