hyphae/traits/operators/
cold.rs

1//! `cold` operator — drop the synchronous-on-subscribe initial emission.
2//!
3//! `cold()` produces a pipeline whose output is `Arc<T>` (cheap forwarding) and
4//! whose materialized cell type is `Cell<Option<Arc<T>>>`, initialized to
5//! `None`. The first source emission (the synchronous replay of the source's
6//! current value) is swallowed; every subsequent emission lifts to
7//! `Some(Arc<value>)`. Useful for trigger/event semantics where retained values
8//! should not fire downstream effects.
9//!
10//! When used inside `switch_map`, each re-creation gets a fresh cold pipeline
11//! (with its own first-skip), providing per-reconnection suppression of
12//! retained values.
13
14use std::{
15    marker::PhantomData,
16    sync::{
17        Arc,
18        atomic::{AtomicBool, Ordering},
19    },
20};
21
22use super::CellValue;
23use crate::{
24    pipeline::{Empty, Pipeline, PipelineInstall, Seedness},
25    signal::Signal,
26    subscription::SubscriptionGuard,
27};
28
29/// Pipeline node representing `source.cold()`. Output is `Arc<T>` so chains
30/// stay zero-copy; materialize returns `Cell<Option<Arc<T>>>`.
31pub struct ColdPipeline<S, T, Sd = crate::pipeline::Definite> {
32    source: S,
33    _t: PhantomData<fn(T)>,
34    _sd: PhantomData<fn(Sd)>,
35}
36
37impl<S, T, Sd> PipelineInstall<Arc<T>> for ColdPipeline<S, T, Sd>
38where
39    S: PipelineInstall<T> + Send + Sync + 'static,
40    Sd: Seedness,
41    T: CellValue,
42{
43    fn install(&self, callback: Arc<dyn Fn(&Signal<Arc<T>>) + Send + Sync>) -> SubscriptionGuard {
44        let first = Arc::new(AtomicBool::new(true));
45        let wrapped: Arc<dyn Fn(&Signal<T>) + Send + Sync> =
46            Arc::new(move |signal: &Signal<T>| match signal {
47                Signal::Value(v) => {
48                    if first.swap(false, Ordering::SeqCst) {
49                        return;
50                    }
51                    // Forward as Signal<Arc<T>> by Arc-cloning the inner Arc.
52                    callback(&Signal::value_arc(Arc::new(v.clone())));
53                }
54                Signal::Complete => callback(&Signal::Complete),
55                Signal::Error(e) => callback(&Signal::Error(e.clone())),
56            });
57        self.source.install(wrapped)
58    }
59}
60
61#[allow(private_bounds)]
62impl<S, T, Sd> Pipeline<Arc<T>, Empty> for ColdPipeline<S, T, Sd>
63where
64    S: Pipeline<T, Sd>,
65    Sd: Seedness,
66    T: CellValue,
67{
68}
69
70#[allow(private_bounds)]
71pub trait ColdExt<T: CellValue, S: Seedness>: Pipeline<T, S> {
72    /// Drop the synchronous-on-subscribe initial emission; subsequent values
73    /// flow through wrapped in `Arc` for cheap forwarding.
74    ///
75    /// Materializes to `Cell<Option<Arc<T>>>`, initialized to `None`. Once a
76    /// post-subscribe emission arrives, the cell flips to `Some(Arc<value>)`
77    /// and stays `Some` from then on (it tracks the most recent emission).
78    #[track_caller]
79    fn cold(self) -> impl crate::Materialize<Arc<T>, Empty> {
80        ColdPipeline {
81            source: self,
82            _t: PhantomData,
83            _sd: PhantomData,
84        }
85    }
86}
87
88impl<T: CellValue, S: Seedness, P: Pipeline<T, S>> ColdExt<T, S> for P {}
89
90#[cfg(test)]
91mod tests {
92    use std::sync::atomic::{AtomicU64, Ordering as AtomicOrdering};
93
94    use super::*;
95    use crate::{Cell, Gettable, Materialize, Mutable, traits::Watchable};
96
97    #[test]
98    fn test_cold_starts_as_none() {
99        let source = Cell::new(42u64);
100        let cold = source.cold().materialize();
101        assert_eq!(cold.get(), None);
102    }
103
104    #[test]
105    fn test_cold_emits_some_on_change() {
106        let source = Cell::new(42u64);
107        let cold = source.clone().cold().materialize();
108
109        assert_eq!(cold.get(), None);
110
111        source.set(100);
112        assert_eq!(cold.get(), Some(Arc::new(100)));
113    }
114
115    #[test]
116    fn test_cold_does_not_replay_retained_value() {
117        let source = Cell::new(42u64);
118        let cold = source.clone().cold().materialize();
119        let emission_count = Arc::new(AtomicU64::new(0));
120
121        let count = emission_count.clone();
122        let _guard = cold.subscribe(move |signal| {
123            if let Signal::Value(_) = signal {
124                count.fetch_add(1, AtomicOrdering::SeqCst);
125            }
126        });
127
128        // Subscribe fires once with initial None value
129        assert_eq!(emission_count.load(AtomicOrdering::SeqCst), 1);
130        assert_eq!(cold.get(), None); // retained source value (42) was NOT replayed
131
132        source.set(100);
133        assert_eq!(emission_count.load(AtomicOrdering::SeqCst), 2);
134        assert_eq!(cold.get(), Some(Arc::new(100)));
135
136        source.set(200);
137        assert_eq!(emission_count.load(AtomicOrdering::SeqCst), 3);
138        assert_eq!(cold.get(), Some(Arc::new(200)));
139    }
140}
hyphae/traits/operators/cold.rs

hyphae/traits/operators/
cold.rs
