mumuflow/
lib.rs

1// src/lib.rs
2// flow/src/lib.rs
3
4use std::ffi::c_void;
5use std::mem::{align_of, size_of};
6
7use mumu::parser::interpreter::Interpreter;
8use mumu::parser::types::{FunctionValue, IteratorHandle, IteratorKind, Value};
9
10mod common;
11mod trans;
12mod compose;
13mod slice;
14mod to_array;
15mod filter;
16mod throttle;
17mod throttle_with; // NEW
18// New modules
19mod reduce;
20mod of;
21mod empty;
22mod concat;
23mod chain;
24mod ap;
25mod drain;
26mod read;         // keep read
27mod drain_async;  // async drain (micro-bursts)
28mod drain_then;   // async drain with required callback at EOF
29mod stop_when;    // NEW: stop pipeline when predicate becomes true
30
31pub use trans::register_flow_trans;
32pub use compose::register_flow_compose;
33pub use slice::register_flow_slice;
34pub use to_array::register_flow_to_array;
35pub use filter::register_flow_filter;
36pub use throttle::register_flow_throttle;
37pub use throttle_with::register_flow_throttle_with; // NEW
38
39// New exports
40pub use trans::register_flow_map_alias;
41pub use reduce::register_flow_reduce;
42pub use of::register_flow_of;
43pub use empty::register_flow_empty;
44pub use concat::register_flow_concat;
45pub use chain::register_flow_chain;
46pub use ap::{register_flow_ap_zip, register_flow_ap};
47pub use drain::register_flow_drain;
48pub use read::register_flow_read;
49pub use drain_async::register_flow_drain_async;
50pub use drain_then::register_flow_drain_then;
51pub use stop_when::register_flow_stop_when;
52
53fn log_abi_info(interp: &Interpreter) {
54    if !interp.is_verbose() {
55        return;
56    }
57
58    eprintln!("[flow][abi] sizeof(Value)        = {}", size_of::<Value>());
59    eprintln!("[flow][abi] alignof(Value)        = {}", align_of::<Value>());
60    eprintln!("[flow][abi] sizeof(FunctionValue) = {}", size_of::<FunctionValue>());
61    eprintln!("[flow][abi] alignof(FunctionValue) = {}", align_of::<FunctionValue>());
62    eprintln!("[flow][abi] sizeof(IteratorHandle) = {}", size_of::<IteratorHandle>());
63    eprintln!("[flow][abi] sizeof(IteratorKind)   = {}", size_of::<IteratorKind>());
64
65    let dbg_fn = Value::Function(Box::new(FunctionValue::Named("dbg".to_string())));
66    let dbg_it = Value::Iterator(IteratorHandle {
67        kind: IteratorKind::Core(std::sync::Arc::new(std::sync::Mutex::new(
68            mumu::parser::types::IteratorState { current: 0, end: 1, done: false },
69        ))),
70    });
71    eprintln!("[flow][abi] variant(Function) discriminant ok? {:?}", dbg_fn);
72    eprintln!("[flow][abi] variant(Iterator) discriminant ok?  {:?}", dbg_it);
73}
74
75#[export_name = "Cargo_lock"]
76pub unsafe extern "C" fn cargo_lock(interp_ptr: *mut c_void, extra_str: *const c_void) -> i32 {
77    if interp_ptr.is_null() {
78        return 1;
79    }
80    let interp = &mut *(interp_ptr as *mut Interpreter);
81
82    if interp.is_verbose() {
83        eprintln!("[flow:Cargo.lock] => bridging flow plugin...");
84        if !extra_str.is_null() {
85            use std::ffi::CStr;
86            let c_str = CStr::from_ptr(extra_str as *const i8);
87            eprintln!("[flow:Cargo.lock] => extra arg='{}'", c_str.to_string_lossy());
88        }
89    }
90
91    log_abi_info(interp);
92
93    // Core stages
94    register_flow_trans(interp);
95    register_flow_compose(interp);
96    register_flow_slice(interp);
97    register_flow_to_array(interp);
98    register_flow_filter(interp);
99    register_flow_throttle(interp);
100    register_flow_throttle_with(interp); // NEW
101
102    // Aliases / extra operators
103    register_flow_map_alias(interp);
104
105    // Functional set
106    register_flow_of(interp);
107    register_flow_empty(interp);
108    register_flow_concat(interp);
109    register_flow_reduce(interp);
110    register_flow_chain(interp);
111    register_flow_ap_zip(interp);
112    register_flow_ap(interp);
113
114    // Sinks & utilities
115    register_flow_drain_async(interp);
116    register_flow_drain_then(interp);
117    register_flow_drain(interp);
118    register_flow_read(interp);
119
120    // NEW control operator
121    register_flow_stop_when(interp);
122
123    if interp.is_verbose() {
124        eprintln!("[flow:Cargo.lock] => done bridging flow plugin");
125    }
126
127    0
128}
mumuflow/lib.rs

mumuflow/
lib.rs
