amethystate_core/primitives/
pipeline.rs

1use crate::primitives::signal::{Signal, SignalSubscription};
2use std::sync::{Arc, Mutex};
3use uuid::Uuid;
4
5pub trait Reactive<T>: Clone + Send + Sync + 'static
6where
7    T: Clone + Send + Sync + 'static,
8{
9    fn get(&self) -> T;
10
11    fn subscribe_with_source<F>(&self, callback: F) -> SignalSubscription
12    where
13        F: Fn(T, Option<Uuid>) + Send + Sync + 'static;
14
15    fn subscribe<F>(&self, callback: F) -> SignalSubscription
16    where
17        F: Fn(T) + Send + Sync + 'static,
18    {
19        self.subscribe_with_source(move |v, _src| callback(v))
20    }
21
22    fn keepalive(&self) -> Option<Arc<dyn Send + Sync>> {
23        None
24    }
25}
26
27pub trait IntoPipeline<T>
28where
29    T: Clone + Send + Sync + 'static,
30{
31    fn pipe(self) -> Pipeline<T>;
32}
33
34struct PipelineInner<T> {
35    signal: Arc<Signal<T>>,
36    _source_subs: Vec<SignalSubscription>,
37    _keepalive: Vec<Arc<dyn Send + Sync>>,
38}
39
40pub struct Pipeline<T> {
41    inner: Arc<PipelineInner<T>>,
42}
43
44impl<T> Clone for Pipeline<T> {
45    fn clone(&self) -> Self {
46        Self {
47            inner: Arc::clone(&self.inner),
48        }
49    }
50}
51
52impl<T> Pipeline<T>
53where
54    T: Clone + Send + Sync + 'static,
55{
56    pub fn from_signal(
57        signal: Arc<Signal<T>>,
58        source_subs: Vec<SignalSubscription>,
59        keepalive: Vec<Arc<dyn Send + Sync>>,
60    ) -> Self {
61        Self {
62            inner: Arc::new(PipelineInner {
63                signal,
64                _source_subs: source_subs,
65                _keepalive: keepalive,
66            }),
67        }
68    }
69
70    pub fn get(&self) -> T {
71        self.inner.signal.get()
72    }
73
74    pub fn subscribe_with_source<F>(&self, callback: F) -> SignalSubscription
75    where
76        F: Fn(T, Option<Uuid>) + Send + Sync + 'static,
77    {
78        self.inner
79            .signal
80            .subscribe_with_source(move |val, src| callback(val.clone(), src))
81    }
82
83    pub fn subscribe<F>(&self, callback: F) -> SignalSubscription
84    where
85        F: Fn(T) + Send + Sync + 'static,
86    {
87        self.subscribe_with_source(move |val, _src| callback(val))
88    }
89
90    pub fn map<U, F>(self, f: F) -> Pipeline<U>
91    where
92        U: Clone + Send + Sync + 'static,
93        F: Fn(T) -> U + Send + Sync + 'static,
94    {
95        let f = Arc::new(f);
96        let initial = f(self.get());
97        let signal = Arc::new(Signal::new(initial));
98        let target = Arc::clone(&signal);
99        let mapper = Arc::clone(&f);
100
101        let sub = self.subscribe_with_source(move |value, source| {
102            target.set(mapper(value), source);
103        });
104
105        Pipeline {
106            inner: Arc::new(PipelineInner {
107                signal,
108                _source_subs: vec![sub],
109                _keepalive: vec![self.inner],
110            }),
111        }
112    }
113
114    pub fn filter_map<U, F>(self, f: F) -> Pipeline<U>
115    where
116        U: Default + Clone + Send + Sync + 'static,
117        F: Fn(T) -> Option<U> + Send + Sync + 'static,
118    {
119        let f = Arc::new(f);
120        let initial = f(self.get()).unwrap_or_default();
121        let signal = Arc::new(Signal::new(initial));
122        let target = Arc::clone(&signal);
123        let mapper = Arc::clone(&f);
124
125        let sub = self.subscribe_with_source(move |value, source| {
126            if let Some(mapped) = mapper(value) {
127                target.set(mapped, source);
128            }
129        });
130
131        Pipeline {
132            inner: Arc::new(PipelineInner {
133                signal,
134                _source_subs: vec![sub],
135                _keepalive: vec![self.inner],
136            }),
137        }
138    }
139
140    pub fn inspect<F>(self, f: F) -> Pipeline<T>
141    where
142        F: Fn(&T) + Send + Sync + 'static,
143    {
144        let f = Arc::new(f);
145        let initial = self.get();
146        f(&initial);
147
148        let signal = Arc::new(Signal::new(initial));
149        let target = Arc::clone(&signal);
150        let inspector = Arc::clone(&f);
151
152        let sub = self.subscribe_with_source(move |value, source| {
153            inspector(&value);
154            target.set(value, source);
155        });
156
157        Pipeline {
158            inner: Arc::new(PipelineInner {
159                signal,
160                _source_subs: vec![sub],
161                _keepalive: vec![self.inner],
162            }),
163        }
164    }
165
166    pub fn dedupe(self) -> Pipeline<T>
167    where
168        T: PartialEq,
169    {
170        let initial = self.get();
171        let last = Arc::new(Mutex::new(initial.clone()));
172        let signal = Arc::new(Signal::new(initial));
173        let target = Arc::clone(&signal);
174        let last_seen = Arc::clone(&last);
175
176        let sub = self.subscribe_with_source(move |value, source| {
177            let mut last = last_seen.lock().unwrap();
178            if *last != value {
179                *last = value.clone();
180                target.set(value, source);
181            }
182        });
183
184        Pipeline {
185            inner: Arc::new(PipelineInner {
186                signal,
187                _source_subs: vec![sub],
188                _keepalive: vec![self.inner],
189            }),
190        }
191    }
192}
193
194impl<T> Reactive<T> for Pipeline<T>
195where
196    T: Clone + Send + Sync + 'static,
197{
198    fn get(&self) -> T {
199        self.get()
200    }
201
202    fn subscribe_with_source<F>(&self, callback: F) -> SignalSubscription
203    where
204        F: Fn(T, Option<Uuid>) + Send + Sync + 'static,
205    {
206        self.subscribe_with_source(callback)
207    }
208
209    fn keepalive(&self) -> Option<Arc<dyn Send + Sync>> {
210        Some(self.inner.clone())
211    }
212}
213
214#[derive(Default, Clone)]
215pub struct ReactiveScope {
216    subs: Vec<SignalSubscription>,
217}
218
219impl ReactiveScope {
220    pub fn new() -> Self {
221        Self::default()
222    }
223
224    pub fn watch(&mut self, sub: SignalSubscription) {
225        self.subs.push(sub);
226    }
227
228    pub fn watch_scope(&mut self, mut other: Self) {
229        self.subs.append(&mut other.subs);
230    }
231
232    pub fn clear(&mut self) {
233        self.subs.clear();
234    }
235
236    pub fn len(&self) -> usize {
237        self.subs.len()
238    }
239
240    pub fn is_empty(&self) -> bool {
241        self.subs.is_empty()
242    }
243}
244
245impl<R, T> IntoPipeline<T> for R
246where
247    R: Reactive<T>,
248    T: Clone + Send + Sync + 'static,
249{
250    fn pipe(self) -> Pipeline<T> {
251        let initial = self.get();
252        let signal = Arc::new(Signal::new(initial));
253        let target = Arc::clone(&signal);
254        let sub = self.subscribe_with_source(move |value, source| {
255            target.set(value, source);
256        });
257
258        Pipeline {
259            inner: Arc::new(PipelineInner {
260                signal,
261                _source_subs: vec![sub],
262                _keepalive: self.keepalive().into_iter().collect(),
263            }),
264        }
265    }
266}
267
268macro_rules! impl_tuple_pipeline {
269    ($(($source_ty:ident, $source:ident, $value_ty:ident, $value:ident)),+ $(,)?) => {
270        impl<$($source_ty, $value_ty),+> IntoPipeline<($($value_ty,)+)> for ($($source_ty,)+)
271        where
272            $(
273                $source_ty: Reactive<$value_ty>,
274                $value_ty: Clone + Send + Sync + 'static,
275            )+
276        {
277            #[allow(non_snake_case)]
278            fn pipe(self) -> Pipeline<($($value_ty,)+)> {
279                let ($($source,)+) = self;
280                let initial = ($($source.get(),)+);
281                let signal = Arc::new(Signal::new(initial));
282                let mut source_subs = Vec::new();
283                let mut keepalive = Vec::new();
284
285                let refresh: Arc<dyn Fn() -> ($($value_ty,)+) + Send + Sync> = {
286                    $(let $value = $source.clone();)+
287                    Arc::new(move || ($($value.get(),)+))
288                };
289
290                $(
291                    if let Some(inner) = $source.keepalive() {
292                        keepalive.push(inner);
293                    }
294
295                    let target = Arc::clone(&signal);
296                    let refresh_cb = Arc::clone(&refresh);
297                    source_subs.push($source.subscribe_with_source(move |_, src| {
298                        target.set(refresh_cb(), src);
299                    }));
300                )+
301
302                Pipeline::from_signal(signal, source_subs, keepalive)
303            }
304        }
305    };
306}
307
308impl_tuple_pipeline!((RA, ra, A, a), (RB, rb, B, b));
309impl_tuple_pipeline!((RA, ra, A, a), (RB, rb, B, b), (RC, rc, C, c));
310impl_tuple_pipeline!(
311    (RA, ra, A, a),
312    (RB, rb, B, b),
313    (RC, rc, C, c),
314    (RD, rd, D, d)
315);
amethystate_core/primitives/pipeline.rs

amethystate_core/primitives/
pipeline.rs
