rx_rust/operators/combining/
start_with.rs

1use crate::disposable::{DisposableExt, option_disposal::OptionDisposal};
2use crate::utils::types::MaybeSend;
3use crate::{
4    observable::{Observable, Subscription},
5    observer::Observer,
6};
7use educe::Educe;
8
9/// Emits a specified sequence of values before beginning to emit the items from the source Observable.
10/// See <https://reactivex.io/documentation/operators/startwith.html>
11///
12/// # Examples
13/// ```rust
14/// use rx_rust::{
15///     observable::ObservableExt,
16///     observer::Termination,
17///     operators::{
18///         combining::start_with::StartWith,
19///         creating::from_iter::FromIter,
20///     },
21/// };
22///
23/// let mut values = Vec::new();
24/// let mut terminations = Vec::new();
25///
26/// let observable = StartWith::new(FromIter::new(vec![3, 4]), vec![1, 2]);
27/// observable.subscribe_with_callback(
28///     |value| values.push(value),
29///     |termination| terminations.push(termination),
30/// );
31///
32/// assert_eq!(values, vec![1, 2, 3, 4]);
33/// assert_eq!(terminations, vec![Termination::Completed]);
34/// ```
35#[derive(Educe)]
36#[educe(Debug, Clone)]
37pub struct StartWith<OE, I> {
38    source: OE,
39    values: I,
40}
41
42impl<OE, I> StartWith<OE, I> {
43    pub fn new<'or, T, E>(source: OE, values: I) -> Self
44    where
45        OE: Observable<'or, T, E>,
46        I: IntoIterator<Item = T>,
47    {
48        Self { source, values }
49    }
50}
51
52impl<'or, T, E, OE, I> Observable<'or, T, E> for StartWith<OE, I>
53where
54    OE: Observable<'or, T, E>,
55    I: IntoIterator<Item = T>,
56{
57    type D = OptionDisposal<Subscription<OE::D>>;
58
59    fn subscribe(
60        self,
61        mut observer: impl Observer<T, E> + MaybeSend + 'or,
62    ) -> Subscription<Self::D> {
63        for value in self.values.into_iter() {
64            if observer.on_next(value).is_stop() {
65                // The prepended values ended the stream, so the source is never subscribed to and
66                // there is nothing to dispose of.
67                return OptionDisposal::none().into_subscription();
68            }
69        }
70        self.source
71            .subscribe(observer)
72            .into_option()
73            .into_subscription()
74    }
75}
rx_rust/operators/combining/start_with.rs

rx_rust/operators/combining/
start_with.rs
