scirs2_core/reactive/
dataflow.rs

1//! Push-pull dataflow network.
2//!
3//! A dataflow network is a directed acyclic graph (DAG) of processing nodes.
4//! Data flows from [`Source`] nodes, through transformation nodes ([`Map`],
5//! [`Filter`], [`Zip`], [`Buffer`]), and is consumed by [`Sink`] nodes.
6//!
7//! ## Execution model
8//!
9//! The network is **lazy on the pull side** and **eager on the push side**:
10//!
11//! - Each node implements the [`DataflowNode<T>`] trait that exposes both
12//!   `push` (send a value downstream) and `pull` (request the next value from
13//!   upstream).
14//! - [`DataflowGraph`] manages a set of named nodes and their connections.
15//!   Calling [`DataflowGraph::run`] drains all `Source` nodes, pushing every
16//!   available value through the graph in topological order.
17//!
18//! ## Example
19//!
20//! ```rust
21//! use scirs2_core::reactive::dataflow::{DataflowGraph, Source, Map, Filter, Sink};
22//!
23//! let mut graph = DataflowGraph::new();
24//! let src = Source::from_iter(0..10i32);
25//! let map = Map::new(|x: i32| x * 2);
26//! let filter = Filter::new(|x: &i32| *x > 8);
27//! let sink: Sink<i32> = Sink::new();
28//!
29//! let src_id = graph.add_source(src);
30//! let map_id = graph.add_map(map);
31//! let flt_id = graph.add_filter(filter);
32//! let snk_id = graph.add_sink(sink);
33//!
34//! graph.connect(src_id, map_id);
35//! graph.connect(map_id, flt_id);
36//! graph.connect(flt_id, snk_id);
37//!
38//! graph.run();
39//!
40//! let results = graph.collect_sink(snk_id);
41//! assert_eq!(results, vec![10, 12, 14, 16, 18]);
42//! ```
43
44use std::sync::{Arc, Mutex};
45
46// ---------------------------------------------------------------------------
47// NodeId
48// ---------------------------------------------------------------------------
49
50/// Opaque node identifier within a [`DataflowGraph`].
51#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
52pub struct NodeId(usize);
53
54// ---------------------------------------------------------------------------
55// DataflowNode trait (type-erased)
56// ---------------------------------------------------------------------------
57
58/// Trait for nodes in the dataflow graph.
59///
60/// Both `push` and `pull` are provided so a node can act as either a consumer
61/// (pull from upstream, push downstream) or a pure transformer.
62pub trait DataflowNode<T>: Send + Sync {
63    /// Push a value into this node.  The node may buffer or forward it.
64    fn push(&self, value: T);
65
66    /// Pull the next available value from this node's output queue.
67    fn pull(&self) -> Option<T>;
68}
69
70// ---------------------------------------------------------------------------
71// Source<T>
72// ---------------------------------------------------------------------------
73
74struct SourceInner<T> {
75    buffer: std::collections::VecDeque<T>,
76}
77
78/// A source node that produces values from an iterator or via manual push.
79pub struct Source<T> {
80    inner: Arc<Mutex<SourceInner<T>>>,
81}
82
83impl<T: Clone + Send + Sync + 'static> Source<T> {
84    /// Create a source pre-loaded with values from an iterator.
85    #[allow(clippy::should_implement_trait)]
86    pub fn from_iter(iter: impl Iterator<Item = T>) -> Self {
87        let buffer: std::collections::VecDeque<T> = iter.collect();
88        Source {
89            inner: Arc::new(Mutex::new(SourceInner { buffer })),
90        }
91    }
92
93    /// Create an empty source (values added via `push_value`).
94    pub fn empty() -> Self {
95        Source {
96            inner: Arc::new(Mutex::new(SourceInner {
97                buffer: std::collections::VecDeque::new(),
98            })),
99        }
100    }
101
102    /// Manually push a value into the source buffer.
103    pub fn push_value(&self, value: T) {
104        if let Ok(mut g) = self.inner.lock() {
105            g.buffer.push_back(value);
106        }
107    }
108
109    /// Number of buffered values.
110    pub fn len(&self) -> usize {
111        self.inner.lock().map(|g| g.buffer.len()).unwrap_or(0)
112    }
113
114    /// `true` if the source buffer is empty.
115    pub fn is_empty(&self) -> bool {
116        self.len() == 0
117    }
118}
119
120impl<T: Clone + Send + Sync + 'static> DataflowNode<T> for Source<T> {
121    fn push(&self, value: T) {
122        if let Ok(mut g) = self.inner.lock() {
123            g.buffer.push_back(value);
124        }
125    }
126
127    fn pull(&self) -> Option<T> {
128        self.inner.lock().ok()?.buffer.pop_front()
129    }
130}
131
132// ---------------------------------------------------------------------------
133// Sink<T>
134// ---------------------------------------------------------------------------
135
136struct SinkInner<T> {
137    collected: Vec<T>,
138    callback: Option<Box<dyn Fn(T) + Send + Sync + 'static>>,
139}
140
141/// A sink node that consumes values and optionally applies a callback.
142pub struct Sink<T> {
143    inner: Arc<Mutex<SinkInner<T>>>,
144}
145
146impl<T: Clone + Send + Sync + 'static> Sink<T> {
147    /// Create a collecting sink (values are stored in an internal `Vec`).
148    pub fn new() -> Self {
149        Sink {
150            inner: Arc::new(Mutex::new(SinkInner {
151                collected: Vec::new(),
152                callback: None,
153            })),
154        }
155    }
156
157    /// Create a sink that applies `f` to every received value.
158    pub fn for_each(f: impl Fn(T) + Send + Sync + 'static) -> Self {
159        Sink {
160            inner: Arc::new(Mutex::new(SinkInner {
161                collected: Vec::new(),
162                callback: Some(Box::new(f)),
163            })),
164        }
165    }
166
167    /// Return the collected values, leaving the internal buffer empty.
168    pub fn drain(&self) -> Vec<T> {
169        self.inner
170            .lock()
171            .map(|mut g| std::mem::take(&mut g.collected))
172            .unwrap_or_default()
173    }
174
175    /// Return a clone of collected values without draining.
176    pub fn collect(&self) -> Vec<T>
177    where
178        T: Clone,
179    {
180        self.inner
181            .lock()
182            .map(|g| g.collected.clone())
183            .unwrap_or_default()
184    }
185}
186
187impl<T: Clone + Send + Sync + 'static> DataflowNode<T> for Sink<T> {
188    fn push(&self, value: T) {
189        if let Ok(mut g) = self.inner.lock() {
190            if let Some(cb) = &g.callback {
191                cb(value);
192            } else {
193                g.collected.push(value);
194            }
195        }
196    }
197
198    fn pull(&self) -> Option<T> {
199        // Sinks do not produce values.
200        None
201    }
202}
203
204// ---------------------------------------------------------------------------
205// Map<T, U>
206// ---------------------------------------------------------------------------
207
208/// A transformation node that applies a function `T -> U`.
209pub struct Map<T, U> {
210    func: Arc<dyn Fn(T) -> U + Send + Sync + 'static>,
211    output: Arc<Mutex<std::collections::VecDeque<U>>>,
212}
213
214impl<T: Send + Sync + 'static, U: Send + Sync + 'static> Map<T, U> {
215    /// Create a `Map` node with the given transformation function.
216    pub fn new(f: impl Fn(T) -> U + Send + Sync + 'static) -> Self {
217        Map {
218            func: Arc::new(f),
219            output: Arc::new(Mutex::new(std::collections::VecDeque::new())),
220        }
221    }
222}
223
224impl<T: Send + Sync + 'static, U: Send + Sync + 'static> DataflowNode<T> for Map<T, U> {
225    fn push(&self, value: T) {
226        let out = (self.func)(value);
227        if let Ok(mut q) = self.output.lock() {
228            q.push_back(out);
229        }
230    }
231
232    fn pull(&self) -> Option<T> {
233        // Map produces U, not T; this method is not meaningful here.
234        // It is only called on Map<T,T> or via the graph which uses
235        // the type-erased output queue directly.
236        None
237    }
238}
239
240impl<T: Send + Sync + 'static, U: Send + Sync + 'static> Map<T, U> {
241    /// Pull from the output queue (produces `U`).
242    pub fn pull_out(&self) -> Option<U> {
243        self.output.lock().ok()?.pop_front()
244    }
245}
246
247// ---------------------------------------------------------------------------
248// Filter<T>
249// ---------------------------------------------------------------------------
250
251/// A filter node that forwards values satisfying a predicate.
252pub struct Filter<T> {
253    pred: Arc<dyn Fn(&T) -> bool + Send + Sync + 'static>,
254    output: Arc<Mutex<std::collections::VecDeque<T>>>,
255}
256
257impl<T: Send + Sync + 'static> Filter<T> {
258    /// Create a `Filter` node with the given predicate.
259    pub fn new(pred: impl Fn(&T) -> bool + Send + Sync + 'static) -> Self {
260        Filter {
261            pred: Arc::new(pred),
262            output: Arc::new(Mutex::new(std::collections::VecDeque::new())),
263        }
264    }
265}
266
267impl<T: Send + Sync + 'static> DataflowNode<T> for Filter<T> {
268    fn push(&self, value: T) {
269        if (self.pred)(&value) {
270            if let Ok(mut q) = self.output.lock() {
271                q.push_back(value);
272            }
273        }
274    }
275
276    fn pull(&self) -> Option<T> {
277        self.output.lock().ok()?.pop_front()
278    }
279}
280
281// ---------------------------------------------------------------------------
282// Zip<T, U>
283// ---------------------------------------------------------------------------
284
285/// A zip node that combines values from two input queues into pairs.
286///
287/// A pair `(T, U)` is emitted only when both queues have at least one element.
288pub struct Zip<T, U> {
289    left: Arc<Mutex<std::collections::VecDeque<T>>>,
290    right: Arc<Mutex<std::collections::VecDeque<U>>>,
291    output: Arc<Mutex<std::collections::VecDeque<(T, U)>>>,
292}
293
294impl<T: Send + Sync + 'static, U: Send + Sync + 'static> Zip<T, U> {
295    /// Create a new `Zip` node.
296    pub fn new() -> Self {
297        Zip {
298            left: Arc::new(Mutex::new(std::collections::VecDeque::new())),
299            right: Arc::new(Mutex::new(std::collections::VecDeque::new())),
300            output: Arc::new(Mutex::new(std::collections::VecDeque::new())),
301        }
302    }
303
304    /// Push a value for the **left** stream.
305    pub fn push_left(&self, value: T) {
306        if let Ok(mut l) = self.left.lock() {
307            l.push_back(value);
308        }
309        self.try_pair();
310    }
311
312    /// Push a value for the **right** stream.
313    pub fn push_right(&self, value: U) {
314        if let Ok(mut r) = self.right.lock() {
315            r.push_back(value);
316        }
317        self.try_pair();
318    }
319
320    fn try_pair(&self) {
321        loop {
322            let pair = {
323                let mut l = match self.left.lock() {
324                    Ok(g) => g,
325                    Err(_) => break,
326                };
327                let mut r = match self.right.lock() {
328                    Ok(g) => g,
329                    Err(_) => break,
330                };
331                match (l.pop_front(), r.pop_front()) {
332                    (Some(lv), Some(rv)) => (lv, rv),
333                    (Some(lv), None) => {
334                        l.push_front(lv);
335                        break;
336                    }
337                    (None, Some(rv)) => {
338                        r.push_front(rv);
339                        break;
340                    }
341                    (None, None) => break,
342                }
343            };
344            if let Ok(mut out) = self.output.lock() {
345                out.push_back(pair);
346            }
347        }
348    }
349
350    /// Pull the next `(T, U)` pair.
351    pub fn pull_pair(&self) -> Option<(T, U)> {
352        self.output.lock().ok()?.pop_front()
353    }
354}
355
356// ---------------------------------------------------------------------------
357// Buffer<T>
358// ---------------------------------------------------------------------------
359
360/// A buffering node that accumulates `batch_size` items before releasing them
361/// as a batch (`Vec<T>`).
362pub struct Buffer<T> {
363    batch_size: usize,
364    input: Arc<Mutex<std::collections::VecDeque<T>>>,
365    output: Arc<Mutex<std::collections::VecDeque<Vec<T>>>>,
366}
367
368impl<T: Send + Sync + 'static> Buffer<T> {
369    /// Create a `Buffer` that emits batches of `batch_size` items.
370    pub fn new(batch_size: usize) -> Self {
371        Buffer {
372            batch_size: batch_size.max(1),
373            input: Arc::new(Mutex::new(std::collections::VecDeque::new())),
374            output: Arc::new(Mutex::new(std::collections::VecDeque::new())),
375        }
376    }
377
378    fn flush_if_ready(&self) {
379        loop {
380            let batch: Option<Vec<T>> = {
381                let mut inp = match self.input.lock() {
382                    Ok(g) => g,
383                    Err(_) => break,
384                };
385                if inp.len() >= self.batch_size {
386                    Some(inp.drain(..self.batch_size).collect())
387                } else {
388                    None
389                }
390            };
391            match batch {
392                Some(b) => {
393                    if let Ok(mut out) = self.output.lock() {
394                        out.push_back(b);
395                    }
396                }
397                None => break,
398            }
399        }
400    }
401
402    /// Pull the next complete batch.
403    pub fn pull_batch(&self) -> Option<Vec<T>> {
404        self.output.lock().ok()?.pop_front()
405    }
406
407    /// Number of complete batches available.
408    pub fn batch_count(&self) -> usize {
409        self.output.lock().map(|g| g.len()).unwrap_or(0)
410    }
411}
412
413impl<T: Send + Sync + 'static> DataflowNode<T> for Buffer<T> {
414    fn push(&self, value: T) {
415        if let Ok(mut inp) = self.input.lock() {
416            inp.push_back(value);
417        }
418        self.flush_if_ready();
419    }
420
421    fn pull(&self) -> Option<T> {
422        // Buffer exposes batches, not individual items.
423        None
424    }
425}
426
427// ---------------------------------------------------------------------------
428// DataflowGraph
429// ---------------------------------------------------------------------------
430
431/// Type-erased node wrapper stored inside the graph.
432enum AnyNode {
433    SourceI32(Arc<Source<i32>>),
434    SinkI32(Arc<Sink<i32>>),
435    MapI32(Arc<Map<i32, i32>>),
436    FilterI32(Arc<Filter<i32>>),
437    BufferI32(Arc<Buffer<i32>>),
438    SourceF64(Arc<Source<f64>>),
439    SinkF64(Arc<Sink<f64>>),
440}
441
442/// A typed node descriptor for building the graph with [`DataflowGraph`].
443///
444/// The graph stores nodes as typed variants internally; connections between
445/// nodes of compatible types are resolved at `run()` time.
446#[allow(missing_debug_implementations)]
447pub struct DataflowGraph {
448    nodes: Vec<AnyNode>,
449    edges: Vec<(NodeId, NodeId)>,
450}
451
452impl DataflowGraph {
453    /// Create a new, empty graph.
454    pub fn new() -> Self {
455        DataflowGraph {
456            nodes: Vec::new(),
457            edges: Vec::new(),
458        }
459    }
460
461    // --- i32 nodes ---------------------------------------------------------
462
463    /// Add an `i32` source node and return its [`NodeId`].
464    pub fn add_source(&mut self, src: Source<i32>) -> NodeId {
465        let id = NodeId(self.nodes.len());
466        self.nodes.push(AnyNode::SourceI32(Arc::new(src)));
467        id
468    }
469
470    /// Add an `i32 → i32` map node and return its [`NodeId`].
471    pub fn add_map(&mut self, map: Map<i32, i32>) -> NodeId {
472        let id = NodeId(self.nodes.len());
473        self.nodes.push(AnyNode::MapI32(Arc::new(map)));
474        id
475    }
476
477    /// Add an `i32` filter node and return its [`NodeId`].
478    pub fn add_filter(&mut self, filter: Filter<i32>) -> NodeId {
479        let id = NodeId(self.nodes.len());
480        self.nodes.push(AnyNode::FilterI32(Arc::new(filter)));
481        id
482    }
483
484    /// Add an `i32` sink node and return its [`NodeId`].
485    pub fn add_sink(&mut self, sink: Sink<i32>) -> NodeId {
486        let id = NodeId(self.nodes.len());
487        self.nodes.push(AnyNode::SinkI32(Arc::new(sink)));
488        id
489    }
490
491    /// Add an `i32` buffer node and return its [`NodeId`].
492    pub fn add_buffer(&mut self, buf: Buffer<i32>) -> NodeId {
493        let id = NodeId(self.nodes.len());
494        self.nodes.push(AnyNode::BufferI32(Arc::new(buf)));
495        id
496    }
497
498    // --- f64 nodes ---------------------------------------------------------
499
500    /// Add an `f64` source node and return its [`NodeId`].
501    pub fn add_source_f64(&mut self, src: Source<f64>) -> NodeId {
502        let id = NodeId(self.nodes.len());
503        self.nodes.push(AnyNode::SourceF64(Arc::new(src)));
504        id
505    }
506
507    /// Add an `f64` sink node and return its [`NodeId`].
508    pub fn add_sink_f64(&mut self, sink: Sink<f64>) -> NodeId {
509        let id = NodeId(self.nodes.len());
510        self.nodes.push(AnyNode::SinkF64(Arc::new(sink)));
511        id
512    }
513
514    // --- edges -------------------------------------------------------------
515
516    /// Connect the output of `src` to the input of `dst`.
517    pub fn connect(&mut self, src: NodeId, dst: NodeId) {
518        self.edges.push((src, dst));
519    }
520
521    // --- execution ---------------------------------------------------------
522
523    /// Drive all source nodes and push values through the graph until all
524    /// sources are empty.
525    pub fn run(&self) {
526        // Build an adjacency list: for each node, which nodes receive its output?
527        let mut adjacency: Vec<Vec<usize>> = vec![Vec::new(); self.nodes.len()];
528        for &(NodeId(src), NodeId(dst)) in &self.edges {
529            if src < adjacency.len() {
530                adjacency[src].push(dst);
531            }
532        }
533
534        // Pull from every source and propagate.
535        let mut changed = true;
536        while changed {
537            changed = false;
538            for (src_idx, node) in self.nodes.iter().enumerate() {
539                match node {
540                    AnyNode::SourceI32(src) => {
541                        while let Some(v) = src.pull() {
542                            changed = true;
543                            self.propagate_i32(v, &adjacency[src_idx]);
544                        }
545                    }
546                    AnyNode::MapI32(map) => {
547                        while let Some(v) = map.pull_out() {
548                            changed = true;
549                            self.propagate_i32(v, &adjacency[src_idx]);
550                        }
551                    }
552                    AnyNode::FilterI32(flt) => {
553                        while let Some(v) = flt.pull() {
554                            changed = true;
555                            self.propagate_i32(v, &adjacency[src_idx]);
556                        }
557                    }
558                    AnyNode::SourceF64(src) => {
559                        while let Some(v) = src.pull() {
560                            changed = true;
561                            self.propagate_f64(v, &adjacency[src_idx]);
562                        }
563                    }
564                    _ => {}
565                }
566            }
567        }
568    }
569
570    fn propagate_i32(&self, value: i32, dst_indices: &[usize]) {
571        for &dst in dst_indices {
572            match self.nodes.get(dst) {
573                Some(AnyNode::MapI32(map)) => map.push(value),
574                Some(AnyNode::FilterI32(flt)) => flt.push(value),
575                Some(AnyNode::SinkI32(sink)) => sink.push(value),
576                Some(AnyNode::BufferI32(buf)) => buf.push(value),
577                _ => {}
578            }
579        }
580    }
581
582    fn propagate_f64(&self, value: f64, dst_indices: &[usize]) {
583        for &dst in dst_indices {
584            if let Some(AnyNode::SinkF64(sink)) = self.nodes.get(dst) {
585                sink.push(value)
586            }
587        }
588    }
589
590    /// Collect and drain all values accumulated in the `i32` sink at `id`.
591    pub fn collect_sink(&self, id: NodeId) -> Vec<i32> {
592        match self.nodes.get(id.0) {
593            Some(AnyNode::SinkI32(sink)) => sink.drain(),
594            _ => Vec::new(),
595        }
596    }
597
598    /// Collect and drain all values accumulated in the `f64` sink at `id`.
599    pub fn collect_sink_f64(&self, id: NodeId) -> Vec<f64> {
600        match self.nodes.get(id.0) {
601            Some(AnyNode::SinkF64(sink)) => sink.drain(),
602            _ => Vec::new(),
603        }
604    }
605
606    /// Return available batches from a `Buffer` node at `id`.
607    pub fn collect_buffer(&self, id: NodeId) -> Vec<Vec<i32>> {
608        match self.nodes.get(id.0) {
609            Some(AnyNode::BufferI32(buf)) => {
610                let mut batches = Vec::new();
611                while let Some(b) = buf.pull_batch() {
612                    batches.push(b);
613                }
614                batches
615            }
616            _ => Vec::new(),
617        }
618    }
619}
620
621// ---------------------------------------------------------------------------
622// Tests
623// ---------------------------------------------------------------------------
624
625#[cfg(test)]
626mod tests {
627    use super::*;
628
629    #[test]
630    fn test_dataflow_map() {
631        let mut graph = DataflowGraph::new();
632        let src = Source::from_iter(0..5i32);
633        let map = Map::new(|x: i32| x * 3);
634        let sink: Sink<i32> = Sink::new();
635
636        let src_id = graph.add_source(src);
637        let map_id = graph.add_map(map);
638        let snk_id = graph.add_sink(sink);
639
640        graph.connect(src_id, map_id);
641        graph.connect(map_id, snk_id);
642        graph.run();
643
644        let res = graph.collect_sink(snk_id);
645        assert_eq!(res, vec![0, 3, 6, 9, 12]);
646    }
647
648    #[test]
649    fn test_dataflow_filter() {
650        let mut graph = DataflowGraph::new();
651        let src = Source::from_iter(0..10i32);
652        let flt = Filter::new(|x: &i32| x % 2 == 0);
653        let sink: Sink<i32> = Sink::new();
654
655        let src_id = graph.add_source(src);
656        let flt_id = graph.add_filter(flt);
657        let snk_id = graph.add_sink(sink);
658
659        graph.connect(src_id, flt_id);
660        graph.connect(flt_id, snk_id);
661        graph.run();
662
663        let res = graph.collect_sink(snk_id);
664        assert_eq!(res, vec![0, 2, 4, 6, 8]);
665    }
666
667    #[test]
668    fn test_dataflow_source_sink() {
669        let mut graph = DataflowGraph::new();
670        let src = Source::from_iter(1..=5i32);
671        let sink: Sink<i32> = Sink::new();
672
673        let src_id = graph.add_source(src);
674        let snk_id = graph.add_sink(sink);
675
676        graph.connect(src_id, snk_id);
677        graph.run();
678
679        let res = graph.collect_sink(snk_id);
680        assert_eq!(res, vec![1, 2, 3, 4, 5]);
681    }
682
683    #[test]
684    fn test_dataflow_buffer() {
685        let mut graph = DataflowGraph::new();
686        let src = Source::from_iter(0..9i32);
687        let buf = Buffer::new(3);
688        let src_id = graph.add_source(src);
689        let snk_buf_id = graph.add_buffer(buf);
690
691        graph.connect(src_id, snk_buf_id);
692        graph.run();
693
694        let batches = graph.collect_buffer(snk_buf_id);
695        assert_eq!(batches.len(), 3);
696        assert_eq!(batches[0], vec![0, 1, 2]);
697        assert_eq!(batches[1], vec![3, 4, 5]);
698        assert_eq!(batches[2], vec![6, 7, 8]);
699    }
700
701    #[test]
702    fn test_dataflow_zip() {
703        let zip: Zip<i32, i32> = Zip::new();
704        zip.push_left(1);
705        zip.push_left(2);
706        zip.push_right(10);
707        zip.push_right(20);
708        zip.push_left(3);
709        zip.push_right(30);
710
711        let mut pairs = Vec::new();
712        while let Some(p) = zip.pull_pair() {
713            pairs.push(p);
714        }
715        assert_eq!(pairs, vec![(1, 10), (2, 20), (3, 30)]);
716    }
717
718    #[test]
719    fn test_source_manual_push() {
720        let src: Source<i32> = Source::empty();
721        src.push_value(5);
722        src.push_value(6);
723        assert_eq!(src.pull(), Some(5));
724        assert_eq!(src.pull(), Some(6));
725        assert_eq!(src.pull(), None);
726    }
727
728    #[test]
729    fn test_sink_collect() {
730        let sink: Sink<i32> = Sink::new();
731        sink.push(1);
732        sink.push(2);
733        sink.push(3);
734        assert_eq!(sink.collect(), vec![1, 2, 3]);
735    }
736
737    #[test]
738    fn test_dataflow_map_filter_pipeline() {
739        let mut graph = DataflowGraph::new();
740        let src = Source::from_iter(0..10i32);
741        let map = Map::new(|x: i32| x * 2);
742        let filter = Filter::new(|x: &i32| *x > 8);
743        let sink: Sink<i32> = Sink::new();
744
745        let src_id = graph.add_source(src);
746        let map_id = graph.add_map(map);
747        let flt_id = graph.add_filter(filter);
748        let snk_id = graph.add_sink(sink);
749
750        graph.connect(src_id, map_id);
751        graph.connect(map_id, flt_id);
752        graph.connect(flt_id, snk_id);
753        graph.run();
754
755        let res = graph.collect_sink(snk_id);
756        assert_eq!(res, vec![10, 12, 14, 16, 18]);
757    }
758}
scirs2_core/reactive/dataflow.rs

scirs2_core/reactive/
dataflow.rs
