flow-mumu 0.2.0-rc.9

// src/stop_when.rs
//
// flow:stop_when(pred)
// Stage that **terminates the pipeline** when `pred(item)` becomes truthy.
// - Returns a stage Function that expects a **source** (Iterator or zero-arg transform).
// - The produced transform yields items unchanged **until** the predicate
//   returns true for an item; it then **stops** (returns NO_MORE_DATA) and
//   does **not** emit that triggering item.
// - No sleeping/blocking. Temporary no-data errors (e.g., "AGAIN") are
//   propagated unchanged so cooperative scheduling keeps working.

use std::sync::{Arc, Mutex};
use mumu::parser::interpreter::{Interpreter, DynamicFnInfo};
use mumu::parser::interpreter::apply::apply_n_ary_function_value;
use mumu::parser::types::{FunctionValue, Value};
use crate::common::to_transform;

pub fn register_flow_stop_when(interp: &mut Interpreter) {
    let f = Arc::new(Mutex::new(flow_stop_when_bridge));
    let info = DynamicFnInfo::new(f, true);
    interp.register_dynamic_function_ex("flow:stop_when", info);

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

fn flow_stop_when_bridge(_interp: &mut Interpreter, args: Vec<Value>) -> Result<Value, String> {
    match args.len() {
        // Accept just the predicate now; compose will pass the source later.
        1 => {
            let pred = match &args[0] {
                Value::Function(fb) => fb.clone(),
                other => return Err(format!("flow:stop_when => first argument must be a Function, got {:?}", other)),
            };
            Ok(make_stage_partial(Some(pred), None))
        }
        // Support 2-arg direct form: (pred, source)
        2 => {
            let pred = match &args[0] {
                Value::Function(fb) => fb.clone(),
                other => return Err(format!("flow:stop_when => first argument must be a Function, got {:?}", other)),
            };
            let src = args[1].clone();
            finalize_stop_when(pred, src)
        }
        // Partial (await pred and/or source)
        0 => Ok(make_stage_partial(None, None)),
        n => Err(format!("flow:stop_when => expected 0, 1 or 2 arguments, got {}", n)),
    }
}

fn make_stage_partial(pred_opt: Option<Box<FunctionValue>>, src_opt: Option<Value>) -> Value {
    use mumu::parser::types::FunctionValue::RustClosure;

    Value::Function(Box::new(RustClosure(
        "flow:stop_when-partial".to_string(),
        Arc::new(Mutex::new(move |_interp: &mut Interpreter, new_args: Vec<Value>| {
            let mut pred = pred_opt.clone();
            let mut src  = src_opt.clone();

            for a in new_args {
                if pred.is_none() {
                    if let Value::Function(fb) = a {
                        pred = Some(fb);
                    } else {
                        return Err("flow:stop_when => expected predicate Function".to_string());
                    }
                    continue;
                }
                if src.is_none() {
                    src = Some(a);
                    continue;
                }
                return Err("flow:stop_when => too many arguments".to_string());
            }

            match (pred.clone(), src.clone()) {
                (Some(p), Some(s)) => finalize_stop_when(p, s),
                _ => Ok(make_stage_partial(pred, src)),
            }
        })),
        0
    )))
}

fn finalize_stop_when(pred: Box<FunctionValue>, source: Value) -> Result<Value, String> {
    let upstream_tf = to_transform(source)?;

    use mumu::parser::types::FunctionValue::RustClosure;

    #[derive(Debug)]
    struct Env {
        pred: Box<FunctionValue>,
        upstream: Box<FunctionValue>,
        done: bool,
    }

    let shared = Arc::new(Mutex::new(Env {
        pred,
        upstream: upstream_tf,
        done: false,
    }));

    let tf = RustClosure(
        "flow:stop_when-transform".to_string(),
        Arc::new(Mutex::new(move |interp: &mut Interpreter, _args: Vec<Value>| {
            let mut st = shared.lock().map_err(|_| "flow:stop_when => lock error".to_string())?;

            if st.done {
                return Err("NO_MORE_DATA".to_string());
            }

            // Pull one from upstream
            let item = match apply_n_ary_function_value(interp, st.upstream.clone(), vec![]) {
                Ok(v) => v,
                Err(e) => return Err(e), // propagate "AGAIN" or "NO_MORE_DATA" or other
            };

            // Evaluate predicate on the item
            let pred_res = apply_n_ary_function_value(interp, st.pred.clone(), vec![item.clone()])?;
            let stop_now = truthy(&pred_res)?;

            if stop_now {
                st.done = true;
                // DO NOT emit the triggering item; signal end of stream immediately
                return Err("NO_MORE_DATA".to_string());
            }

            Ok(item)
        })),
        0,
    );

    Ok(Value::Function(Box::new(tf)))
}

/// Minimal truthiness (align with flow:filter behavior).
fn 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),
        other => Err(format!("flow:stop_when => predicate must return bool/int/float/long, got {:?}", other)),
    }
}