hyphae 1.0.1

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

//! `tap` operator — pure side effect; chains fuse into one closure on root.

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

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

/// Pipeline node representing `source.tap(f)`. Does not allocate a cell.
pub struct TapPipeline<S, T, F> {
    source: S,
    f: Arc<F>,
    _t: PhantomData<fn(T)>,
}

impl<S, T, F> PipelineInstall<T> for TapPipeline<S, T, F>
where
    S: PipelineInstall<T> + Send + Sync + 'static,
    T: CellValue,
    F: Fn(&T) + Send + Sync + 'static,
{
    fn install(&self, callback: Arc<dyn Fn(&Signal<T>) + Send + Sync>) -> SubscriptionGuard {
        let f = Arc::clone(&self.f);
        let wrapped: Arc<dyn Fn(&Signal<T>) + Send + Sync> = Arc::new(move |signal: &Signal<T>| {
            if let Signal::Value(v) = signal {
                (f)(v.as_ref());
            }
            callback(signal);
        });
        self.source.install(wrapped)
    }
}

impl<S, T, F> PipelineSeed<T> for TapPipeline<S, T, F>
where
    S: PipelineSeed<T>,
    T: CellValue,
    F: Fn(&T) + Send + Sync + 'static,
{
    fn seed(&self) -> T {
        let v = self.source.seed();
        (self.f)(&v);
        v
    }
}

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

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

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

/// Extension trait for side-effecting observation.
#[allow(private_bounds)]
pub trait TapExt<T: CellValue, S: Seedness>: Pipeline<T, S> {
    /// Run `f(&value)` for side effects and forward the value untransformed.
    ///
    /// Returns a [`TapPipeline`] node. Materialize to observe.
    #[track_caller]
    fn tap<F>(self, f: F) -> TapPipeline<Self, T, F>
    where
        F: Fn(&T) + Send + Sync + 'static,
    {
        TapPipeline {
            source: self,
            f: Arc::new(f),
            _t: PhantomData,
        }
    }
}

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

#[cfg(test)]
mod tests {
    use std::sync::{
        Arc,
        atomic::{AtomicU64, Ordering},
    };

    use super::*;
    use crate::{Cell, Gettable, MaterializeDefinite, Mutable};

    #[test]
    fn test_tap_side_effect() {
        let source = Cell::new(0u64);
        let side_effect = Arc::new(AtomicU64::new(0));

        let se = side_effect.clone();
        let tapped = source
            .clone()
            .tap(move |v| {
                se.store(*v, Ordering::SeqCst);
            })
            .materialize();

        source.set(42);
        assert_eq!(side_effect.load(Ordering::SeqCst), 42);
        assert_eq!(tapped.get(), 42); // value passes through unchanged
    }
}