hyphae 1.0.3

Reactive cells and runtime primitives for rship
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160

//! `scan(initial, f)` operator — fold over each emission, emitting the new accumulator.
//!
//! [`Definite`] when source is `Definite`. The seed of the materialized cell is
//! `f(initial, source.seed())` (the accumulator after one source emission), so
//! `scan(0, +).materialize().get()` on `Cell::new(1)` returns `1`.

use std::{
    marker::PhantomData,
    sync::{Arc, Mutex},
};

use super::CellValue;
use crate::{
    pipeline::{
        Definite, Empty, MaterializeDefinite, MaterializeEmpty, Pipeline, PipelineInstall,
        PipelineSeed, Seedness,
    },
    signal::Signal,
    subscription::SubscriptionGuard,
};

/// Pipeline node representing `source.scan(initial, f)`.
pub struct ScanPipeline<S, T, U, F, Sd = Definite> {
    source: S,
    initial: U,
    f: Arc<F>,
    _t: PhantomData<fn(T)>,
    _sd: PhantomData<fn(Sd)>,
}

impl<S, T, U, F, Sd> PipelineInstall<U> for ScanPipeline<S, T, U, F, Sd>
where
    S: PipelineInstall<T> + Send + Sync + 'static,
    Sd: Seedness,
    T: CellValue,
    U: CellValue,
    F: Fn(&U, &T) -> U + Send + Sync + 'static,
{
    fn install(&self, callback: Arc<dyn Fn(&Signal<U>) + Send + Sync>) -> SubscriptionGuard {
        let f = Arc::clone(&self.f);
        // Capture a fresh accumulator seeded with `initial`. The first
        // emission (the synchronous initial replay) advances it to first_acc,
        // which matches the cell's seed.
        let acc: Arc<Mutex<U>> = Arc::new(Mutex::new(self.initial.clone()));
        let wrapped: Arc<dyn Fn(&Signal<T>) + Send + Sync> =
            Arc::new(move |signal: &Signal<T>| match signal {
                Signal::Value(v) => {
                    let next = {
                        let mut guard = acc.lock().expect("scan poisoned");
                        let next = f(&*guard, v.as_ref());
                        *guard = next.clone();
                        next
                    };
                    callback(&Signal::value(next));
                }
                Signal::Complete => callback(&Signal::Complete),
                Signal::Error(e) => callback(&Signal::Error(e.clone())),
            });
        self.source.install(wrapped)
    }
}

impl<S, T, U, F> PipelineSeed<U> for ScanPipeline<S, T, U, F, Definite>
where
    S: PipelineSeed<T>,
    T: CellValue,
    U: CellValue,
    F: Fn(&U, &T) -> U + Send + Sync + 'static,
{
    fn seed(&self) -> U {
        (self.f)(&self.initial, &self.source.seed())
    }
}

#[allow(private_bounds)]
impl<S, T, U, F, Sd> Pipeline<U, Sd> for ScanPipeline<S, T, U, F, Sd>
where
    S: Pipeline<T, Sd>,
    Sd: Seedness,
    T: CellValue,
    U: CellValue,
    F: Fn(&U, &T) -> U + Send + Sync + 'static,
{
}

impl<S, T, U, F> MaterializeDefinite<U> for ScanPipeline<S, T, U, F, Definite>
where
    S: Pipeline<T, Definite> + PipelineSeed<T>,
    T: CellValue,
    U: CellValue,
    F: Fn(&U, &T) -> U + Send + Sync + 'static,
{
}

impl<S, T, U, F> MaterializeEmpty<U> for ScanPipeline<S, T, U, F, Empty>
where
    S: Pipeline<T, Empty>,
    T: CellValue,
    U: CellValue,
    F: Fn(&U, &T) -> U + Send + Sync + 'static,
{
}

#[allow(private_bounds)]
pub trait ScanExt<T: CellValue, S: Seedness>: Pipeline<T, S> {
    #[track_caller]
    fn scan<U, F>(self, initial: U, f: F) -> ScanPipeline<Self, T, U, F, S>
    where
        U: CellValue,
        F: Fn(&U, &T) -> U + Send + Sync + 'static,
    {
        ScanPipeline {
            source: self,
            initial,
            f: Arc::new(f),
            _t: PhantomData,
            _sd: PhantomData,
        }
    }
}

impl<T: CellValue, S: Seedness, P: Pipeline<T, S>> ScanExt<T, S> for P {}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{Cell, Gettable, MaterializeDefinite, Mutable};

    #[test]
    fn test_scan_accumulates() {
        let source = Cell::new(1u64);
        let sum = source.clone().scan(0u64, |acc, x| acc + x).materialize();

        // Initial: 0 + 1 = 1
        assert_eq!(sum.get(), 1);

        source.set(2);
        assert_eq!(sum.get(), 3); // 1 + 2

        source.set(3);
        assert_eq!(sum.get(), 6); // 3 + 3
    }

    #[test]
    fn test_scan_with_different_types() {
        let source = Cell::new(1);
        let collected = source
            .clone()
            .scan(String::new(), |acc, x| format!("{}{}", acc, x))
            .materialize();

        assert_eq!(collected.get(), "1");

        source.set(2);
        assert_eq!(collected.get(), "12");

        source.set(3);
        assert_eq!(collected.get(), "123");
    }
}