rx_rust/operators/utility/
subscribe_on.rs

1use crate::{
2    delegate_disposal,
3    disposable::{Disposable, chain_disposal::ChainDisposal, shared_disposal::SharedDisposal},
4    observable::{Observable, Subscription},
5    observer::Observer,
6    scheduler::Scheduler,
7    utils::types::{MarkerType, MaybeSend},
8};
9use educe::Educe;
10
11/// Specifies the `Scheduler` on which an observer will subscribe to this Observable.
12/// See <https://reactivex.io/documentation/operators/subscribeon.html>
13///
14/// # Examples
15/// ```rust
16/// # #[cfg(not(feature = "tokio-scheduler"))]
17/// # fn main() {}
18/// # #[cfg(feature = "tokio-scheduler")]
19/// #[tokio::main]
20/// async fn main() {
21///     use rx_rust::{
22///         observable::ObservableExt,
23///         observer::Termination,
24///         operators::{
25///             creating::from_iter::FromIter,
26///             utility::subscribe_on::SubscribeOn,
27///         },
28///     };
29///     use std::sync::{Arc, Mutex};
30///     use tokio::time::{sleep, Duration};
31///
32///     let handle = tokio::runtime::Handle::current();
33///     let values = Arc::new(Mutex::new(Vec::new()));
34///     let terminations = Arc::new(Mutex::new(Vec::new()));
35///     let values_observer = Arc::clone(&values);
36///     let terminations_observer = Arc::clone(&terminations);
37///
38///     let subscription = SubscribeOn::new(FromIter::new(vec![1, 2, 3]), handle.clone())
39///         .subscribe_with_callback(
40///             move |value| values_observer.lock().unwrap().push(value),
41///             move |termination| terminations_observer
42///                 .lock()
43///                 .unwrap()
44///                 .push(termination),
45///         );
46///
47///     sleep(Duration::from_millis(10)).await;
48///     drop(subscription);
49///
50///     assert_eq!(&*values.lock().unwrap(), &[1, 2, 3]);
51///     assert_eq!(
52///         &*terminations.lock().unwrap(),
53///         &[Termination::Completed]
54///     );
55/// }
56/// ```
57#[derive(Educe)]
58#[educe(Debug, Clone)]
59pub struct SubscribeOn<'or, OE, S> {
60    source: OE,
61    scheduler: S,
62    _marker: MarkerType<&'or ()>,
63}
64
65impl<'or, OE, S> SubscribeOn<'or, OE, S> {
66    pub fn new(source: OE, scheduler: S) -> Self {
67        Self {
68            source,
69            scheduler,
70            _marker: Default::default(),
71        }
72    }
73}
74
75delegate_disposal!(
76    Disposal<SD, D>,
77    ChainDisposal<SD, SharedDisposal<Subscription<D>>>,
78    where SD: Disposable, D: Disposable
79);
80
81impl<'or, T, E, OE, S> Observable<'static, T, E> for SubscribeOn<'or, OE, S>
82where
83    OE: Observable<'or, T, E> + MaybeSend + 'static,
84    OE::D: MaybeSend + 'static,
85    S: Scheduler,
86{
87    type D = Disposal<S::D, OE::D>;
88
89    fn subscribe(
90        self,
91        observer: impl Observer<T, E> + MaybeSend + 'static,
92    ) -> Subscription<Self::D> {
93        let shared_sub = SharedDisposal::default();
94        let shared_sub_cloned = shared_sub.clone();
95        let disposal = self.scheduler.schedule(
96            move || shared_sub_cloned.replace(|| self.source.subscribe(observer)),
97            None,
98        );
99        disposal.then(shared_sub).map_into()
100    }
101}
rx_rust/operators/utility/subscribe_on.rs

rx_rust/operators/utility/
subscribe_on.rs
