timely/dataflow/operators/
filter.rs
use crate::Data;
use crate::dataflow::channels::pact::Pipeline;
use crate::dataflow::{Stream, Scope};
use crate::dataflow::operators::generic::operator::Operator;
pub trait Filter<D: Data> {
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> {
self.unary(Pipeline, "Filter", move |_,_| move |input, output| {
input.for_each(|time, data| {
data.retain(|x| predicate(x));
if !data.is_empty() {
output.session(&time).give_container(data);
}
});
})
}
}