rx_rust/operators/others/
hook_on_subscription.rs

1use crate::utils::types::MaybeSend;
2use crate::{
3    disposable::Disposable,
4    observable::{Observable, Subscription},
5    observer::{Observer, boxed_observer::BoxedObserver},
6};
7use educe::Educe;
8
9/// Invokes a callback when the Observable is subscribed to.
10///
11/// # Examples
12/// ```rust
13/// use rx_rust::{
14///     observable::ObservableExt,
15///     observer::Termination,
16///     operators::{
17///         creating::from_iter::FromIter,
18///         others::hook_on_subscription::HookOnSubscription,
19///     },
20/// };
21/// use rx_rust::observable::Observable;
22/// use std::cell::Cell;
23/// use std::rc::Rc;
24///
25/// let mut values = Vec::new();
26/// let mut terminations = Vec::new();
27/// let subscribed = Rc::new(Cell::new(false));
28/// let subscribed_flag = Rc::clone(&subscribed);
29///
30/// let observable = HookOnSubscription::new(FromIter::new(vec![1, 2]), move |source, observer| {
31///     subscribed_flag.set(true);
32///     source.subscribe(observer)
33/// });
34/// observable.subscribe_with_callback(
35///     |value| values.push(value),
36///     |termination| terminations.push(termination),
37/// );
38///
39/// assert!(subscribed.get());
40/// assert_eq!(values, vec![1, 2]);
41/// assert_eq!(terminations, vec![Termination::Completed]);
42/// ```
43#[derive(Educe)]
44#[educe(Debug, Clone)]
45pub struct HookOnSubscription<OE, F> {
46    source: OE,
47    callback: F,
48}
49
50impl<OE, F> HookOnSubscription<OE, F> {
51    pub fn new<'or, T, E, D>(source: OE, callback: F) -> Self
52    where
53        OE: Observable<'or, T, E>,
54        D: Disposable,
55        F: FnOnce(OE, BoxedObserver<'or, T, E>) -> Subscription<D>,
56    {
57        Self { source, callback }
58    }
59}
60
61impl<'or, T, E, OE, F, D> Observable<'or, T, E> for HookOnSubscription<OE, F>
62where
63    OE: Observable<'or, T, E>,
64    D: Disposable,
65    F: FnOnce(OE, BoxedObserver<'or, T, E>) -> Subscription<D>,
66{
67    type D = D;
68
69    fn subscribe(self, observer: impl Observer<T, E> + MaybeSend + 'or) -> Subscription<Self::D> {
70        (self.callback)(self.source, BoxedObserver::new(observer))
71    }
72}
rx_rust/operators/others/hook_on_subscription.rs

rx_rust/operators/others/
hook_on_subscription.rs
