rx_rust/utils/
subscribe_with_auto_dispose_on_termination.rs

1use crate::{
2    delegate_disposal,
3    disposable::{Disposable, DisposableExt, shared_disposal::SharedDisposal},
4    observable::Subscription,
5    observer::{Flow, Observer, Termination},
6    utils::on_panic::on_panic,
7};
8use educe::Educe;
9
10delegate_disposal!(
11    Disposal<D>,
12    SharedDisposal<Subscription<D>>,
13    where D: Disposable
14);
15
16/// Wraps subscription creation so that termination from the observer automatically disposes the inner subscription.
17pub fn subscribe_with_auto_dispose_on_termination<OR, D, F>(
18    observer: OR,
19    builder: F,
20) -> Subscription<Disposal<D>>
21where
22    D: Disposable,
23    F: FnOnce(AutoDisposeOnTerminationObserver<OR, D>) -> Subscription<D>,
24{
25    let shared_disposal = SharedDisposal::default();
26    let observer = AutoDisposeOnTerminationObserver {
27        observer,
28        shared_disposal: shared_disposal.clone(),
29    };
30    shared_disposal.replace(|| builder(observer));
31
32    shared_disposal.into_subscription()
33}
34
35/// Whether `OR` is an [`AutoDisposeOnTerminationObserver`], whatever its generic arguments are.
36///
37/// Only the outermost type is recognized: an auto-disposing observer wrapped inside another
38/// observer is not detected. This is a best-effort check meant for `debug_assert!`, not a complete
39/// one.
40pub(crate) fn is_auto_dispose_on_termination_observer<OR>() -> bool {
41    fn type_name_without_generics<T>() -> &'static str {
42        let name = std::any::type_name::<T>();
43        name.split_once('<').map_or(name, |(name, _)| name)
44    }
45
46    type_name_without_generics::<OR>()
47        == type_name_without_generics::<AutoDisposeOnTerminationObserver<(), ()>>()
48}
49
50#[derive(Educe)]
51#[educe(Debug)]
52pub struct AutoDisposeOnTerminationObserver<OR, D: Disposable> {
53    observer: OR,
54    shared_disposal: SharedDisposal<Subscription<D>>,
55}
56
57impl<T, E, OR, D> Observer<T, E> for AutoDisposeOnTerminationObserver<OR, D>
58where
59    OR: Observer<T, E>,
60    D: Disposable,
61{
62    fn on_next(&mut self, value: T) -> Flow {
63        let flow = self.observer.on_next(value);
64        if flow.is_stop() {
65            // The observer ended its own stream, which is what a termination does too, so the
66            // subscription is disposed here as well: the source is asked to stop by the flow this
67            // returns, and released by the disposal whether or not it honors it.
68            self.shared_disposal.clone().dispose();
69        }
70        flow
71    }
72
73    fn on_termination(self, termination: Termination<E>) {
74        let Self {
75            observer,
76            shared_disposal,
77        } = self;
78        // Without the guard the source stays subscribed after the termination, until the
79        // subscription `subscribe_with_auto_dispose_on_termination` returned is dropped. The
80        // returning path disposes right after the callback, so the panicking path takes exactly
81        // the locks the returning one would: nothing new can deadlock on the unwinding thread.
82        let guard = on_panic(|| shared_disposal.clone().dispose());
83        observer.on_termination(termination);
84        drop(guard);
85        shared_disposal.dispose();
86    }
87}
rx_rust/utils/subscribe_with_auto_dispose_on_termination.rs

rx_rust/utils/
subscribe_with_auto_dispose_on_termination.rs
