rx_rust/operators/creating/
from_iter.rs

1use crate::utils::types::MaybeSend;
2use crate::{
3    observable::{Observable, Subscription},
4    observer::{Observer, Termination},
5};
6use educe::Educe;
7use std::convert::Infallible;
8
9/// Converts an `IntoIterator` into an Observable.
10/// See <https://reactivex.io/documentation/operators/from.html>
11///
12/// # Examples
13/// ```rust
14/// use rx_rust::{
15///     observable::ObservableExt,
16///     observer::Termination,
17///     operators::creating::from_iter::FromIter,
18/// };
19///
20/// let mut values = Vec::new();
21/// let mut terminations = Vec::new();
22///
23/// FromIter::new(vec![1, 2, 3]).subscribe_with_callback(
24///     |value| values.push(value),
25///     |termination| terminations.push(termination),
26/// );
27///
28/// assert_eq!(values, vec![1, 2, 3]);
29/// assert_eq!(terminations, vec![Termination::Completed]);
30/// ```
31#[derive(Educe)]
32#[educe(Debug, Clone)]
33pub struct FromIter<I>(I);
34
35impl<I> FromIter<I> {
36    pub fn new(into_iterator: I) -> Self {
37        Self(into_iterator)
38    }
39}
40
41impl<'or, T, I> Observable<'or, T, Infallible> for FromIter<I>
42where
43    I: IntoIterator<Item = T>,
44{
45    type D = ();
46
47    fn subscribe(
48        self,
49        mut observer: impl Observer<T, Infallible> + MaybeSend + 'or,
50    ) -> Subscription<Self::D> {
51        for value in self.0.into_iter() {
52            if observer.on_next(value).is_stop() {
53                // The observer ended its own stream, so the iteration stops here instead of
54                // running to an end an infinite iterator never reaches, and nothing is completed:
55                // the observer is released like a disposed one. Nothing else could stop it — the
56                // subscription only exists once this returns.
57                return Subscription::default();
58            }
59        }
60        observer.on_termination(Termination::Completed);
61        Subscription::default()
62    }
63}
rx_rust/operators/creating/from_iter.rs

rx_rust/operators/creating/
from_iter.rs
