rx_rust/operators/utility/
do_before_subscription.rs

1use crate::utils::types::MaybeSend;
2use crate::{observable::Observable, observable::Subscription, observer::Observer};
3use educe::Educe;
4
5/// Invokes a callback when the Observable is subscribed to, before the subscription is established.
6/// See <https://reactivex.io/documentation/operators/do.html>
7///
8/// # Examples
9/// ```rust
10/// use rx_rust::{
11///     observable::ObservableExt,
12///     operators::{
13///         creating::from_iter::FromIter,
14///         utility::do_before_subscription::DoBeforeSubscription,
15///     },
16/// };
17/// use std::sync::{Arc, Mutex};
18///
19/// let called = Arc::new(Mutex::new(false));
20/// let called_observer = Arc::clone(&called);
21///
22/// DoBeforeSubscription::new(FromIter::new(vec![1]), move || {
23///     *called_observer.lock().unwrap() = true;
24/// })
25/// .subscribe_with_callback(|_| {}, |_| {});
26///
27/// assert!(*called.lock().unwrap());
28/// ```
29#[derive(Educe)]
30#[educe(Debug, Clone)]
31pub struct DoBeforeSubscription<OE, F> {
32    source: OE,
33    callback: F,
34}
35
36impl<OE, F> DoBeforeSubscription<OE, F> {
37    pub fn new<'or, T, E>(source: OE, callback: F) -> Self
38    where
39        OE: Observable<'or, T, E>,
40        F: FnOnce(),
41    {
42        Self { source, callback }
43    }
44}
45
46impl<'or, T, E, OE, F> Observable<'or, T, E> for DoBeforeSubscription<OE, F>
47where
48    T: 'or,
49    E: 'or,
50    OE: Observable<'or, T, E>,
51    F: FnOnce(),
52{
53    type D = OE::D;
54
55    fn subscribe(self, observer: impl Observer<T, E> + MaybeSend + 'or) -> Subscription<Self::D> {
56        (self.callback)();
57        self.source.subscribe(observer)
58    }
59}
rx_rust/operators/utility/do_before_subscription.rs

rx_rust/operators/utility/
do_before_subscription.rs
