Struct reflux::Transformer

pub struct Transformer<I, O, E> { /* private fields */ }

An object that enables data mutation. After processing data a Transformer can yield three variants:

• TransformerResult::Transformed - The data has been processed and can move along the rest of the Reflux network
• TransformerResult::NeedsMoreWork - The data still needs additional processing. The data is sent back into the Transformer
• TransformerResult::Error - There was an error in processing the data. This error is simply logged to stderr

Using a Transformer yields the following benefits:

• Mutation of data in a Reflux network.

Example

 #![feature(coroutines, coroutine_trait, stmt_expr_attributes)]
 #![feature(unboxed_closures)]
 use reflux::{Transformer, TransformerContext, TransformerResult};
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
 use crossbeam_channel::{Receiver, Sender};
 use reflux::add_routine;
 use crossbeam_channel::unbounded;
 use std::time::Duration;
 use std::thread::sleep;
 
 #[derive(Clone)]
 struct InnerContext {
    inc_val: i32
 }

 let ctx = InnerContext {
    inc_val: 1
 };

 let stop_flag = Arc::new(AtomicBool::new(false));

 let (transformer, input, output): (Transformer<i32, String, String>, Sender<i32>, Receiver<String>) = Transformer::new(
    add_routine!(#[coroutine] |input: TransformerContext<i32, InnerContext>| {
        let mut data = input.data.unwrap();
        while data < 5 {
            data += input.globals.inc_val;
            yield TransformerResult::NeedsMoreWork(data);
        }
    yield TransformerResult::Transformed(format!("The number is {data}"));
 }), stop_flag.clone(), ctx);

 input.send(0).unwrap();

 let result = output.recv().unwrap();
 stop_flag.store(true, Ordering::Relaxed);
 transformer.join().unwrap();

 assert_eq!(result, "The number is 5".to_string())

Implementations

impl<I, O, E> Transformer<I, O, E>

pub fn new<Ctx, F, C>( transform_fn: F, stop_sig: Arc<AtomicBool>, context: Ctx, ) -> (Self, Sender<I>, Receiver<O>)

where F: Fn() -> C + Send + 'static, C: Coroutine<TransformerContext<I, Ctx>> + Send + 'static + Unpin, Ctx: Send + 'static + Clone, I: Send + 'static, O: Send + 'static, E: Send + 'static + Display, TransformerResult<I, O, E>: Send + 'static + From<<C as Coroutine<TransformerContext<I, Ctx>>>::Yield>,

Creates a new Transformer object with an unbounded internal channel.

Parameters

• transform_fn - A coroutine that will transform data. The use of the add_routine! macro is necessary when passing in a coroutine.
• stop_sig - A flag to signal the Router object to terminate
• context - An object of immutable values for the transformer_fn to use during computation

Returns

A Transformer object

pub fn new_unbounded<Ctx, F, C>( transform_fn: F, stop_sig: Arc<AtomicBool>, context: Ctx, data_limit: Option<usize>, ) -> (Self, Sender<I>, Receiver<O>)

where F: Fn() -> C + Send + 'static, C: Coroutine<TransformerContext<I, Ctx>> + Send + 'static + Unpin, Ctx: Send + 'static + Clone, I: Send + 'static, O: Send + 'static, E: Send + 'static + Display, TransformerResult<I, O, E>: Send + 'static + From<<C as Coroutine<TransformerContext<I, Ctx>>>::Yield>,

Creates a new Transformer object with a bounded internal channel.

Parameters

• transform_fn - A coroutine that will transform data. The use of the add_routine! macro is necessary when passing in a coroutine.
• stop_sig - A flag to signal the Router object to terminate
• context - An object of immutable values for the transformer_fn to use during computation

Returns

A Transformer object

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

Waits for the Transformer object to finish execution