graphrefly/
time.rs

1//! Wall-clock time operator factories (D52/D111/B53).
2//!
3//! Operators stay graph-layer sugar over visible helper deps. Raw time work lives
4//! only in [`crate::sources::timer`]; helper timer nodes are created in the owning
5//! node's graph arena/dispatcher through a crate-private `Ctx` initializer.
6
7use std::cell::RefCell;
8use std::panic::{catch_unwind, AssertUnwindSafe};
9use std::rc::Rc;
10
11use crate::ctx::{Ctx, DepTerminal, WaveData};
12use crate::higher_order::{merge_map_with_ctx, switch_map_with_ctx};
13use crate::node::{Core, Node, NodeOpts};
14use crate::operators::Operator;
15use crate::protocol::Message;
16use crate::sources::{interval, timer};
17
18type Body = Rc<dyn Fn(&Ctx)>;
19type BodyCell = Rc<RefCell<Option<Body>>>;
20type AuditProject<S> = Rc<dyn Fn(&Ctx, &S) -> Core>;
21
22fn delayed_value<S: Clone + 'static>(ctx: &Ctx, value: S, ms: u64) -> Node<S> {
23    let tick = ctx.init_node_in_scope(timer(ms), vec![]);
24    ctx.init_node_in_scope(
25        Operator::with_opts(
26            "delayedValue",
27            crate::node::NodeOpts {
28                partial: true,
29                error_when_deps_error: false,
30                complete_when_deps_complete: false,
31                terminal_as_real_input: true,
32                ..crate::node::NodeOpts::default()
33            },
34            move |ctx| {
35                if let Some(DepTerminal::Error(error)) = ctx.terminal(0) {
36                    ctx.down(vec![Message::Error(error.to_string().into())]);
37                } else if dep_has_data(ctx, 0) || is_complete(ctx.terminal(0)) {
38                    ctx.down(vec![
39                        Message::Data(Rc::new(value.clone())),
40                        Message::Complete,
41                    ]);
42                }
43            },
44        ),
45        vec![tick.erased()],
46    )
47}
48
49/// delay: shift every DATA value by `ms`, preserving every occurrence.
50pub fn delay<S: Clone + 'static>(ms: u64) -> Operator<S> {
51    merge_map_with_ctx("delay", move |ctx, value: &S| {
52        delayed_value(ctx, value.clone(), ms)
53    })
54}
55
56/// debounce: emit the latest value after `ms` of quiet.
57pub fn debounce<S: Clone + 'static>(ms: u64) -> Operator<S> {
58    switch_map_with_ctx("debounce", move |ctx, value: &S| {
59        delayed_value(ctx, value.clone(), ms)
60    })
61}
62
63/// debounce_time: RxJS-shaped alias of [`debounce`] with its own factory name.
64pub fn debounce_time<S: Clone + 'static>(ms: u64) -> Operator<S> {
65    switch_map_with_ctx("debounceTime", move |ctx, value: &S| {
66        delayed_value(ctx, value.clone(), ms)
67    })
68}
69
70/// throttle: leading-edge throttle. Emit immediately, then ignore source DATA for `ms`.
71pub fn throttle<S: Clone + 'static>(ms: u64) -> Operator<S> {
72    throttle_with_factory("throttle", ms)
73}
74
75/// throttle_time: RxJS-shaped alias of [`throttle`] with its own factory name.
76pub fn throttle_time<S: Clone + 'static>(ms: u64) -> Operator<S> {
77    throttle_with_factory("throttleTime", ms)
78}
79
80#[derive(Clone)]
81struct AuditState<S> {
82    window_open: bool,
83    latest: Option<S>,
84    notifier: Option<Core>,
85    suppress_next_notifier: bool,
86}
87
88#[derive(Clone)]
89struct TimeoutState {
90    timer: Option<Core>,
91}
92
93#[derive(Clone)]
94struct BufferTimeState<S> {
95    buffer: Vec<S>,
96    interval: Option<Core>,
97}
98
99#[derive(Clone)]
100struct ThrottleState {
101    timer: Option<Core>,
102    source_done: bool,
103}
104
105fn throttle_with_factory<S: Clone + 'static>(factory: &'static str, ms: u64) -> Operator<S> {
106    let body_cell: BodyCell = Rc::new(RefCell::new(None));
107    let body_cell_for_body = body_cell.clone();
108    let body: Body = Rc::new(move |ctx| {
109        run_throttle_body::<S>(ctx, ms, &body_cell_for_body);
110    });
111    *body_cell.borrow_mut() = Some(body.clone());
112
113    Operator::with_opts(
114        factory,
115        crate::node::NodeOpts {
116            partial: true,
117            error_when_deps_error: false,
118            complete_when_deps_complete: false,
119            terminal_as_real_input: true,
120            ..crate::node::NodeOpts::default()
121        },
122        move |ctx| body(ctx),
123    )
124}
125
126fn run_throttle_body<S: Clone + 'static>(ctx: &Ctx, ms: u64, body_cell: &BodyCell) {
127    ctx.state_persist(true);
128    let mut st = ctx
129        .state_get::<ThrottleState>()
130        .map(|v| (*v).clone())
131        .unwrap_or(ThrottleState {
132            timer: None,
133            source_done: false,
134        });
135
136    let timer_dep = if st.source_done { 0 } else { 1 };
137    let mut to_remove = Vec::new();
138    let mut to_add = None;
139    let mut set_to_timer_only = false;
140    let mut complete = false;
141
142    if st.timer.is_some() && (dep_has_data(ctx, timer_dep) || is_complete(ctx.terminal(timer_dep)))
143    {
144        if let Some(timer) = st.timer.take() {
145            to_remove.push(timer);
146        }
147        if st.source_done {
148            complete = true;
149        }
150    }
151
152    if let Some(error) = first_error(ctx) {
153        if let Some(timer) = st.timer.take() {
154            to_remove.push(timer);
155        }
156        ctx.state_set(ThrottleState {
157            timer: None,
158            source_done: true,
159        });
160        for timer in to_remove {
161            ctx.rewire_next_unsubscribe_dep(timer, rewire_body(body_cell));
162        }
163        ctx.down(vec![Message::Error(error.into())]);
164        return;
165    }
166
167    if !st.source_done {
168        let source_batch = ctx.batch::<S>(0);
169        if st.timer.is_none() {
170            if let Some(value) = source_batch.first() {
171                ctx.down(vec![Message::Data(Rc::new((**value).clone()))]);
172                let timer = ctx.init_node_in_scope(timer(ms), vec![]).erased();
173                st.timer = Some(timer.clone());
174                to_add = Some(timer);
175            }
176        }
177
178        if is_complete(ctx.terminal(0)) {
179            st.source_done = true;
180            if st.timer.is_some() {
181                set_to_timer_only = true;
182            } else {
183                complete = true;
184            }
185        }
186    }
187
188    ctx.state_set(st.clone());
189    for timer in to_remove {
190        ctx.rewire_next_unsubscribe_dep(timer, rewire_body(body_cell));
191    }
192    if set_to_timer_only {
193        let timer = st
194            .timer
195            .clone()
196            .expect("source_done with live throttle timer");
197        ctx.rewire_next_replace_deps(vec![timer], rewire_body(body_cell));
198    } else if let Some(timer) = to_add {
199        ctx.rewire_next_subscribe_dep(timer, rewire_body(body_cell));
200    }
201    if complete {
202        ctx.down(vec![Message::Complete]);
203    }
204}
205
206/// audit: value-triggered trailing throttle. The selector returns the duration notifier node.
207pub fn audit<S: Clone + 'static, N: 'static>(
208    duration_selector: impl Fn(&S) -> Node<N> + 'static,
209) -> Operator<S> {
210    let project: AuditProject<S> = Rc::new(move |_ctx, value| duration_selector(value).erased());
211    audit_with_ctx("audit", project)
212}
213
214/// audit_time: `audit` specialized to a graph-scoped `timer(ms)` duration.
215pub fn audit_time<S: Clone + 'static>(ms: u64) -> Operator<S> {
216    let project: AuditProject<S> =
217        Rc::new(move |ctx, _value| ctx.init_node_in_scope(timer(ms), vec![]).erased());
218    audit_with_ctx("auditTime", project)
219}
220
221fn audit_with_ctx<S: Clone + 'static>(
222    factory: &'static str,
223    duration_selector: AuditProject<S>,
224) -> Operator<S> {
225    let body_cell: BodyCell = Rc::new(RefCell::new(None));
226    let body_cell_for_body = body_cell.clone();
227    let body: Body = Rc::new(move |ctx| {
228        run_audit_body(ctx, &duration_selector, &body_cell_for_body);
229    });
230    *body_cell.borrow_mut() = Some(body.clone());
231
232    Operator::with_opts(
233        factory,
234        crate::node::NodeOpts {
235            partial: true,
236            error_when_deps_error: false,
237            complete_when_deps_complete: false,
238            terminal_as_real_input: true,
239            ..crate::node::NodeOpts::default()
240        },
241        move |ctx| body(ctx),
242    )
243}
244
245fn run_audit_body<S: Clone + 'static>(
246    ctx: &Ctx,
247    duration_selector: &AuditProject<S>,
248    body_cell: &BodyCell,
249) {
250    let mut st = ctx
251        .state_get::<AuditState<S>>()
252        .map(|v| (*v).clone())
253        .unwrap_or(AuditState {
254            window_open: false,
255            latest: None,
256            notifier: None,
257            suppress_next_notifier: false,
258        });
259
260    let source_batch = ctx.batch::<S>(0);
261    let notifier_signaled = dep_has_data(ctx, 1) || is_complete(ctx.terminal(1));
262    let notifier_fired = st.window_open && notifier_signaled && !st.suppress_next_notifier;
263    let fired_notifier = notifier_fired.then(|| st.notifier.clone()).flatten();
264    if st.window_open && notifier_signaled && st.suppress_next_notifier {
265        st.suppress_next_notifier = false;
266    }
267    if notifier_fired {
268        if let Some(value) = st.latest.clone() {
269            ctx.down(vec![Message::Data(Rc::new(value))]);
270        }
271        let old = st.notifier.take();
272        st.window_open = false;
273        st.latest = None;
274        st.suppress_next_notifier = false;
275        if let Some(old) = old {
276            ctx.rewire_next_unsubscribe_dep(old, rewire_body(body_cell));
277        }
278    }
279
280    if let Some(error) = first_error(ctx) {
281        if let Some(old) = st.notifier.take() {
282            ctx.rewire_next_unsubscribe_dep(old, rewire_body(body_cell));
283        }
284        ctx.state_set(AuditState::<S> {
285            window_open: false,
286            latest: None,
287            notifier: None,
288            suppress_next_notifier: false,
289        });
290        ctx.down(vec![Message::Error(error.into())]);
291        return;
292    }
293
294    for value in &source_batch {
295        st.latest = Some((**value).clone());
296    }
297
298    if is_complete(ctx.terminal(0)) {
299        if let Some(value) = st.latest.clone() {
300            ctx.down(vec![Message::Data(Rc::new(value))]);
301        }
302        if let Some(old) = st.notifier.take() {
303            ctx.rewire_next_unsubscribe_dep(old, rewire_body(body_cell));
304        }
305        ctx.state_set(AuditState::<S> {
306            window_open: false,
307            latest: None,
308            notifier: None,
309            suppress_next_notifier: false,
310        });
311        ctx.down(vec![Message::Complete]);
312        return;
313    }
314
315    if !st.window_open && !source_batch.is_empty() {
316        if let Some(value) = st.latest.as_ref() {
317            match catch_unwind(AssertUnwindSafe(|| duration_selector(ctx, value))) {
318                Ok(notifier) => {
319                    st.window_open = true;
320                    st.suppress_next_notifier = fired_notifier
321                        .as_ref()
322                        .is_some_and(|old| old.ptr_eq(&notifier));
323                    st.notifier = Some(notifier.clone());
324                    ctx.rewire_next_subscribe_dep(notifier, rewire_body(body_cell));
325                }
326                Err(payload) => {
327                    ctx.state_set(AuditState::<S> {
328                        window_open: false,
329                        latest: None,
330                        notifier: None,
331                        suppress_next_notifier: false,
332                    });
333                    ctx.down(vec![Message::Error(panic_payload(payload).into())]);
334                    return;
335                }
336            }
337        }
338    }
339
340    ctx.state_set(st);
341}
342
343/// timeout: subscribe-armed idle watchdog. The helper is a free node constructor,
344/// not an operator factory or graph method (D114).
345pub fn timeout<S: Clone + 'static>(source: &Node<S>, ms: u64) -> Node<S> {
346    let source_core = source.erased();
347    let initial_timer = scoped_timer(&source_core, ms).erased();
348    let body_cell: BodyCell = Rc::new(RefCell::new(None));
349    let body_cell_for_body = body_cell.clone();
350    let initial_for_body = initial_timer.clone();
351    let source_for_body = source_core.clone();
352    let body: Body = Rc::new(move |ctx| {
353        run_timeout_body::<S>(
354            ctx,
355            ms,
356            &initial_for_body,
357            &source_for_body,
358            &body_cell_for_body,
359        );
360    });
361    *body_cell.borrow_mut() = Some(body.clone());
362
363    let node = crate::operators::init_node_in_arena_with_dispatcher(
364        Operator::with_opts("timeout", time_helper_opts(), move |ctx| {
365            body(ctx);
366        }),
367        &source_core.arena(),
368        source_core.dispatcher(),
369        vec![initial_timer, source_core.clone()],
370        NodeOpts::default(),
371    );
372    node.erased().set_environment(source_core.environment());
373    node
374}
375
376fn run_timeout_body<S: Clone + 'static>(
377    ctx: &Ctx,
378    ms: u64,
379    initial_timer: &Core,
380    source: &Core,
381    body_cell: &BodyCell,
382) {
383    let mut st = ctx
384        .state_get::<TimeoutState>()
385        .map(|v| (*v).clone())
386        .unwrap_or_else(|| TimeoutState {
387            timer: Some(initial_timer.clone()),
388        });
389
390    let source_batch = ctx.batch::<S>(1);
391    for value in &source_batch {
392        ctx.down(vec![Message::Data(Rc::new((**value).clone()))]);
393    }
394
395    if is_complete(ctx.terminal(1)) {
396        if let Some(timer) = st.timer.take() {
397            ctx.rewire_next_unsubscribe_dep(timer, rewire_body(body_cell));
398        }
399        ctx.state_set(TimeoutState { timer: None });
400        ctx.down(vec![Message::Complete]);
401        return;
402    }
403
404    if let Some(DepTerminal::Error(error)) = ctx.terminal(1) {
405        if let Some(timer) = st.timer.take() {
406            ctx.rewire_next_unsubscribe_dep(timer, rewire_body(body_cell));
407        }
408        ctx.state_set(TimeoutState { timer: None });
409        ctx.down(vec![Message::Error(error.to_string().into())]);
410        return;
411    }
412
413    if !source_batch.is_empty() {
414        let old = st.timer.take();
415        let next = ctx.init_node_in_scope(timer(ms), vec![]).erased();
416        st.timer = Some(next.clone());
417        ctx.state_set(st);
418        if old.is_some() {
419            ctx.rewire_next_replace_deps(vec![next, source.clone()], rewire_body(body_cell));
420        } else {
421            ctx.rewire_next_subscribe_dep(next, rewire_body(body_cell));
422        }
423        return;
424    }
425
426    if let Some(DepTerminal::Error(error)) = ctx.terminal(0) {
427        ctx.state_set(TimeoutState { timer: None });
428        ctx.down(vec![Message::Error(error.to_string().into())]);
429        return;
430    }
431
432    if dep_has_data(ctx, 0) || is_complete(ctx.terminal(0)) {
433        ctx.state_set(TimeoutState { timer: None });
434        ctx.down(vec![Message::Error(
435            format!("timeout: no value within {ms}ms").into(),
436        )]);
437        return;
438    }
439
440    ctx.state_set(st);
441}
442
443/// buffer_time: subscribe-armed interval buffer helper (D114).
444pub fn buffer_time<S: Clone + 'static>(source: &Node<S>, ms: u64) -> Node<Vec<S>> {
445    let source_core = source.erased();
446    let interval_node = scoped_interval(&source_core, ms).erased();
447    let body_cell: BodyCell = Rc::new(RefCell::new(None));
448    let body_cell_for_body = body_cell.clone();
449    let interval_for_body = interval_node.clone();
450    let body: Body = Rc::new(move |ctx| {
451        run_buffer_time_body::<S>(ctx, &interval_for_body, &body_cell_for_body);
452    });
453    *body_cell.borrow_mut() = Some(body.clone());
454
455    let node = crate::operators::init_node_in_arena_with_dispatcher(
456        Operator::with_opts("bufferTime", time_helper_opts(), move |ctx| {
457            body(ctx);
458        }),
459        &source_core.arena(),
460        source_core.dispatcher(),
461        vec![interval_node, source_core.clone()],
462        NodeOpts::default(),
463    );
464    node.erased().set_environment(source_core.environment());
465    node
466}
467
468fn run_buffer_time_body<S: Clone + 'static>(
469    ctx: &Ctx,
470    initial_interval: &Core,
471    body_cell: &BodyCell,
472) {
473    let mut st = ctx
474        .state_get::<BufferTimeState<S>>()
475        .map(|v| (*v).clone())
476        .unwrap_or_else(|| BufferTimeState {
477            buffer: Vec::new(),
478            interval: Some(initial_interval.clone()),
479        });
480
481    for value in ctx.batch::<S>(1) {
482        st.buffer.push((*value).clone());
483    }
484
485    if is_complete(ctx.terminal(1)) {
486        if !st.buffer.is_empty() {
487            ctx.down(vec![Message::Data(Rc::new(st.buffer.clone()))]);
488        }
489        if let Some(interval) = st.interval.take() {
490            ctx.rewire_next_unsubscribe_dep(interval, rewire_body(body_cell));
491        }
492        ctx.state_set(BufferTimeState::<S> {
493            buffer: Vec::new(),
494            interval: None,
495        });
496        ctx.down(vec![Message::Complete]);
497        return;
498    }
499
500    if let Some(DepTerminal::Error(error)) = ctx.terminal(1) {
501        if let Some(interval) = st.interval.take() {
502            ctx.rewire_next_unsubscribe_dep(interval, rewire_body(body_cell));
503        }
504        ctx.state_set(BufferTimeState::<S> {
505            buffer: Vec::new(),
506            interval: None,
507        });
508        ctx.down(vec![Message::Error(error.to_string().into())]);
509        return;
510    }
511
512    if let Some(DepTerminal::Error(error)) = ctx.terminal(0) {
513        ctx.state_set(BufferTimeState::<S> {
514            buffer: Vec::new(),
515            interval: None,
516        });
517        ctx.down(vec![Message::Error(error.to_string().into())]);
518        return;
519    }
520
521    if dep_has_data(ctx, 0) {
522        ctx.down(vec![Message::Data(Rc::new(st.buffer.clone()))]);
523        st.buffer.clear();
524    }
525
526    ctx.state_set(st);
527}
528
529fn scoped_timer(anchor: &Core, ms: u64) -> Node<u64> {
530    scoped_time_source(anchor, timer(ms))
531}
532
533fn scoped_interval(anchor: &Core, ms: u64) -> Node<u64> {
534    scoped_time_source(anchor, interval(ms))
535}
536
537fn scoped_time_source(anchor: &Core, op: Operator<u64>) -> Node<u64> {
538    let node = crate::operators::init_node_in_arena_with_dispatcher(
539        op,
540        &anchor.arena(),
541        anchor.dispatcher(),
542        vec![],
543        NodeOpts::default(),
544    );
545    node.erased().set_environment(anchor.environment());
546    node
547}
548
549fn time_helper_opts() -> NodeOpts {
550    NodeOpts {
551        partial: true,
552        complete_when_deps_complete: false,
553        error_when_deps_error: false,
554        terminal_as_real_input: true,
555        ..NodeOpts::default()
556    }
557}
558
559fn rewire_body(body_cell: &BodyCell) -> impl Fn(&Ctx) + 'static {
560    let body_cell = body_cell.clone();
561    move |ctx| {
562        let body = body_cell
563            .borrow()
564            .as_ref()
565            .expect("audit body initialized")
566            .clone();
567        body(ctx);
568    }
569}
570
571fn dep_has_data(ctx: &Ctx, dep: usize) -> bool {
572    ctx.wave_data().get(dep).is_some_and(|waves| {
573        waves
574            .iter()
575            .flatten()
576            .any(|v| matches!(v, WaveData::Data(_)))
577    })
578}
579
580fn is_complete(term: Option<&DepTerminal>) -> bool {
581    matches!(term, Some(DepTerminal::Complete))
582}
583
584fn first_error(ctx: &Ctx) -> Option<String> {
585    (0..ctx.dep_len()).find_map(|i| match ctx.terminal(i) {
586        Some(DepTerminal::Error(error)) => Some(error.to_string()),
587        _ => None,
588    })
589}
590
591fn panic_payload(payload: Box<dyn std::any::Any + Send>) -> String {
592    if let Some(s) = payload.downcast_ref::<&str>() {
593        (*s).to_owned()
594    } else if let Some(s) = payload.downcast_ref::<String>() {
595        s.clone()
596    } else {
597        "time operator panicked".to_owned()
598    }
599}
graphrefly/time.rs

graphrefly/
time.rs
