1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38
//! Filters a stream by a predicate.
use crate::Data;
use crate::dataflow::channels::pact::Pipeline;
use crate::dataflow::{Stream, Scope};
use crate::dataflow::operators::generic::operator::Operator;
/// Extension trait for filtering.
pub trait Filter<D: Data> {
/// Returns a new instance of `self` containing only records satisfying `predicate`.
///
/// # Examples
/// ```
/// use timely::dataflow::operators::{ToStream, Filter, Inspect};
///
/// timely::example(|scope| {
/// (0..10).to_stream(scope)
/// .filter(|x| *x % 2 == 0)
/// .inspect(|x| println!("seen: {:?}", x));
/// });
/// ```
fn filter<P: FnMut(&D)->bool+'static>(&self, predicate: P) -> Self;
}
impl<G: Scope, D: Data> Filter<D> for Stream<G, D> {
fn filter<P: FnMut(&D)->bool+'static>(&self, mut predicate: P) -> Stream<G, D> {
let mut vector = Vec::new();
self.unary(Pipeline, "Filter", move |_,_| move |input, output| {
input.for_each(|time, data| {
data.swap(&mut vector);
vector.retain(|x| predicate(x));
if !vector.is_empty() {
output.session(&time).give_container(&mut vector);
}
});
})
}
}