rx_rust/operators/transforming/
window.rs

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

rx_rust/operators/transforming/
window.rs
