rx_rust/operators/transforming/
window_with_count.rs

1use crate::disposable::{DisposableExt, option_disposal::OptionDisposal};
2use crate::utils::types::MaybeSend;
3use crate::{
4    observable::Observable,
5    observable::Subscription,
6    observer::{Flow, Observer, Termination},
7    subject::unicast_subject::{UnicastObservable, UnicastSender, unicast_subject},
8};
9use educe::Educe;
10use std::{cmp::Ordering, num::NonZeroUsize};
11
12/// Periodically subdivides items from an Observable into Observable windows, each containing a specified number of items.
13///
14/// A window is emitted before its first item is delivered, and each window can be subscribed to
15/// once. Items emitted while a window has no subscriber are buffered and replayed to a later
16/// subscriber; dropping a window without subscribing to it discards its items.
17///
18/// Disposing the subscription of a single window does not necessarily release its observer where
19/// it happens: the window releases it on its next item, when it ends, or when the outer
20/// subscription is disposed, whichever comes first. See
21/// [the unicast subject](crate::subject::unicast_subject) each window is built on.
22/// See <https://reactivex.io/documentation/operators/window.html>
23///
24/// # Examples
25/// ```rust
26/// use rx_rust::{
27///     observable::ObservableExt,
28///     observer::Termination,
29///     operators::{
30///         creating::from_iter::FromIter,
31///         transforming::window_with_count::WindowWithCount,
32///     },
33/// };
34/// use std::{num::NonZeroUsize, sync::{Arc, Mutex}};
35///
36/// let windows = Arc::new(Mutex::new(Vec::<Vec<i32>>::new()));
37/// let terminations = Arc::new(Mutex::new(Vec::new()));
38/// let inner_subscriptions = Arc::new(Mutex::new(Vec::new()));
39/// let windows_observer = Arc::clone(&windows);
40/// let terminations_observer = Arc::clone(&terminations);
41/// let inner_subscriptions_observer = Arc::clone(&inner_subscriptions);
42///
43/// let subscription = WindowWithCount::new(
44///     FromIter::new(vec![1, 2, 3, 4]),
45///     NonZeroUsize::new(2).unwrap(),
46/// )
47/// .subscribe_with_callback(
48///     move |window| {
49///         let index = {
50///             let mut windows = windows_observer.lock().unwrap();
51///             windows.push(Vec::new());
52///             windows.len() - 1
53///         };
54///         let windows_for_values = Arc::clone(&windows_observer);
55///         let sub = window.subscribe_with_callback(
56///             move |value| {
57///                 windows_for_values.lock().unwrap()[index].push(value);
58///             },
59///             |_| {},
60///         );
61///         inner_subscriptions_observer.lock().unwrap().push(sub);
62///     },
63///     move |termination| terminations_observer
64///         .lock()
65///         .unwrap()
66///         .push(termination),
67/// );
68///
69/// drop(subscription);
70/// inner_subscriptions.lock().unwrap().drain(..).for_each(drop);
71///
72/// assert_eq!(
73///     &*windows.lock().unwrap(),
74///     &[vec![1, 2], vec![3, 4], vec![]]
75/// );
76/// assert_eq!(
77///     &*terminations.lock().unwrap(),
78///     &[Termination::Completed]
79/// );
80/// ```
81#[derive(Educe)]
82#[educe(Debug, Clone)]
83pub struct WindowWithCount<OE> {
84    source: OE,
85    count: NonZeroUsize,
86}
87
88impl<OE> WindowWithCount<OE> {
89    pub fn new(source: OE, count: NonZeroUsize) -> Self {
90        Self { source, count }
91    }
92}
93
94impl<'or, T, E, OE> Observable<'or, UnicastObservable<'or, T, E>, E> for WindowWithCount<OE>
95where
96    T: MaybeSend + 'or,
97    E: Clone + MaybeSend + 'or,
98    OE: Observable<'or, T, E>,
99{
100    type D = OptionDisposal<Subscription<OE::D>>;
101
102    fn subscribe(
103        self,
104        mut observer: impl Observer<UnicastObservable<'or, T, E>, E> + MaybeSend + 'or,
105    ) -> Subscription<Self::D> {
106        let (sender, window) = unicast_subject();
107        if observer.on_next(window).is_stop() {
108            // The first window ended the stream, so the source is never subscribed to.
109            return OptionDisposal::none().into_subscription();
110        }
111
112        let observer = WindowWithCountObserver {
113            observer,
114            sender,
115            count: self.count,
116            sent_count: 0,
117        };
118        self.source
119            .subscribe(observer)
120            .into_option()
121            .into_subscription()
122    }
123}
124
125struct WindowWithCountObserver<'or, T, E, OR> {
126    observer: OR,
127    sender: UnicastSender<'or, T, E>,
128    count: NonZeroUsize,
129    sent_count: usize,
130}
131
132impl<'or, T, E, OR> Observer<T, E> for WindowWithCountObserver<'or, T, E, OR>
133where
134    E: Clone,
135    OR: Observer<UnicastObservable<'or, T, E>, E>,
136{
137    fn on_next(&mut self, value: T) -> Flow {
138        // The consumer of one window stops that window, not the operator: only what the observer
139        // of the windows themselves answers can stop the source.
140        match (self.sent_count + 1).cmp(&self.count.get()) {
141            Ordering::Less => {
142                let _ = self.sender.on_next(value);
143                self.sent_count += 1;
144                Flow::Continue
145            }
146            Ordering::Equal => {
147                let (new_sender, new_window) = unicast_subject();
148                let mut old_sender = std::mem::replace(&mut self.sender, new_sender);
149                let _ = old_sender.on_next(value);
150                old_sender.on_termination(Termination::Completed);
151                self.sent_count = 0;
152                self.observer.on_next(new_window)
153            }
154            Ordering::Greater => unreachable!(),
155        }
156    }
157
158    fn on_termination(self, termination: Termination<E>) {
159        self.sender.on_termination(termination.clone());
160        self.observer.on_termination(termination);
161    }
162}
rx_rust/operators/transforming/window_with_count.rs

rx_rust/operators/transforming/
window_with_count.rs
