Struct Window 

pub struct Window<OE, OE1> { /* private fields */ }

Periodically subdivides items from an Observable into Observable windows.

A new window is emitted whenever the boundary Observable emits an item. Completing the boundary stops future window rotation without terminating the current window or the outer Observable. An error from the boundary terminates the current window and the outer Observable.

Each window is a single-consumer pipe: it can be subscribed to once, it buffers the items that arrive while it has no subscriber, and dropping it without subscribing discards its items. A window is serialized on its own rather than together with the outer Observable, so the events of a window keep their order among themselves, but they are not ordered against the emission of a later window. Disposing the outer subscription drops the observer of the open window without notifying it.

Disposing the subscription of a single window does not necessarily release its observer where it happens: the window releases it on its next item, when it ends, or when the outer subscription is disposed, whichever comes first. See the unicast subject each window is built on. See https://reactivex.io/documentation/operators/window.html

Examples

use rx_rust::{
    observable::ObservableExt,
    observer::{Observer, Termination},
    operators::transforming::window::Window,
    subject::publish_subject::PublishSubject,
};
use std::{convert::Infallible, sync::{Arc, Mutex}};

let windows = Arc::new(Mutex::new(Vec::<Vec<i32>>::new()));
let terminations = Arc::new(Mutex::new(Vec::new()));
let inner_subscriptions = Arc::new(Mutex::new(Vec::new()));

let mut source: PublishSubject<'_, i32, Infallible> = PublishSubject::default();
let mut boundary: PublishSubject<'_, (), Infallible> = PublishSubject::default();
let windows_observer = Arc::clone(&windows);
let terminations_observer = Arc::clone(&terminations);
let inner_subscriptions_observer = Arc::clone(&inner_subscriptions);

let subscription = Window::new(source.clone(), boundary.clone()).subscribe_with_callback(
    move |window| {
        let index = {
            let mut windows = windows_observer.lock().unwrap();
            windows.push(Vec::new());
            windows.len() - 1
        };
        let windows_for_values = Arc::clone(&windows_observer);
        let sub = window.subscribe_with_callback(
            move |value| {
                windows_for_values.lock().unwrap()[index].push(value);
            },
            |_| {},
        );
        inner_subscriptions_observer.lock().unwrap().push(sub);
    },
    move |termination| terminations_observer
        .lock()
        .unwrap()
        .push(termination),
);

source.on_next(1);
source.on_next(2);
boundary.on_next(());
source.on_next(3);
source.on_termination(Termination::Completed);
drop(subscription);
inner_subscriptions.lock().unwrap().drain(..).for_each(drop);

assert_eq!(
    &*windows.lock().unwrap(),
    &[vec![1, 2], vec![3]]
);
assert_eq!(
    &*terminations.lock().unwrap(),
    &[Termination::Completed]
);

Implementations

impl<OE, OE1> Window<OE, OE1>

pub fn new<'or, T, E>(source: OE, boundary: OE1) -> Self

where OE: Observable<'or, T, E>, OE1: Observable<'or, (), E>,

Trait Implementations

impl<OE, OE1> Clone for Window<OE, OE1>

where OE: Clone, OE1: Clone,

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<OE, OE1> Debug for Window<OE, OE1>

where OE: Debug, OE1: Debug,

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<'or, T, E, OE, OE1> Observable<'or, UnicastObservable<'or, T, E>, E> for Window<OE, OE1>

where T: MaybeSend + 'or, E: Clone + MaybeSend + 'or, OE: Observable<'or, T, E>, OE::D: MaybeSend + 'or, OE1: Observable<'or, (), E>, OE1::D: MaybeSend + 'or,

type D = BoxedDisposal<'or>

fn subscribe( self, observer: impl Observer<UnicastObservable<'or, T, E>, E> + MaybeSend + 'or, ) -> Subscription<Self::D>

Subscribes an observer to this observable. When an observer is subscribed, it will start receiving events from the observable. The subscribe method returns a Subscription which can be used to unsubscribe the observer from the observable. We use Subscription struct instead of trait like impl Cancellable, because we need to cancel the subscription when the Subscription is dropped. It’s not possible to implement Drop for a trait object.