Function callbag::filter

pub fn filter<I: 'static, F: 'static, S>(
    condition: F
) -> Box<dyn Fn(S) -> Source<I>> where
    F: Fn(&I) -> bool + Send + Sync + Clone,
    S: Into<Arc<Source<I>>>, 

Callbag operator that conditionally lets data pass through.

Works on either pullable or listenable sources.

See https://github.com/staltz/callbag-filter/blob/01212b2d17622cae31545200235e9db3f1b0e235/readme.js#L23-L36

Examples

use crossbeam_queue::SegQueue;
use std::sync::Arc;

use callbag::{filter, for_each, from_iter};

let actual = Arc::new(SegQueue::new());

let source = filter(|x| x % 2 == 1)(from_iter([1, 2, 3, 4, 5]));

for_each({
    let actual = Arc::clone(&actual);
    move |x| {
        println!("{}", x);
        actual.push(x);
    }
})(source);

assert_eq!(
    &{
        let mut v = vec![];
        for _i in 0..actual.len() {
            v.push(actual.pop().unwrap());
        }
        v
    }[..],
    [1, 3, 5]
);