amethystate_core/primitives/
pipeline.rs

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

amethystate_core/primitives/
pipeline.rs
