graphrefly/
higher_order.rs

1//! Higher-order operator factories (D6/D24/B53).
2//!
3//! These are graph-layer sugar over the existing R-rewire-deferred substrate:
4//! inners are runtime deps added/removed with `ctx.rewire_next_*`, not hidden
5//! subscriptions. Projectors must return graph-local nodes.
6
7use std::cell::RefCell;
8use std::collections::VecDeque;
9use std::panic::{catch_unwind, AssertUnwindSafe};
10use std::rc::Rc;
11
12use crate::ctx::{Ctx, DepTerminal, WaveData};
13use crate::node::{Core, Node, NodeOpts};
14use crate::operators::Operator;
15use crate::protocol::Message;
16
17type Body = Rc<dyn Fn(&Ctx)>;
18type BodyCell = Rc<RefCell<Option<Body>>>;
19type Project<TIn, TOut> = Rc<dyn Fn(&Ctx, &TIn) -> Node<TOut>>;
20
21#[derive(Clone, Copy)]
22enum Mode {
23    Merge,
24    Switch,
25    Concat,
26    Exhaust,
27}
28
29#[derive(Clone)]
30struct MapState<TIn> {
31    inners: Vec<Core>,
32    queue: VecDeque<TIn>,
33    source_done: bool,
34}
35
36/// Options for [`merge_map_with_options`].
37#[derive(Clone, Copy, Debug, Default)]
38pub struct MergeMapOptions {
39    /// Maximum number of live inner deps at once. `None` keeps the default unbounded merge_map.
40    pub concurrent: Option<usize>,
41}
42
43/// switch_map: project each source DATA to an inner node, cancelling the prior live inner.
44pub fn switch_map<TIn: Clone + 'static, TOut: 'static>(
45    project: impl Fn(&TIn) -> Node<TOut> + 'static,
46) -> Operator<TOut> {
47    map_operator(
48        "switchMap",
49        move |_ctx, value| project(value),
50        Mode::Switch,
51        None,
52    )
53}
54
55/// merge_map: project every source DATA to an inner node and merge all live inners.
56pub fn merge_map<TIn: Clone + 'static, TOut: 'static>(
57    project: impl Fn(&TIn) -> Node<TOut> + 'static,
58) -> Operator<TOut> {
59    map_operator(
60        "mergeMap",
61        move |_ctx, value| project(value),
62        Mode::Merge,
63        None,
64    )
65}
66
67/// merge_map_with_options: merge_map plus optional in-flight inner limiting.
68///
69/// `concurrent = Some(n)` means at most `n` live inner deps are subscribed at once; excess
70/// outer DATA values are queued and projected lazily when an inner COMPLETE frees a slot.
71#[must_use]
72pub fn merge_map_with_options<TIn: Clone + 'static, TOut: 'static>(
73    project: impl Fn(&TIn) -> Node<TOut> + 'static,
74    opts: MergeMapOptions,
75) -> Operator<TOut> {
76    if let Some(0) = opts.concurrent {
77        panic!("merge_map_with_options: concurrent must be positive");
78    }
79    map_operator(
80        "mergeMap",
81        move |_ctx, value| project(value),
82        Mode::Merge,
83        opts.concurrent,
84    )
85}
86
87/// flat_map: alias-shaped Rust helper for [`merge_map`].
88pub fn flat_map<TIn: Clone + 'static, TOut: 'static>(
89    project: impl Fn(&TIn) -> Node<TOut> + 'static,
90) -> Operator<TOut> {
91    map_operator(
92        "flatMap",
93        move |_ctx, value| project(value),
94        Mode::Merge,
95        None,
96    )
97}
98
99/// concat_map: queue source values and run one projected inner at a time.
100pub fn concat_map<TIn: Clone + 'static, TOut: 'static>(
101    project: impl Fn(&TIn) -> Node<TOut> + 'static,
102) -> Operator<TOut> {
103    map_operator(
104        "concatMap",
105        move |_ctx, value| project(value),
106        Mode::Concat,
107        None,
108    )
109}
110
111/// exhaust_map: project the first source DATA while no inner is live; ignore source DATA while busy.
112pub fn exhaust_map<TIn: Clone + 'static, TOut: 'static>(
113    project: impl Fn(&TIn) -> Node<TOut> + 'static,
114) -> Operator<TOut> {
115    map_operator(
116        "exhaustMap",
117        move |_ctx, value| project(value),
118        Mode::Exhaust,
119        None,
120    )
121}
122
123pub(crate) fn switch_map_with_ctx<TIn: Clone + 'static, TOut: 'static>(
124    factory: &'static str,
125    project: impl Fn(&Ctx, &TIn) -> Node<TOut> + 'static,
126) -> Operator<TOut> {
127    map_operator(factory, project, Mode::Switch, None)
128}
129
130pub(crate) fn merge_map_with_ctx<TIn: Clone + 'static, TOut: 'static>(
131    factory: &'static str,
132    project: impl Fn(&Ctx, &TIn) -> Node<TOut> + 'static,
133) -> Operator<TOut> {
134    map_operator(factory, project, Mode::Merge, None)
135}
136
137#[derive(Clone)]
138struct RepeatState {
139    started: bool,
140    round: usize,
141    inner: Option<Core>,
142}
143
144/// repeat: run a fresh source from `factory` `count` times in sequence.
145///
146/// The factory must return a fresh node for each round. Reusing the same node is
147/// not a repeat in the clean-slate substrate: same-boundary
148/// unsubscribe_dep+subscribe_dep is a no-op under D47. D115 keeps the model to
149/// ordinary unsubscribe plus a later subscribe, so same-node repeat needs a
150/// separate future design.
151pub fn repeat<T: 'static>(factory: impl Fn() -> Node<T> + 'static, count: usize) -> Operator<T> {
152    assert!(count > 0, "repeat: count must be positive");
153
154    let factory: Rc<dyn Fn() -> Node<T>> = Rc::new(factory);
155    let body_cell: BodyCell = Rc::new(RefCell::new(None));
156    let body_cell_for_body = body_cell.clone();
157    let body: Body = Rc::new(move |ctx| {
158        run_repeat_body(ctx, &factory, count, &body_cell_for_body);
159    });
160    *body_cell.borrow_mut() = Some(body.clone());
161
162    Operator::with_opts(
163        "repeat",
164        NodeOpts {
165            error_when_deps_error: false,
166            complete_when_deps_complete: false,
167            terminal_as_real_input: true,
168            ..NodeOpts::default()
169        },
170        move |ctx| body(ctx),
171    )
172}
173
174fn map_operator<TIn: Clone + 'static, TOut: 'static>(
175    factory: &'static str,
176    project: impl Fn(&Ctx, &TIn) -> Node<TOut> + 'static,
177    mode: Mode,
178    concurrent: Option<usize>,
179) -> Operator<TOut> {
180    let project: Project<TIn, TOut> = Rc::new(project);
181    let body_cell: BodyCell = Rc::new(RefCell::new(None));
182    let body_cell_for_body = body_cell.clone();
183    let body: Body = Rc::new(move |ctx| {
184        run_map_body(ctx, &project, mode, concurrent, &body_cell_for_body);
185    });
186    *body_cell.borrow_mut() = Some(body.clone());
187    Operator::with_opts(
188        factory,
189        NodeOpts {
190            partial: true,
191            error_when_deps_error: false,
192            complete_when_deps_complete: false,
193            terminal_as_real_input: true,
194            ..NodeOpts::default()
195        },
196        move |ctx| body(ctx),
197    )
198}
199
200fn run_map_body<TIn: Clone + 'static, TOut: 'static>(
201    ctx: &Ctx,
202    project: &Project<TIn, TOut>,
203    mode: Mode,
204    concurrent: Option<usize>,
205    body_cell: &BodyCell,
206) {
207    let mut st = ctx
208        .state_get::<MapState<TIn>>()
209        .map(|v| (*v).clone())
210        .unwrap_or(MapState {
211            inners: Vec::new(),
212            queue: VecDeque::new(),
213            source_done: false,
214        });
215
216    for i in 1..ctx.dep_len() {
217        forward_data(ctx, i);
218    }
219
220    if let Some(error) = first_error_terminal(ctx) {
221        cleanup_all_inners(ctx, &mut st, body_cell);
222        ctx.state_set(st);
223        ctx.down(vec![Message::Error(error.into())]);
224        return;
225    }
226
227    if is_complete(ctx.terminal(0)) {
228        st.source_done = true;
229    }
230
231    let mut to_remove = Vec::new();
232    let mut survivors = Vec::new();
233    for (i, inner) in st.inners.iter().enumerate() {
234        if is_complete(ctx.terminal(i + 1)) {
235            push_unique(&mut to_remove, inner.clone());
236        } else {
237            survivors.push(inner.clone());
238        }
239    }
240    st.inners = survivors;
241
242    let mut to_add = Vec::new();
243    let source_batch = ctx.batch::<TIn>(0);
244    if !source_batch.is_empty() {
245        match mode {
246            Mode::Switch => {
247                let latest = source_batch
248                    .last()
249                    .expect("source_batch is not empty")
250                    .as_ref()
251                    .clone();
252                let Some(inner) = project_inner(ctx, project, &latest, &mut st, body_cell) else {
253                    return;
254                };
255                for live in &st.inners {
256                    if !live.ptr_eq(&inner) {
257                        push_unique(&mut to_remove, live.clone());
258                    }
259                }
260                if contains_core(&to_remove, &inner) {
261                    st.inners.clear();
262                } else {
263                    if !contains_core(&st.inners, &inner) {
264                        to_add.push(inner.clone());
265                    }
266                    st.inners = vec![inner];
267                }
268            }
269            Mode::Merge => {
270                if concurrent.is_some() {
271                    for value in source_batch {
272                        st.queue.push_back(value.as_ref().clone());
273                    }
274                } else {
275                    for value in source_batch {
276                        let Some(inner) =
277                            project_inner(ctx, project, value.as_ref(), &mut st, body_cell)
278                        else {
279                            return;
280                        };
281                        if !contains_core(&st.inners, &inner) && !contains_core(&to_remove, &inner)
282                        {
283                            st.inners.push(inner.clone());
284                            to_add.push(inner);
285                        }
286                    }
287                }
288            }
289            Mode::Concat => {
290                for value in source_batch {
291                    st.queue.push_back(value.as_ref().clone());
292                }
293            }
294            Mode::Exhaust => {
295                if st.inners.is_empty() {
296                    let first = source_batch
297                        .first()
298                        .expect("source_batch is not empty")
299                        .as_ref()
300                        .clone();
301                    let Some(inner) = project_inner(ctx, project, &first, &mut st, body_cell)
302                    else {
303                        return;
304                    };
305                    if !contains_core(&to_remove, &inner) {
306                        st.inners.push(inner.clone());
307                        to_add.push(inner);
308                    }
309                }
310            }
311        }
312    }
313
314    if matches!(mode, Mode::Merge) {
315        while !st.queue.is_empty() && concurrent.is_none_or(|max| st.inners.len() < max) {
316            let value = st.queue.pop_front().expect("queue checked as non-empty");
317            let Some(inner) = project_inner(ctx, project, &value, &mut st, body_cell) else {
318                return;
319            };
320            if !contains_core(&st.inners, &inner) && !contains_core(&to_remove, &inner) {
321                st.inners.push(inner.clone());
322                to_add.push(inner);
323            }
324        }
325    }
326
327    if matches!(mode, Mode::Concat) && st.inners.is_empty() && to_add.is_empty() {
328        if let Some(value) = st.queue.pop_front() {
329            let Some(inner) = project_inner(ctx, project, &value, &mut st, body_cell) else {
330                return;
331            };
332            if !contains_core(&to_remove, &inner) {
333                st.inners.push(inner.clone());
334                to_add.push(inner);
335            }
336        }
337    }
338
339    ctx.state_set(st.clone());
340    for dep in to_remove {
341        ctx.rewire_next_unsubscribe_dep(dep, rewire_body(body_cell));
342    }
343    for dep in to_add {
344        ctx.rewire_next_subscribe_dep(dep, rewire_body(body_cell));
345    }
346
347    if st.source_done && st.inners.is_empty() && st.queue.is_empty() {
348        ctx.down(vec![Message::Complete]);
349    }
350}
351
352fn run_repeat_body<T: 'static>(
353    ctx: &Ctx,
354    factory: &Rc<dyn Fn() -> Node<T>>,
355    count: usize,
356    body_cell: &BodyCell,
357) {
358    let mut st = ctx
359        .state_get::<RepeatState>()
360        .map(|v| (*v).clone())
361        .unwrap_or(RepeatState {
362            started: false,
363            round: 0,
364            inner: None,
365        });
366
367    forward_data(ctx, 0);
368
369    if let Some(error) = first_error_terminal(ctx) {
370        if let Some(inner) = st.inner.take() {
371            ctx.rewire_next_unsubscribe_dep(inner, rewire_body(body_cell));
372        }
373        ctx.state_set(st);
374        ctx.down(vec![Message::Error(error.into())]);
375        return;
376    }
377
378    if !st.started {
379        let Some(inner) = make_repeat_inner(ctx, factory) else {
380            return;
381        };
382        st.started = true;
383        st.round = 0;
384        st.inner = Some(inner.clone());
385        ctx.state_set(st);
386        ctx.rewire_next_subscribe_dep(inner, rewire_body(body_cell));
387        return;
388    }
389
390    if st.inner.is_some() && is_complete(ctx.terminal(0)) {
391        let old = st.inner.take().expect("repeat inner was checked as Some");
392        ctx.rewire_next_unsubscribe_dep(old, rewire_body(body_cell));
393        if st.round + 1 < count {
394            st.round += 1;
395            let Some(next) = make_repeat_inner(ctx, factory) else {
396                ctx.state_set(RepeatState {
397                    started: true,
398                    round: st.round,
399                    inner: None,
400                });
401                return;
402            };
403            st.inner = Some(next.clone());
404            ctx.state_set(st);
405            ctx.rewire_next_subscribe_dep(next, rewire_body(body_cell));
406        } else {
407            ctx.state_set(RepeatState {
408                started: true,
409                round: st.round,
410                inner: None,
411            });
412            ctx.down(vec![Message::Complete]);
413        }
414    }
415}
416
417fn make_repeat_inner<T: 'static>(ctx: &Ctx, factory: &Rc<dyn Fn() -> Node<T>>) -> Option<Core> {
418    match catch_unwind(AssertUnwindSafe(|| factory().erased())) {
419        Ok(core) => Some(core),
420        Err(payload) => {
421            ctx.down(vec![Message::Error(panic_payload(payload).into())]);
422            None
423        }
424    }
425}
426
427fn project_inner<TIn: Clone + 'static, TOut: 'static>(
428    ctx: &Ctx,
429    project: &Project<TIn, TOut>,
430    value: &TIn,
431    st: &mut MapState<TIn>,
432    body_cell: &BodyCell,
433) -> Option<Core> {
434    match catch_unwind(AssertUnwindSafe(|| project(ctx, value).erased())) {
435        Ok(core) => Some(core),
436        Err(payload) => {
437            cleanup_all_inners(ctx, st, body_cell);
438            ctx.state_set(st.clone());
439            ctx.down(vec![Message::Error(panic_payload(payload).into())]);
440            None
441        }
442    }
443}
444
445fn cleanup_all_inners<TIn: Clone + 'static>(
446    ctx: &Ctx,
447    st: &mut MapState<TIn>,
448    body_cell: &BodyCell,
449) {
450    let mut seen = Vec::new();
451    for inner in st.inners.drain(..) {
452        if !contains_core(&seen, &inner) {
453            seen.push(inner);
454        }
455    }
456    st.queue.clear();
457    st.source_done = true;
458    for inner in seen {
459        ctx.rewire_next_unsubscribe_dep(inner, rewire_body(body_cell));
460    }
461}
462
463fn rewire_body(body_cell: &BodyCell) -> impl Fn(&Ctx) + 'static {
464    let body_cell = body_cell.clone();
465    move |ctx| {
466        let body = body_cell
467            .borrow()
468            .as_ref()
469            .expect("higher-order body initialized")
470            .clone();
471        body(ctx);
472    }
473}
474
475fn forward_data(ctx: &Ctx, dep: usize) {
476    for wave in ctx
477        .wave_data()
478        .get(dep)
479        .into_iter()
480        .flat_map(|waves| waves.iter())
481    {
482        for item in wave.iter() {
483            if let WaveData::Data(value) = item {
484                ctx.down(vec![Message::Data(value.clone())]);
485            }
486        }
487    }
488}
489
490fn first_error_terminal(ctx: &Ctx) -> Option<String> {
491    for i in 0..ctx.dep_len() {
492        if let Some(DepTerminal::Error(error)) = ctx.terminal(i) {
493            return Some(error.to_string());
494        }
495    }
496    None
497}
498
499fn is_complete(terminal: Option<&DepTerminal>) -> bool {
500    matches!(terminal, Some(DepTerminal::Complete))
501}
502
503fn push_unique(out: &mut Vec<Core>, core: Core) {
504    if !contains_core(out, &core) {
505        out.push(core);
506    }
507}
508
509fn contains_core(haystack: &[Core], needle: &Core) -> bool {
510    haystack.iter().any(|core| core.ptr_eq(needle))
511}
512
513fn panic_payload(payload: Box<dyn std::any::Any + Send>) -> String {
514    if let Some(s) = payload.downcast_ref::<&str>() {
515        (*s).to_owned()
516    } else if let Some(s) = payload.downcast_ref::<String>() {
517        s.clone()
518    } else {
519        "higher-order projector panicked".to_owned()
520    }
521}
graphrefly/higher_order.rs

graphrefly/
higher_order.rs
