Struct reflux::Filter

source ·

pub struct Filter { /* private fields */ }

Expand description

An object that uses a predicate function to determine whether some data can be passed through a network node

Using a Filter yields the following benefits:

Conditionally allow data to flow through a Reflux network

§Example

 #![feature(coroutines, coroutine_trait, stmt_expr_attributes)]
 #![feature(unboxed_closures)]
use reflux::Filter;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
 use crossbeam_channel::Receiver;
 use reflux::add_routine;
 use crossbeam_channel::unbounded;
use std::time::Duration;
use std::thread::sleep;
let stop_flag = Arc::new(AtomicBool::new(false));
let fun = |data: &String| -> bool {
    data.contains("hello")
 };
 
 let stop_flag = Arc::new(AtomicBool::new(false));
 let (tx, rx) = unbounded();
 
 let (filter, filter_sink) = Filter::new(fun, rx, stop_flag.clone());
 
 tx.send("hello world".to_string()).unwrap();
 let data = filter_sink.recv().unwrap();
 
 assert_eq!(data, "hello world");
 
 tx.send("goodbye world".to_string()).unwrap();
 let res = filter_sink.recv_timeout(Duration::from_secs(1));
 assert!(res.is_err());
 
 tx.send("hello there".to_string()).unwrap();
 let data = filter_sink.recv().unwrap();
 
 assert_eq!(data, "hello there");
         
 stop_flag.store(true, Ordering::Relaxed);
 filter.join().unwrap()

Implementations§

source §

impl Filter

source

pub fn new<T, F>( filter_fn: F, source: Receiver<T>, stop_sig: Arc<AtomicBool>, ) -> (Self, Receiver<T>)
where T: Send + 'static, F: Fn(&T) -> bool + Send + 'static,

Creates a new Filter object with an unbounded internal channel.

§Parameters

filter - A function that takes a reference, determines if the data meets some condition and returns a boolean.
source - A Receiver channel object from which to receive data.
stop_sig - A flag to signal the Inlet object to terminate

§Returns

A Filter object
A Receiver

source

pub fn new_unbounded<T, F>( filter_fn: F, source: Receiver<T>, stop_sig: Arc<AtomicBool>, data_limit: Option<usize>, ) -> (Self, Receiver<T>)
where T: Send + 'static, F: Fn(&T) -> bool + Send + 'static,

Creates a new Filter object with a bounded internal channel.

§Parameters

filter - A function that takes a reference, determines if the data meets some condition and returns a boolean.
source - A Receiver channel object from which to receive data.
stop_sig - A flag to signal the Inlet object to terminate