Struct Filter

pub struct Filter { /* private fields */ }

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, None, stop_flag.clone(), None);
 
 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

impl Filter

pub fn new<T, F>( filter_fn: F, source: Receiver<T>, pause_sig: Option<Arc<AtomicBool>>, 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.

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.
• pause_sig - A flag to signal the Filter object to pause execution.
• stop_sig - A flag to signal the Filter object to terminate execution.

Returns

• A Filter object.
• A Receiver channel.

pub fn join(self) -> Result<()>

Waits for the Filter object to finish execution.