hyphae/traits/operators/
scan.rs

1//! `scan(initial, f)` operator — fold over each emission, emitting the new accumulator.
2//!
3//! [`Definite`] when source is `Definite`. The seed of the materialized cell is
4//! `f(initial, source.seed())` (the accumulator after one source emission), so
5//! `scan(0, +).materialize().get()` on `Cell::new(1)` returns `1`.
6
7use parking_lot::Mutex;
8use std::{marker::PhantomData, sync::Arc};
9
10use super::CellValue;
11use crate::{
12    pipeline::{Definite, Pipeline, PipelineInstall, PipelineSeed, Seedness},
13    signal::Signal,
14    subscription::SubscriptionGuard,
15};
16
17/// Pipeline node representing `source.scan(initial, f)`.
18pub struct ScanPipeline<S, T, U, F, Sd = Definite> {
19    source: S,
20    initial: U,
21    f: Arc<F>,
22    _t: PhantomData<fn(T)>,
23    _sd: PhantomData<fn(Sd)>,
24}
25
26impl<S, T, U, F, Sd> PipelineInstall<U> for ScanPipeline<S, T, U, F, Sd>
27where
28    S: PipelineInstall<T> + Send + Sync + 'static,
29    Sd: Seedness,
30    T: CellValue,
31    U: CellValue,
32    F: Fn(&U, &T) -> U + Send + Sync + 'static,
33{
34    fn install(&self, callback: Arc<dyn Fn(&Signal<U>) + Send + Sync>) -> SubscriptionGuard {
35        let f = Arc::clone(&self.f);
36        // Capture a fresh accumulator seeded with `initial`. The first
37        // emission (the synchronous initial replay) advances it to first_acc,
38        // which matches the cell's seed.
39        let acc: Arc<Mutex<U>> = Arc::new(Mutex::new(self.initial.clone()));
40        let wrapped: Arc<dyn Fn(&Signal<T>) + Send + Sync> =
41            Arc::new(move |signal: &Signal<T>| match signal {
42                Signal::Value(v) => {
43                    let next = {
44                        let mut guard = acc.lock();
45                        let next = f(&*guard, v.as_ref());
46                        *guard = next.clone();
47                        next
48                    };
49                    callback(&Signal::value(next));
50                }
51                Signal::Complete => callback(&Signal::Complete),
52                Signal::Error(e) => callback(&Signal::Error(e.clone())),
53            });
54        self.source.install(wrapped)
55    }
56}
57
58impl<S, T, U, F, Sd> PipelineSeed<U> for ScanPipeline<S, T, U, F, Sd>
59where
60    S: Pipeline<T, crate::pipeline::Definite>,
61    Sd: Seedness,
62    T: CellValue,
63    U: CellValue,
64    F: Fn(&U, &T) -> U + Send + Sync + 'static,
65{
66    fn seed(&self) -> U {
67        (self.f)(&self.initial, &self.source.pipeline_seed())
68    }
69}
70
71#[allow(private_bounds)]
72impl<S, T, U, F> Pipeline<U, crate::pipeline::Definite>
73    for ScanPipeline<S, T, U, F, crate::pipeline::Definite>
74where
75    S: Pipeline<T, crate::pipeline::Definite>,
76    T: CellValue,
77    U: CellValue,
78    F: Fn(&U, &T) -> U + Send + Sync + 'static,
79{
80}
81
82impl<S, T, U, F> Pipeline<U, crate::pipeline::Empty>
83    for ScanPipeline<S, T, U, F, crate::pipeline::Empty>
84where
85    S: Pipeline<T, crate::pipeline::Empty>,
86    T: CellValue,
87    U: CellValue,
88    F: Fn(&U, &T) -> U + Send + Sync + 'static,
89{
90}
91
92#[allow(private_bounds)]
93pub trait ScanExt<T: CellValue, S: Seedness>: Pipeline<T, S> {
94    #[track_caller]
95    fn scan<U, F>(self, initial: U, f: F) -> impl crate::Materialize<U, S>
96    where
97        U: CellValue,
98        F: Fn(&U, &T) -> U + Send + Sync + 'static;
99}
100
101impl<T: CellValue, P: Pipeline<T, crate::pipeline::Definite>> ScanExt<T, crate::pipeline::Definite>
102    for P
103{
104    fn scan<U, F>(self, initial: U, f: F) -> impl crate::Materialize<U, crate::pipeline::Definite>
105    where
106        U: CellValue,
107        F: Fn(&U, &T) -> U + Send + Sync + 'static,
108    {
109        ScanPipeline {
110            source: self,
111            initial,
112            f: Arc::new(f),
113            _t: PhantomData,
114            _sd: PhantomData,
115        }
116    }
117}
118
119impl<T: CellValue, P: Pipeline<T, crate::pipeline::Empty>> ScanExt<T, crate::pipeline::Empty>
120    for P
121{
122    fn scan<U, F>(self, initial: U, f: F) -> impl crate::Materialize<U, crate::pipeline::Empty>
123    where
124        U: CellValue,
125        F: Fn(&U, &T) -> U + Send + Sync + 'static,
126    {
127        ScanPipeline {
128            source: self,
129            initial,
130            f: Arc::new(f),
131            _t: PhantomData,
132            _sd: PhantomData,
133        }
134    }
135}
136
137#[cfg(test)]
138mod tests {
139    use super::*;
140    use crate::{Cell, Gettable, Materialize, Mutable};
141
142    #[test]
143    fn test_scan_accumulates() {
144        let source = Cell::new(1u64);
145        let sum = source.clone().scan(0u64, |acc, x| acc + x).materialize();
146
147        // Initial: 0 + 1 = 1
148        assert_eq!(sum.get(), 1);
149
150        source.set(2);
151        assert_eq!(sum.get(), 3); // 1 + 2
152
153        source.set(3);
154        assert_eq!(sum.get(), 6); // 3 + 3
155    }
156
157    #[test]
158    fn test_scan_with_different_types() {
159        let source = Cell::new(1);
160        let collected = source
161            .clone()
162            .scan(String::new(), |acc, x| format!("{acc}{x}"))
163            .materialize();
164
165        assert_eq!(collected.get(), "1");
166
167        source.set(2);
168        assert_eq!(collected.get(), "12");
169
170        source.set(3);
171        assert_eq!(collected.get(), "123");
172    }
173}
hyphae/traits/operators/scan.rs

hyphae/traits/operators/
scan.rs
