ferroui_reactive/
observable_base.rs

1//! `ObservableBase<T>` (`ObservableBase.cs`).
2//!
3//! Upstream is an abstract class whose subclasses override `SubscribeCore`; Rust has no
4//! inheritance, and a blanket implementation of the foreign trait `IObservable<T>` for the types
5//! of a trait is not allowed. The port is a struct that wraps the subscription of the user, given
6//! as an [`ObservableCore`] or as a function, and subscribes through it as the base class does.
7
8use crate::internal::try_subscribe_auto_detach;
9use ferroui_system::{IDisposable, IObservable, IObserver, ObservableError};
10use std::rc::Rc;
11
12/// The subscription of an [`ObservableBase`] (the abstract member `SubscribeCore` of
13/// `ObservableBase<T>`).
14pub trait ObservableCore<T> {
15    /// Subscribes `observer` and returns the disposable that ends the subscription
16    /// (`SubscribeCore`); an `Err` is the exception upstream's `SubscribeCore` throws, which is
17    /// delivered to the observer as its error.
18    fn subscribe_core(&self, observer: Rc<dyn IObserver<T>>) -> Result<Rc<dyn IDisposable>, ObservableError>;
19}
20
21/// The core of an observable made of a function.
22struct FnCore<F>(F);
23
24impl<T, F> ObservableCore<T> for FnCore<F>
25where
26    F: Fn(Rc<dyn IObserver<T>>) -> Result<Rc<dyn IDisposable>, ObservableError>,
27{
28    fn subscribe_core(&self, observer: Rc<dyn IObserver<T>>) -> Result<Rc<dyn IDisposable>, ObservableError> {
29        (self.0)(observer)
30    }
31}
32
33/// An observable whose subscription is the user's [`ObservableCore`] (`ObservableBase<T>`).
34///
35/// The observer the core receives ignores everything after a terminal notification, and a
36/// terminal notification, or a panic while the observer handles a value, disposes the
37/// subscription (`AutoDetachObserver<T>`). When no work of the current-thread scheduler is
38/// running, the subscription runs through its trampoline, so that work it schedules there runs
39/// after the subscription has been set up.
40pub struct ObservableBase<T> {
41    core: Rc<dyn ObservableCore<T>>,
42}
43
44impl<T: 'static> ObservableBase<T> {
45    /// An observable whose subscription is `core` (a subclass of `ObservableBase<T>` upstream).
46    pub fn new(core: impl ObservableCore<T> + 'static) -> Self {
47        Self { core: Rc::new(core) }
48    }
49
50    /// An observable whose subscription is the function `subscribe_core`.
51    pub fn from_fn(subscribe_core: impl Fn(Rc<dyn IObserver<T>>) -> Rc<dyn IDisposable> + 'static) -> Self {
52        Self::new(FnCore(move |observer: Rc<dyn IObserver<T>>| -> Result<Rc<dyn IDisposable>, ObservableError> {
53            Ok(subscribe_core(observer))
54        }))
55    }
56
57    /// An observable whose subscription is the function `subscribe_core`, which may fail: an `Err`
58    /// is delivered to the observer as its error (a `SubscribeCore` that throws upstream).
59    pub fn try_from_fn(
60        subscribe_core: impl Fn(Rc<dyn IObserver<T>>) -> Result<Rc<dyn IDisposable>, ObservableError> + 'static,
61    ) -> Self {
62        Self::new(FnCore(subscribe_core))
63    }
64}
65
66impl<T: 'static> IObservable<T> for ObservableBase<T> {
67    /// Subscribes through the core (`Subscribe`): the observer is wrapped in an auto-detaching
68    /// observer, which the core receives; an error of the core is delivered to that observer
69    /// (`AutoDetachObserver.Fail`), and panics, where upstream rethrows it, when the observer has
70    /// stopped already.
71    fn subscribe(&self, observer: Rc<dyn IObserver<T>>) -> Rc<dyn IDisposable> {
72        let core = self.core.clone();
73        try_subscribe_auto_detach(observer, move |observer: Rc<dyn IObserver<T>>| core.subscribe_core(observer))
74    }
75}
ferroui_reactive/observable_base.rs

ferroui_reactive/
observable_base.rs
