rx_rust/operators/conditional_boolean/
contains.rs

1use crate::utils::subscribe_with_auto_dispose_on_termination;
2use crate::utils::subscribe_with_auto_dispose_on_termination::subscribe_with_auto_dispose_on_termination;
3use crate::utils::types::MaybeSend;
4use crate::{
5    observable::{Observable, Subscription},
6    observer::{Flow, Observer, Termination},
7};
8use educe::Educe;
9
10/// Emits a single boolean value that indicates whether a source Observable emits a specified item.
11/// See <https://reactivex.io/documentation/operators/contains.html>
12///
13/// # Examples
14/// ```rust
15/// use rx_rust::{
16///     observable::ObservableExt,
17///     observer::Termination,
18///     operators::{
19///         conditional_boolean::contains::Contains,
20///         creating::from_iter::FromIter,
21///     },
22/// };
23///
24/// let mut values = Vec::new();
25/// let mut terminations = Vec::new();
26///
27/// let observable = Contains::new(FromIter::new(vec![1, 2, 3]), 2);
28/// observable.subscribe_with_callback(
29///     |value| values.push(value),
30///     |termination| terminations.push(termination),
31/// );
32///
33/// assert_eq!(values, vec![true]);
34/// assert_eq!(terminations, vec![Termination::Completed]);
35/// ```
36#[derive(Educe)]
37#[educe(Debug, Clone)]
38pub struct Contains<T, OE> {
39    source: OE,
40    item: T,
41}
42
43impl<T, OE> Contains<T, OE> {
44    pub fn new<'or, E>(source: OE, item: T) -> Self
45    where
46        OE: Observable<'or, T, E>,
47    {
48        Self { source, item }
49    }
50}
51
52impl<'or, T, E, OE> Observable<'or, bool, E> for Contains<T, OE>
53where
54    OE: Observable<'or, T, E>,
55    OE::D: MaybeSend + 'or,
56    T: PartialEq + MaybeSend + 'or,
57{
58    type D = subscribe_with_auto_dispose_on_termination::Disposal<OE::D>;
59
60    fn subscribe(
61        self,
62        observer: impl Observer<bool, E> + MaybeSend + 'or,
63    ) -> Subscription<Self::D> {
64        subscribe_with_auto_dispose_on_termination(observer, |observer| {
65            let observer = ContainsObserver {
66                observer: Some(observer),
67                item: self.item,
68            };
69            self.source.subscribe(observer)
70        })
71    }
72}
73
74struct ContainsObserver<T, OR> {
75    observer: Option<OR>,
76    item: T,
77}
78
79impl<T, E, OR> Observer<T, E> for ContainsObserver<T, OR>
80where
81    OR: Observer<bool, E>,
82    T: PartialEq,
83{
84    fn on_next(&mut self, value: T) -> Flow {
85        if self.item != value {
86            return Flow::Continue;
87        }
88        // The item was found, so the result is decided and the rest of the source is of no use.
89        let Some(mut observer) = self.observer.take() else {
90            return Flow::Stop;
91        };
92        if observer.on_next(true).is_continue() {
93            observer.on_termination(Termination::Completed);
94        }
95        Flow::Stop
96    }
97
98    fn on_termination(mut self, termination: Termination<E>) {
99        if let Some(mut observer) = self.observer.take() {
100            match termination {
101                Termination::Completed => drop(observer.on_next(false)),
102                Termination::Error(_) => {}
103            }
104            observer.on_termination(termination);
105        }
106    }
107}
rx_rust/operators/conditional_boolean/contains.rs

rx_rust/operators/conditional_boolean/
contains.rs
