ferroui_base/reactive/
single_subscriber_observable_base.rs

1use super::{IObserver, ObservableError};
2use std::cell::{Cell, RefCell};
3use std::rc::Rc;
4
5/// The base of observables that accept exactly one subscriber and hand
6/// themselves out as the subscription.
7///
8/// The observable embeds this state, implements `IObservable` by calling
9/// [`subscribe`](Self::subscribe) and `IDisposable` by calling
10/// [`dispose`](Self::dispose), passing what has to happen when the
11/// subscription starts and ends.
12pub struct SingleSubscriberObservableBase<T> {
13    error: RefCell<Option<ObservableError>>,
14    observer: RefCell<Option<Rc<dyn IObserver<T>>>>,
15    completed: Cell<bool>,
16}
17
18impl<T> Default for SingleSubscriberObservableBase<T> {
19    fn default() -> Self {
20        Self::new()
21    }
22}
23
24impl<T> SingleSubscriberObservableBase<T> {
25    pub fn new() -> Self {
26        Self { error: RefCell::new(None), observer: RefCell::new(None), completed: Cell::new(false) }
27    }
28
29    /// Registers the observer. `subscribed` runs once it is registered,
30    /// unless the observable has already failed or completed, in which case
31    /// the observer is told so instead.
32    ///
33    /// Panics when there already is a subscriber.
34    pub fn subscribe(&self, observer: Rc<dyn IObserver<T>>, subscribed: impl FnOnce()) {
35        if self.observer.borrow().is_some() {
36            panic!("The observable can only be subscribed once.");
37        }
38        let error = self.error.borrow().clone();
39        if let Some(error) = error {
40            observer.on_error(error);
41        } else if self.completed.get() {
42            observer.on_completed();
43        } else {
44            *self.observer.borrow_mut() = Some(observer);
45            subscribed();
46        }
47    }
48
49    /// Ends the subscription: runs `unsubscribed` and forgets the observer.
50    pub fn dispose(&self, unsubscribed: impl FnOnce()) {
51        unsubscribed();
52        let observer = self.observer.borrow_mut().take();
53        drop(observer);
54    }
55
56    /// Whether an observer is currently subscribed.
57    #[inline]
58    pub fn has_observer(&self) -> bool {
59        self.observer.borrow().is_some()
60    }
61
62    #[inline]
63    fn current(&self) -> Option<Rc<dyn IObserver<T>>> {
64        self.observer.borrow().clone()
65    }
66
67    #[inline]
68    pub fn publish_next(&self, value: T) {
69        if let Some(observer) = self.current() {
70            observer.on_next(value);
71        }
72    }
73
74    pub fn publish_completed(&self, unsubscribed: impl FnOnce()) {
75        self.completed.set(true);
76        if let Some(observer) = self.current() {
77            observer.on_completed();
78            unsubscribed();
79            let observer = self.observer.borrow_mut().take();
80            drop(observer);
81        }
82    }
83
84    pub fn publish_error(&self, error: ObservableError, unsubscribed: impl FnOnce()) {
85        *self.error.borrow_mut() = Some(error.clone());
86        if let Some(observer) = self.current() {
87            observer.on_error(error);
88            unsubscribed();
89            let observer = self.observer.borrow_mut().take();
90            drop(observer);
91        }
92    }
93}
ferroui_base/reactive/single_subscriber_observable_base.rs

ferroui_base/reactive/
single_subscriber_observable_base.rs
