net-mumu 0.2.0-rc.3

// src/lldp/iterator.rs
//
// Iterator glue for the LLDP/CDP engine.
// --------------------------------------
// This module defines the tiny adapter that exposes any `LldpEngine`
// implementation to the MuMu runtime as a standard `IteratorHandle`
// (Plugin iterator), and a helper to turn that handle into a 0-arg
// transform function (for Flow pipelines).
//
// Contract:
//   • `LldpEngine::next()` returns:
//       Ok(Value)            — one event row (add/update/remove) per call
//       Err("AGAIN")         — nothing to emit right now (keep polling)
//       Err("NO_MORE_DATA")  — stream ended (only when an explicit cap is hit)
//   • We pass those outcomes through unchanged.
//
// This keeps the scheduling/backpressure model consistent across the
// net-mumu plugin and Flow operators.
//

use std::sync::{Arc, Mutex};

use mumu::parser::types::{FunctionValue, IteratorHandle, IteratorKind, PluginIterator, Value};

/// Minimal engine contract used by the iterator.
/// Implemented by the real engine (Linux) and any testing engines.
///
/// Semantics:
///   Ok(v)                => yielded one event row
///   Err("AGAIN")         => temporary no-data
///   Err("NO_MORE_DATA")  => terminal EOF (only explicit caps / shutdown)
pub trait LldpEngine: Send + Sync {
    fn next(&mut self) -> Result<Value, String>;
}

/// PluginIterator wrapper that drives an `LldpEngine`.
pub struct LldpIter {
    engine: Box<dyn LldpEngine + Send + Sync + 'static>,
}

impl LldpIter {
    pub fn new(engine: Box<dyn LldpEngine + Send + Sync + 'static>) -> Self {
        Self { engine }
    }
}

impl PluginIterator for LldpIter {
    #[inline]
    fn next_value(&mut self) -> Result<Value, String> {
        self.engine.next()
    }
}

/// Convert an `IteratorHandle` into a **zero-argument transform function**
/// that yields exactly one row per call (or returns "AGAIN"/"NO_MORE_DATA").
///
/// This is the shape Flow expects for sources: a 0-arg callable that produces
/// a single item on each invocation.
pub fn iter_to_transform(handle: IteratorHandle) -> Box<FunctionValue> {
    use mumu::parser::types::FunctionValue::RustClosure;

    Box::new(RustClosure(
        "net:lldp-transform".to_string(),
        Arc::new(Mutex::new(
            move |_interp: &mut mumu::parser::interpreter::Interpreter, _args: Vec<Value>| {
                match &handle.kind {
                    // We don't normally use Core here, but keep a complete implementation.
                    IteratorKind::Core(state_arc) => {
                        let mut s = state_arc
                            .lock()
                            .map_err(|_| "net:lldp-transform: IteratorState lock".to_string())?;
                        if s.done || s.current >= s.end {
                            s.done = true;
                            Err("NO_MORE_DATA".to_string())
                        } else {
                            let v = Value::Int(s.current);
                            s.current += 1;
                            if s.current >= s.end {
                                s.done = true;
                            }
                            Ok(v)
                        }
                    }
                    IteratorKind::Plugin(p) => {
                        let mut it = p
                            .lock()
                            .map_err(|_| "net:lldp-transform: Plugin Iterator lock".to_string())?;
                        it.next_value()
                    }
                }
            },
        )),
        0,
    ))
}

/// Helper exported to engine implementations to wrap an engine into
/// a MuMu `IteratorHandle`.
pub(super) fn handle_from_engine(
    engine: Box<dyn LldpEngine + Send + Sync + 'static>,
) -> IteratorHandle {
    let it = LldpIter::new(engine);
    IteratorHandle {
        kind: IteratorKind::Plugin(Arc::new(Mutex::new(it))),
    }
}