Struct Buffer 

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

Periodically gathers items from an Observable into bundles and emits these bundles as Vec<T>, when a boundary Observable emits an item.

Terminating the boundary terminates the outer Observable: completing it emits the pending bundle when non-empty and then completes, while an error from it discards the pending bundle and errors. Note that this differs from Window, where completing the boundary only stops rotation; a buffer has no consumer to release items to, so leaving it open after the boundary completed would accumulate items without bound. See https://reactivex.io/documentation/operators/buffer.html

Relation to Window

Conceptually a buffer is a window whose items are collected into a Vec, that is source.window(boundary).concat_map(|window| window.to_vec()). It is implemented on its own rather than as that composition, because the two are not equivalent:

• Completing the boundary terminates a buffer, as described above, but only stops the rotation of a window.
• Completing the source with an empty pending bundle emits nothing for a buffer, whereas the composition emits a trailing empty Vec, because collecting the empty open window yields the initial value.
• Window requires E: Clone, since an error is delivered to both the current window and the outer Observable. A buffer has no window to deliver to and so places no such bound on E.

The composition also pays for machinery a buffer does not need: a subject per window, and one queued action per item. Where the three points above do not matter, the equivalence holds; the test_equivalent_to_window_and_collect test in tests/buffer.rs pins it down, alongside two tests that pin down the divergences.

Examples

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

let values = Arc::new(Mutex::new(Vec::new()));
let terminations = Arc::new(Mutex::new(Vec::new()));

let mut source: PublishSubject<'_, i32, Infallible> = PublishSubject::default();
let mut boundary: PublishSubject<'_, (), Infallible> = PublishSubject::default();
let values_observer = Arc::clone(&values);
let terminations_observer = Arc::clone(&terminations);

let subscription = Buffer::new(source.clone(), boundary.clone()).subscribe_with_callback(
    move |value| values_observer.lock().unwrap().push(value),
    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);

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

Implementations

impl<OE, OE1> Buffer<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 Buffer<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 Buffer<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, Vec<T>, E> for Buffer<OE, OE1>

where T: MaybeSend + 'or, E: 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<Vec<T>, 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.