graphrefly/
combinators.rs

1//! Static-dep combinators (D6/D24/B53).
2//!
3//! These are Rust-native graph-visible operator factories. They use declared deps
4//! plus ctx state only; no internal subscribe islands (D45). Heterogeneous TS
5//! tuple helpers are represented here as homogeneous `Vec<T>` or fixed tuple
6//! helpers where Rust can type them cleanly.
7
8use std::collections::VecDeque;
9
10use crate::ctx::{Ctx, DepTerminal, WaveData};
11use crate::node::NodeOpts;
12use crate::operators::Operator;
13use crate::protocol::Message;
14
15/// combine: emit latest values from all deps once every dep has delivered.
16pub fn combine<T: Clone + 'static>() -> Operator<Vec<T>> {
17    Operator::with_opts(
18        "combine",
19        NodeOpts {
20            partial: true,
21            ..NodeOpts::default()
22        },
23        move |ctx| {
24            let mut values = Vec::with_capacity(ctx.dep_len());
25            for i in 0..ctx.dep_len() {
26                let Some(value) = ctx.data::<T>(i) else {
27                    return;
28                };
29                values.push((*value).clone());
30            }
31            ctx.emit(values);
32        },
33    )
34}
35
36/// combine_latest: alias-shaped Rust helper for [`combine`].
37pub fn combine_latest<T: Clone + 'static>() -> Operator<Vec<T>> {
38    combine()
39}
40
41/// with_latest_from: dep 0 drives `(primary, latest_secondary)` emissions.
42pub fn with_latest_from<A: Clone + 'static, B: Clone + 'static>() -> Operator<(A, B)> {
43    Operator::with_opts(
44        "withLatestFrom",
45        NodeOpts {
46            complete_when_deps_complete: false,
47            terminal_as_real_input: true,
48            ..NodeOpts::default()
49        },
50        move |ctx| {
51            if is_complete(ctx.terminal(0)) {
52                ctx.down(vec![Message::Complete]);
53                return;
54            }
55            let Some(secondary) = ctx.data::<B>(1) else {
56                return;
57            };
58            for primary in ctx.batch::<A>(0) {
59                ctx.emit(((*primary).clone(), (*secondary).clone()));
60            }
61        },
62    )
63}
64
65#[derive(Clone)]
66struct ZipState<T> {
67    queues: Vec<VecDeque<T>>,
68    complete: Vec<bool>,
69}
70
71/// zip: buffer per-dep queues and emit one vector when every dep has one value.
72pub fn zip<T: Clone + 'static>() -> Operator<Vec<T>> {
73    Operator::with_opts(
74        "zip",
75        NodeOpts {
76            partial: true,
77            complete_when_deps_complete: false,
78            terminal_as_real_input: true,
79            ..NodeOpts::default()
80        },
81        move |ctx| {
82            let n = ctx.dep_len();
83            if n == 0 {
84                ctx.down(vec![Message::Complete]);
85                return;
86            }
87            let mut st = ctx
88                .state_get::<ZipState<T>>()
89                .map(|v| (*v).clone())
90                .unwrap_or_else(|| ZipState {
91                    queues: vec![VecDeque::new(); n],
92                    complete: vec![false; n],
93                });
94            if st.queues.len() != n {
95                st.queues.resize_with(n, VecDeque::new);
96                st.complete.resize(n, false);
97            }
98
99            for i in 0..n {
100                for value in ctx.batch::<T>(i) {
101                    st.queues[i].push_back((*value).clone());
102                }
103                if is_complete(ctx.terminal(i)) {
104                    st.complete[i] = true;
105                }
106            }
107
108            while st.queues.iter().all(|q| !q.is_empty()) {
109                let tuple = st
110                    .queues
111                    .iter_mut()
112                    .map(|q| q.pop_front().expect("zip queue is non-empty"))
113                    .collect::<Vec<_>>();
114                ctx.emit(tuple);
115            }
116
117            let should_complete = st
118                .complete
119                .iter()
120                .enumerate()
121                .any(|(i, done)| *done && st.queues[i].is_empty());
122            ctx.state_set(st);
123            if should_complete {
124                ctx.down(vec![Message::Complete]);
125            }
126        },
127    )
128}
129
130#[derive(Clone)]
131struct ConcatState<T> {
132    phase: u8,
133    pending: Vec<T>,
134    second_done: bool,
135}
136
137/// concat: forward dep 0, then dep 1; dep-1 early values are buffered.
138pub fn concat<T: Clone + 'static>() -> Operator<T> {
139    Operator::with_opts(
140        "concat",
141        NodeOpts {
142            partial: true,
143            complete_when_deps_complete: false,
144            terminal_as_real_input: true,
145            ..NodeOpts::default()
146        },
147        move |ctx| {
148            let mut st = ctx
149                .state_get::<ConcatState<T>>()
150                .map(|v| (*v).clone())
151                .unwrap_or(ConcatState {
152                    phase: 0,
153                    pending: Vec::new(),
154                    second_done: false,
155                });
156
157            if st.phase == 0 {
158                for value in ctx.batch::<T>(0) {
159                    ctx.emit((*value).clone());
160                }
161                for value in ctx.batch::<T>(1) {
162                    st.pending.push((*value).clone());
163                }
164                if is_complete(ctx.terminal(1)) {
165                    st.second_done = true;
166                }
167                if is_complete(ctx.terminal(0)) {
168                    st.phase = 1;
169                    for value in st.pending.drain(..) {
170                        ctx.emit(value);
171                    }
172                    if st.second_done {
173                        ctx.down(vec![Message::Complete]);
174                    }
175                }
176            } else {
177                for value in ctx.batch::<T>(1) {
178                    ctx.emit((*value).clone());
179                }
180                if is_complete(ctx.terminal(1)) {
181                    ctx.down(vec![Message::Complete]);
182                }
183            }
184            ctx.state_set(st);
185        },
186    )
187}
188
189#[derive(Clone)]
190struct RaceState {
191    winner: Option<usize>,
192    terminals: Vec<bool>,
193}
194
195/// race: first dep to deliver DATA wins; loser terminals are ignored.
196pub fn race<T: Clone + 'static>() -> Operator<T> {
197    Operator::with_opts(
198        "race",
199        NodeOpts {
200            partial: true,
201            error_when_deps_error: false,
202            complete_when_deps_complete: false,
203            terminal_as_real_input: true,
204            ..NodeOpts::default()
205        },
206        move |ctx| {
207            let n = ctx.dep_len();
208            let mut st = ctx
209                .state_get::<RaceState>()
210                .map(|v| (*v).clone())
211                .unwrap_or_else(|| RaceState {
212                    winner: None,
213                    terminals: vec![false; n],
214                });
215            st.terminals.resize(n, false);
216
217            for i in 0..n {
218                if ctx.terminal(i).is_some() {
219                    st.terminals[i] = true;
220                }
221            }
222
223            if let Some(winner) = st.winner {
224                for value in ctx.batch::<T>(winner) {
225                    ctx.emit((*value).clone());
226                }
227                match ctx.terminal(winner) {
228                    Some(DepTerminal::Complete) => {
229                        ctx.state_set(st);
230                        ctx.down(vec![Message::Complete]);
231                        return;
232                    }
233                    Some(DepTerminal::Error(error)) => {
234                        ctx.state_set(st);
235                        ctx.down(vec![Message::Error(error.to_string().into())]);
236                        return;
237                    }
238                    None => {}
239                }
240            } else {
241                for i in 0..n {
242                    let batch = ctx.batch::<T>(i);
243                    if !batch.is_empty() {
244                        st.winner = Some(i);
245                        for value in batch {
246                            ctx.emit((*value).clone());
247                        }
248                        break;
249                    }
250                }
251                if st.winner.is_none() && st.terminals.iter().all(|t| *t) {
252                    ctx.state_set(st);
253                    ctx.down(vec![Message::Complete]);
254                    return;
255                }
256            }
257
258            ctx.state_set(st);
259        },
260    )
261}
262
263/// buffer: collect dep 0 values, flush a Vec on each dep 1 notifier DATA.
264pub fn buffer<T: Clone + 'static>() -> Operator<Vec<T>> {
265    Operator::with_opts(
266        "buffer",
267        NodeOpts {
268            partial: true,
269            complete_when_deps_complete: false,
270            terminal_as_real_input: true,
271            ..NodeOpts::default()
272        },
273        move |ctx| {
274            let mut buf = ctx
275                .state_get::<Vec<T>>()
276                .map(|v| (*v).clone())
277                .unwrap_or_default();
278            for value in ctx.batch::<T>(0) {
279                buf.push((*value).clone());
280            }
281            if is_complete(ctx.terminal(0)) {
282                if !buf.is_empty() {
283                    ctx.emit(buf.clone());
284                }
285                ctx.state_set(Vec::<T>::new());
286                ctx.down(vec![Message::Complete]);
287                return;
288            }
289            if dep_has_data(ctx, 1) {
290                ctx.emit(std::mem::take(&mut buf));
291                ctx.state_set(Vec::<T>::new());
292            } else {
293                ctx.state_set(buf);
294            }
295        },
296    )
297}
298
299/// buffer_count: emit chunks of `count`, flushing the remainder on source COMPLETE.
300pub fn buffer_count<T: Clone + 'static>(count: usize) -> Operator<Vec<T>> {
301    assert!(count > 0, "buffer_count: count must be positive");
302    Operator::with_opts(
303        "bufferCount",
304        NodeOpts {
305            complete_when_deps_complete: false,
306            terminal_as_real_input: true,
307            ..NodeOpts::default()
308        },
309        move |ctx| {
310            let mut buf = ctx
311                .state_get::<Vec<T>>()
312                .map(|v| (*v).clone())
313                .unwrap_or_default();
314            for value in ctx.batch::<T>(0) {
315                buf.push((*value).clone());
316                if buf.len() >= count {
317                    ctx.emit(std::mem::take(&mut buf));
318                }
319            }
320            if is_complete(ctx.terminal(0)) {
321                if !buf.is_empty() {
322                    ctx.emit(buf.clone());
323                }
324                ctx.state_set(Vec::<T>::new());
325                ctx.down(vec![Message::Complete]);
326            } else {
327                ctx.state_set(buf);
328            }
329        },
330    )
331}
332
333#[derive(Clone)]
334struct SampleState<T> {
335    last: Option<T>,
336    source_done: bool,
337}
338
339/// sample: emit dep 0's latest value whenever dep 1 notifier delivers DATA.
340pub fn sample<T: Clone + 'static>() -> Operator<T> {
341    Operator::with_opts(
342        "sample",
343        NodeOpts {
344            partial: true,
345            error_when_deps_error: false,
346            complete_when_deps_complete: false,
347            terminal_as_real_input: true,
348            ..NodeOpts::default()
349        },
350        move |ctx| {
351            for dep in [0, 1] {
352                if let Some(DepTerminal::Error(error)) = ctx.terminal(dep) {
353                    ctx.down(vec![Message::Error(error.to_string().into())]);
354                    return;
355                }
356            }
357
358            let mut st = ctx
359                .state_get::<SampleState<T>>()
360                .map(|v| (*v).clone())
361                .unwrap_or(SampleState {
362                    last: None,
363                    source_done: false,
364                });
365
366            for value in ctx.batch::<T>(0) {
367                st.last = Some((*value).clone());
368            }
369            if is_complete(ctx.terminal(0)) {
370                st.source_done = true;
371                st.last = None;
372            }
373            if is_complete(ctx.terminal(1)) {
374                ctx.state_set(st);
375                ctx.down(vec![Message::Complete]);
376                return;
377            }
378            if dep_has_data(ctx, 1) && !st.source_done {
379                if let Some(value) = st.last.clone() {
380                    ctx.emit(value);
381                }
382            }
383            ctx.state_set(st);
384        },
385    )
386}
387
388/// take_until: forward source values until notifier DATA arrives, then COMPLETE.
389pub fn take_until<T: Clone + 'static>() -> Operator<T> {
390    Operator::with_opts(
391        "takeUntil",
392        NodeOpts {
393            partial: true,
394            complete_when_deps_complete: false,
395            terminal_as_real_input: true,
396            ..NodeOpts::default()
397        },
398        move |ctx: &Ctx| {
399            if dep_has_data(ctx, 1) {
400                ctx.down(vec![Message::Complete]);
401                return;
402            }
403            for value in ctx.batch::<T>(0) {
404                ctx.emit((*value).clone());
405            }
406            if is_complete(ctx.terminal(0)) {
407                ctx.down(vec![Message::Complete]);
408            }
409        },
410    )
411}
412
413fn is_complete(terminal: Option<&DepTerminal>) -> bool {
414    matches!(terminal, Some(DepTerminal::Complete))
415}
416
417fn dep_has_data(ctx: &Ctx, i: usize) -> bool {
418    ctx.wave_data()
419        .get(i)
420        .into_iter()
421        .flat_map(|waves| waves.iter())
422        .flat_map(|wave| wave.iter())
423        .any(|item| matches!(item, WaveData::Data(_)))
424}
graphrefly/combinators.rs

graphrefly/
combinators.rs
