ferroui_base/reactive/
weak_observer_subscription.rs

1use super::{IDisposable, IObservable, IObserver, ObservableError};
2use std::cell::RefCell;
3use std::rc::{Rc, Weak};
4
5/// A subscription that holds its observer weakly and ends itself once the
6/// observer is gone.
7pub struct WeakObserverSubscription<T> {
8    observer: Weak<dyn IObserver<T>>,
9    subscription: RefCell<Option<Rc<dyn IDisposable>>>,
10}
11
12impl<T: 'static> WeakObserverSubscription<T> {
13    fn new(observer: &Rc<dyn IObserver<T>>) -> Self {
14        Self { observer: Rc::downgrade(observer), subscription: RefCell::new(None) }
15    }
16
17    /// Subscribes `observer` to `observable` without keeping the observer
18    /// alive.
19    pub fn subscribe(observable: &dyn IObservable<T>, observer: &Rc<dyn IObserver<T>>) -> Rc<dyn IDisposable> {
20        let subscription = Rc::new(WeakObserverSubscription::new(observer));
21        let inner = observable.subscribe(subscription.clone());
22        *subscription.subscription.borrow_mut() = Some(inner);
23        subscription
24    }
25
26    fn try_get_observer(&self) -> Option<Rc<dyn IObserver<T>>> {
27        if let Some(observer) = self.observer.upgrade() {
28            return Some(observer);
29        }
30
31        // The observer has been collected; unsubscribe from the observable.
32        self.dispose();
33        None
34    }
35}
36
37impl<T: 'static> IObserver<T> for WeakObserverSubscription<T> {
38    fn on_completed(&self) {
39        if let Some(observer) = self.try_get_observer() {
40            observer.on_completed();
41        }
42    }
43
44    fn on_error(&self, error: ObservableError) {
45        if let Some(observer) = self.try_get_observer() {
46            observer.on_error(error);
47        }
48    }
49
50    fn on_next(&self, value: T) {
51        if let Some(observer) = self.try_get_observer() {
52            observer.on_next(value);
53        }
54    }
55}
56
57impl<T: 'static> IDisposable for WeakObserverSubscription<T> {
58    fn dispose(&self) {
59        let subscription = self.subscription.borrow_mut().take();
60        if let Some(subscription) = subscription {
61            subscription.dispose();
62        }
63    }
64}
65
66#[cfg(test)]
67mod tests {
68    // Not from upstream: the original has no tests of this type.
69    use super::*;
70    use crate::reactive::{AnonymousObserver, LightweightSubject};
71
72    #[test]
73    fn forwards_while_the_observer_is_alive_and_unsubscribes_once_it_is_gone() {
74        let subject = LightweightSubject::<i32>::new();
75        let received = Rc::new(RefCell::new(Vec::new()));
76        let r = received.clone();
77        let observer: Rc<dyn IObserver<i32>> = Rc::new(AnonymousObserver::new(move |value: i32| r.borrow_mut().push(value)));
78
79        let _subscription = WeakObserverSubscription::subscribe(&subject, &observer);
80        subject.on_next(1);
81        assert_eq!(vec![1], *received.borrow());
82        assert!(subject.has_observers());
83
84        drop(observer);
85        subject.on_next(2);
86        assert_eq!(vec![1], *received.borrow());
87        assert!(!subject.has_observers());
88    }
89
90    #[test]
91    fn dispose_unsubscribes() {
92        let subject = LightweightSubject::<i32>::new();
93        let observer: Rc<dyn IObserver<i32>> = Rc::new(AnonymousObserver::new(|_: i32| {}));
94
95        let subscription = WeakObserverSubscription::subscribe(&subject, &observer);
96        assert!(subject.has_observers());
97
98        subscription.dispose();
99        assert!(!subject.has_observers());
100    }
101}
ferroui_base/reactive/weak_observer_subscription.rs

ferroui_base/reactive/
weak_observer_subscription.rs
