mumu-flow 0.1.0

Stream tranform tools plugin for the Lava language
Documentation
use mumu::parser::interpreter::{Interpreter, DynamicFnInfo};
use mumu::parser::interpreter::apply_n_ary_function_value;
use mumu::parser::types::{
    Value,
    FunctionValue,
    InkIteratorHandle,
    InkIteratorKind,
};
use std::sync::{Arc, Mutex};

pub fn register_flow_filter(interp: &mut Interpreter) {
    let func = Arc::new(Mutex::new(flow_filter_bridge_fn));
    let info = DynamicFnInfo::new(func, true);
    interp.register_dynamic_function_ex("flow:filter", info);

    interp.set_variable(
        "flow:filter",
        Value::Function(Box::new(FunctionValue::Named("flow:filter".to_string())))
    );
}

fn flow_filter_bridge_fn(_interp: &mut Interpreter, args: Vec<Value>) -> Result<Value, String> {
    match args.len() {
        0 => Ok(make_partial(None, None)),
        1 => {
            let val = &args[0];
            if is_placeholder(val) {
                Ok(make_partial(None, None))
            } else if let Value::Function(fb) = val {
                Ok(make_partial(Some(fb.clone()), None))
            } else {
                Err(format!("flow:filter => first argument must be a predicate function or '_', got {:?}", val))
            }
        }
        2 => {
            let pred_val = &args[0];
            let source_val = &args[1];

            if is_placeholder(pred_val) || is_placeholder(source_val) {
                let maybe_pred = match pred_val {
                    Value::Function(fb) => Some(fb.clone()),
                    _ if is_placeholder(pred_val) => None,
                    other => return Err(format!("flow:filter => first param must be function or '_', got {:?}", other)),
                };
                let maybe_source = if is_placeholder(source_val) {
                    None
                } else {
                    Some(source_val.clone())
                };
                if let (Some(fb), Some(sv)) = (maybe_pred.clone(), maybe_source.clone()) {
                    finalize_filter(_interp, fb, sv)
                } else {
                    Ok(make_partial(maybe_pred, maybe_source))
                }
            } else {
                if let Value::Function(fb) = pred_val {
                    finalize_filter(_interp, fb.clone(), source_val.clone())
                } else {
                    Err(format!("flow:filter => first param must be a Function, got {:?}", pred_val))
                }
            }
        }
        n => Err(format!("flow:filter => expected up to 2 arguments, got {}", n)),
    }
}

fn finalize_filter(_interp: &mut Interpreter, predicate_func: Box<FunctionValue>, source_val: Value) -> Result<Value, String> {
    match source_val {
        Value::InkIterator(handle) => Ok(build_filter_from_iter(predicate_func, handle)),
        Value::InkTransform(tf)    => Ok(build_filter_chain(predicate_func, tf)),
        other => Err(format!("flow:filter => second param must be InkIterator or InkTransform, got {:?}", other)),
    }
}

fn make_partial(pred: Option<Box<FunctionValue>>, source: Option<Value>) -> Value {
    use FunctionValue::RustClosure;

    let state = FilterPartialState { pred, source };
    let env = Arc::new(state);

    let closure = RustClosure(
        "flow:filter-partial".to_string(),
        Arc::new(Mutex::new(move |interp: &mut Interpreter, new_args: Vec<Value>| {
            let mut new_pred = env.pred.clone();
            let mut new_source = env.source.clone();

            for val in &new_args {
                if new_pred.is_none() {
                    if is_placeholder(val) {
                        // remain None
                    } else if let Value::Function(fb) = val {
                        new_pred = Some(fb.clone());
                    } else {
                        return Err(format!("flow:filter => first param must be function or '_', got {:?}", val));
                    }
                } else if new_source.is_none() {
                    if is_placeholder(val) {
                        // remain None
                    } else {
                        new_source = Some(val.clone());
                    }
                } else {
                    return Err("flow:filter => partial => too many arguments".to_string());
                }
            }

            if let (Some(fb), Some(sv)) = (new_pred.clone(), new_source.clone()) {
                finalize_filter(interp, fb, sv)
            } else {
                Ok(make_partial(new_pred, new_source))
            }
        })),
        0
    );

    Value::Function(Box::new(closure))
}

fn build_filter_from_iter(pred: Box<FunctionValue>, iter: InkIteratorHandle) -> Value {
    use FunctionValue::RustClosure;

    let env = Arc::new(FilterEnvIter {
        pred,
        iter,
    });

    let closure = RustClosure(
        "flow:filter-iterator".to_string(),
        Arc::new(Mutex::new(move |interp: &mut Interpreter, _args: Vec<Value>| {
            match &env.iter.kind {
                InkIteratorKind::Core(state_arc) => {
                    loop {
                        let mut guard = state_arc.lock()
                            .map_err(|_| "flow:filter => lock error on InkIterator".to_string())?;
                        if guard.done || guard.current >= guard.end {
                            guard.done = true;
                            return Err("NO_MORE_DATA".to_string());
                        }
                        let item_val = Value::Int(guard.current);
                        guard.current += 1;
                        if guard.current >= guard.end {
                            guard.done = true;
                        }
                        drop(guard);

                        let pred_res = apply_n_ary_function_value(interp, env.pred.clone(), vec![item_val.clone()])?;
                        let passes = is_truthy(&pred_res)?;
                        if passes {
                            return Ok(item_val);
                        }
                    }
                }
                InkIteratorKind::Plugin(plugin_arc) => {
                    let mut plugin = plugin_arc.lock().map_err(|_| "flow:filter => plugin lock error".to_string())?;
                    loop {
                        match plugin.next_value() {
                            Ok(item_val) => {
                                let pred_res = apply_n_ary_function_value(interp, env.pred.clone(), vec![item_val.clone()])?;
                                let passes = is_truthy(&pred_res)?;
                                if passes {
                                    return Ok(item_val);
                                }
                            }
                            Err(e) if e == "NO_MORE_DATA" => return Err(e),
                            Err(e) => return Err(e),
                        }
                    }
                }
            }
        })),
        0
    );

    Value::InkTransform(Box::new(closure))
}

fn build_filter_chain(pred: Box<FunctionValue>, prev_tf: Box<FunctionValue>) -> Value {
    use FunctionValue::RustClosure;

    let env = Arc::new(FilterEnvChain {
        pred,
        prev_tf,
    });

    let closure = RustClosure(
        "flow:filter-transform".to_string(),
        Arc::new(Mutex::new(move |interp: &mut Interpreter, _args: Vec<Value>| {
            loop {
                let item_res = apply_n_ary_function_value(interp, env.prev_tf.clone(), vec![]);
                match item_res {
                    Ok(item_val) => {
                        let pred_res = apply_n_ary_function_value(interp, env.pred.clone(), vec![item_val.clone()])?;
                        let passes = is_truthy(&pred_res)?;
                        if passes {
                            return Ok(item_val);
                        }
                    }
                    Err(e) => {
                        return Err(e);
                    }
                }
            }
        })),
        0
    );

    Value::InkTransform(Box::new(closure))
}

#[derive(Clone)]
struct FilterPartialState {
    pred: Option<Box<FunctionValue>>,
    source: Option<Value>,
}

#[derive(Clone)]
struct FilterEnvIter {
    pred: Box<FunctionValue>,
    iter: InkIteratorHandle,
}

#[derive(Clone)]
struct FilterEnvChain {
    pred: Box<FunctionValue>,
    prev_tf: Box<FunctionValue>,
}

fn is_truthy(v: &Value) -> Result<bool, String> {
    match v {
        Value::Bool(b)  => Ok(*b),
        Value::Int(i)   => Ok(*i != 0),
        Value::Float(f) => Ok(*f != 0.0),
        Value::Long(l)  => Ok(*l != 0),
        _ => Err(format!("flow:filter => predicate must return bool/int/float/long, got {:?}", v)),
    }
}

fn is_placeholder(val: &Value) -> bool {
    match val {
        Value::Placeholder => true,
        Value::SingleString(s) if s == "_" => true,
        Value::StrArray(ss) if ss.len() == 1 && ss[0] == "_" => true,
        _ => false,
    }
}