rx_rust/operators/transforming/
buffer.rs

1use crate::utils::types::{Mutable, NecessarySendSync, Shared};
2use crate::{
3    disposable::subscription::Subscription,
4    observable::Observable,
5    observer::{Observer, Termination},
6    utils::unsub_after_termination::subscribe_unsub_after_termination,
7};
8use crate::{safe_lock, safe_lock_option, safe_lock_option_observer, safe_lock_vec};
9use educe::Educe;
10
11/// Periodically gathers items from an Observable into bundles and emits these bundles as `Vec<T>`, when a `boundary` Observable emits an item.
12/// See <https://reactivex.io/documentation/operators/buffer.html>
13///
14/// # Examples
15/// ```rust
16/// use rx_rust::{
17///     observable::observable_ext::ObservableExt,
18///     observer::{Observer, Termination},
19///     operators::transforming::buffer::Buffer,
20///     subject::publish_subject::PublishSubject,
21/// };
22/// use std::{convert::Infallible, sync::{Arc, Mutex}};
23///
24/// let values = Arc::new(Mutex::new(Vec::new()));
25/// let terminations = Arc::new(Mutex::new(Vec::new()));
26///
27/// let mut source: PublishSubject<'_, i32, Infallible> = PublishSubject::default();
28/// let mut boundary: PublishSubject<'_, (), Infallible> = PublishSubject::default();
29/// let values_observer = Arc::clone(&values);
30/// let terminations_observer = Arc::clone(&terminations);
31///
32/// let subscription = Buffer::new(source.clone(), boundary.clone()).subscribe_with_callback(
33///     move |value| values_observer.lock().unwrap().push(value),
34///     move |termination| terminations_observer
35///         .lock()
36///         .unwrap()
37///         .push(termination),
38/// );
39///
40/// source.on_next(1);
41/// source.on_next(2);
42/// boundary.on_next(());
43/// source.on_next(3);
44/// source.on_termination(Termination::Completed);
45/// drop(subscription);
46///
47/// assert_eq!(&*values.lock().unwrap(), &[vec![1, 2], vec![3]]);
48/// assert_eq!(
49///     &*terminations.lock().unwrap(),
50///     &[Termination::Completed]
51/// );
52/// ```
53#[derive(Educe)]
54#[educe(Debug, Clone)]
55pub struct Buffer<OE, OE1> {
56    source: OE,
57    boundary: OE1,
58}
59
60impl<OE, OE1> Buffer<OE, OE1> {
61    pub fn new<'or, 'sub, T, E>(source: OE, boundary: OE1) -> Self
62    where
63        OE: Observable<'or, 'sub, T, E>,
64        OE1: Observable<'or, 'sub, (), E>,
65    {
66        Self { source, boundary }
67    }
68}
69
70impl<'or, 'sub, T, E, OE, OE1> Observable<'or, 'sub, Vec<T>, E> for Buffer<OE, OE1>
71where
72    T: NecessarySendSync + 'or,
73    OE: Observable<'or, 'sub, T, E>,
74    OE1: Observable<'or, 'sub, (), E>,
75    'sub: 'or,
76{
77    fn subscribe(
78        self,
79        observer: impl Observer<Vec<T>, E> + NecessarySendSync + 'or,
80    ) -> Subscription<'sub> {
81        subscribe_unsub_after_termination(observer, |observer| {
82            let observer = Shared::new(Mutable::new(Some(observer)));
83            let values = Shared::new(Mutable::new(Vec::default()));
84            let subscription_1 = self.boundary.subscribe(BoundaryObserver {
85                observer: observer.clone(),
86                values: values.clone(),
87            });
88            let subscription_2 = self.source.subscribe(BufferObserver { observer, values });
89            subscription_1 + subscription_2
90        })
91    }
92}
93
94struct BufferObserver<T, OR> {
95    observer: Shared<Mutable<Option<OR>>>,
96    values: Shared<Mutable<Vec<T>>>,
97}
98
99impl<T, E, OR> Observer<T, E> for BufferObserver<T, OR>
100where
101    OR: Observer<Vec<T>, E>,
102{
103    fn on_next(&mut self, value: T) {
104        safe_lock_vec!(push: self.values, value);
105    }
106
107    fn on_termination(self, termination: Termination<E>) {
108        match termination {
109            Termination::Completed => {
110                let values = safe_lock!(mem_take: self.values);
111                if !values.is_empty() {
112                    safe_lock_option_observer!(on_next_and_termination: self.observer, values, termination);
113                } else {
114                    safe_lock_option_observer!(on_termination: self.observer, termination);
115                }
116            }
117            Termination::Error(_) => {
118                safe_lock_option_observer!(on_termination: self.observer, termination);
119            }
120        }
121    }
122}
123
124struct BoundaryObserver<T, OR> {
125    observer: Shared<Mutable<Option<OR>>>,
126    values: Shared<Mutable<Vec<T>>>,
127}
128
129impl<T, E, OR> Observer<(), E> for BoundaryObserver<T, OR>
130where
131    OR: Observer<Vec<T>, E>,
132{
133    fn on_next(&mut self, _: ()) {
134        let values = safe_lock!(mem_take: self.values);
135        safe_lock_option_observer!(on_next: self.observer, values);
136    }
137
138    fn on_termination(self, termination: Termination<E>) {
139        match termination {
140            Termination::Completed => {
141                let values = safe_lock!(mem_take: self.values);
142                if !values.is_empty() {
143                    safe_lock_option_observer!(on_next_and_termination: self.observer, values, termination);
144                } else {
145                    safe_lock_option_observer!(on_termination: self.observer, termination);
146                }
147            }
148            Termination::Error(_) => {
149                safe_lock_option_observer!(on_termination: self.observer, termination);
150            }
151        }
152    }
153}
rx_rust/operators/transforming/buffer.rs

rx_rust/operators/transforming/
buffer.rs
