rx_rust/operators/transforming/
buffer.rs

1use crate::utils::serialized_delivery::UpdateOutcome;
2use crate::utils::subscribe_with_context::{
3    self, SubscriptionContext, subscribe_with_context_owning_source,
4};
5use crate::utils::types::MaybeSend;
6use crate::{
7    disposable::Disposable,
8    observable::Observable,
9    observable::Subscription,
10    observer::{Flow, Observer, Termination},
11};
12use educe::Educe;
13
14/// Periodically gathers items from an Observable into bundles and emits these bundles as `Vec<T>`, when a `boundary` Observable emits an item.
15///
16/// Terminating the `boundary` terminates the outer Observable: completing it emits the pending
17/// bundle when non-empty and then completes, while an error from it discards the pending bundle
18/// and errors. Note that this differs from
19/// [`Window`](crate::operators::transforming::window::Window), where completing the `boundary`
20/// only stops rotation; a buffer has no consumer to release items to, so leaving it open after
21/// the `boundary` completed would accumulate items without bound.
22/// See <https://reactivex.io/documentation/operators/buffer.html>
23///
24/// # Relation to `Window`
25///
26/// Conceptually a buffer is a window whose items are collected into a `Vec`, that is
27/// `source.window(boundary).concat_map(|window| window.to_vec())`. It is implemented on its own
28/// rather than as that composition, because the two are not equivalent:
29///
30/// - Completing the `boundary` terminates a buffer, as described above, but only stops the
31///   rotation of a window.
32/// - Completing the `source` with an empty pending bundle emits nothing for a buffer, whereas the
33///   composition emits a trailing empty `Vec`, because collecting the empty open window yields
34///   the initial value.
35/// - [`Window`](crate::operators::transforming::window::Window) requires `E: Clone`, since an
36///   error is delivered to both the current window and the outer Observable. A buffer has no
37///   window to deliver to and so places no such bound on `E`.
38///
39/// The composition also pays for machinery a buffer does not need: a subject per window, and one
40/// queued action per item. Where the three points above do not matter, the equivalence holds; the
41/// `test_equivalent_to_window_and_collect` test in `tests/buffer.rs` pins it down, alongside two
42/// tests that pin down the divergences.
43///
44/// # Examples
45/// ```rust
46/// use rx_rust::{
47///     observable::ObservableExt,
48///     observer::{Observer, Termination},
49///     operators::transforming::buffer::Buffer,
50///     subject::publish_subject::PublishSubject,
51/// };
52/// use std::{convert::Infallible, sync::{Arc, Mutex}};
53///
54/// let values = Arc::new(Mutex::new(Vec::new()));
55/// let terminations = Arc::new(Mutex::new(Vec::new()));
56///
57/// let mut source: PublishSubject<'_, i32, Infallible> = PublishSubject::default();
58/// let mut boundary: PublishSubject<'_, (), Infallible> = PublishSubject::default();
59/// let values_observer = Arc::clone(&values);
60/// let terminations_observer = Arc::clone(&terminations);
61///
62/// let subscription = Buffer::new(source.clone(), boundary.clone()).subscribe_with_callback(
63///     move |value| values_observer.lock().unwrap().push(value),
64///     move |termination| terminations_observer
65///         .lock()
66///         .unwrap()
67///         .push(termination),
68/// );
69///
70/// source.on_next(1);
71/// source.on_next(2);
72/// boundary.on_next(());
73/// source.on_next(3);
74/// source.on_termination(Termination::Completed);
75/// drop(subscription);
76///
77/// assert_eq!(&*values.lock().unwrap(), &[vec![1, 2], vec![3]]);
78/// assert_eq!(
79///     &*terminations.lock().unwrap(),
80///     &[Termination::Completed]
81/// );
82/// ```
83#[derive(Educe)]
84#[educe(Debug, Clone)]
85pub struct Buffer<OE, OE1> {
86    source: OE,
87    boundary: OE1,
88}
89
90impl<OE, OE1> Buffer<OE, OE1> {
91    pub fn new<'or, T, E>(source: OE, boundary: OE1) -> Self
92    where
93        OE: Observable<'or, T, E>,
94        OE1: Observable<'or, (), E>,
95    {
96        Self { source, boundary }
97    }
98}
99
100impl<'or, T, E, OE, OE1> Observable<'or, Vec<T>, E> for Buffer<OE, OE1>
101where
102    T: MaybeSend + 'or,
103    E: MaybeSend + 'or,
104    OE: Observable<'or, T, E>,
105    OE::D: MaybeSend + 'or,
106    OE1: Observable<'or, (), E>,
107    OE1::D: MaybeSend + 'or,
108{
109    type D = subscribe_with_context::OwningDisposal<'or>;
110
111    fn subscribe(
112        self,
113        observer: impl Observer<Vec<T>, E> + MaybeSend + 'or,
114    ) -> Subscription<Self::D> {
115        subscribe_with_context_owning_source(observer, Vec::new(), |context| {
116            let subscription_1 = self.boundary.subscribe(BoundaryObserver(context.clone()));
117            let subscription_2 = self.source.subscribe(BufferObserver(context));
118            subscription_1.preceded_by_bound(subscription_2)
119        })
120    }
121}
122
123struct BufferObserver<T, E, OR, D: Disposable>(SubscriptionContext<Vec<T>, E, OR, Vec<T>, D>);
124
125impl<T, E, OR, D> Observer<T, E> for BufferObserver<T, E, OR, D>
126where
127    OR: Observer<Vec<T>, E>,
128    D: Disposable,
129{
130    fn on_next(&mut self, value: T) -> Flow {
131        self.0.update_flow(|values| {
132            values.push(value);
133            UpdateOutcome::empty()
134        })
135    }
136
137    fn on_termination(self, termination: Termination<E>) {
138        terminate(self.0, termination);
139    }
140}
141
142struct BoundaryObserver<T, E, OR, D: Disposable>(SubscriptionContext<Vec<T>, E, OR, Vec<T>, D>);
143
144impl<T, E, OR, D> Observer<(), E> for BoundaryObserver<T, E, OR, D>
145where
146    OR: Observer<Vec<T>, E>,
147    D: Disposable,
148{
149    fn on_next(&mut self, _: ()) -> Flow {
150        self.0.update_flow(|values| {
151            UpdateOutcome::empty()
152                .with_next_event(std::mem::replace(values, Vec::with_capacity(values.len())))
153        })
154    }
155
156    fn on_termination(self, termination: Termination<E>) {
157        terminate(self.0, termination);
158    }
159}
160
161fn terminate<T, E, OR, D>(
162    context: SubscriptionContext<Vec<T>, E, OR, Vec<T>, D>,
163    termination: Termination<E>,
164) where
165    OR: Observer<Vec<T>, E>,
166    D: Disposable,
167{
168    match termination {
169        completion @ Termination::Completed => {
170            let _ = context.update(|values| {
171                if values.is_empty() {
172                    UpdateOutcome::empty().with_termination_event(completion)
173                } else {
174                    UpdateOutcome::empty()
175                        .with_next_and_termination_events(std::mem::take(values), completion)
176                }
177            });
178        }
179        error @ Termination::Error(_) => {
180            context.send_termination(error);
181        }
182    }
183}
rx_rust/operators/transforming/buffer.rs

rx_rust/operators/transforming/
buffer.rs
