rust_elm/runtime/
store.rs

1use std::collections::VecDeque;
2use std::hash::Hash;
3use std::marker::PhantomData;
4use std::sync::atomic::{AtomicUsize, Ordering};
5use std::sync::Arc;
6use std::time::Duration;
7
8use crossbeam_channel::{Receiver, Sender};
9use key_paths_core::{FieldDiff, hash_value};
10use parking_lot::Mutex;
11
12use crate::bus::BusSender;
13use crate::effect::EffectId;
14use crate::optics::Casepath;
15use crate::runtime::binding::StateBinding;
16use key_paths_core::RefKpTrait;
17use super::interpreter::InterpreterState;
18
19/// Calls [`StoreWork::end_store_work`] on drop unless [`Self::disarm`]d.
20///
21/// Mirrors stack-unwind / `finally` semantics so [`StoreTask`] waiters are not left
22/// hanging when a dispatch aborts (e.g. reducer panic).
23pub(crate) struct StoreWorkUnwindGuard<'a, S, M, B: StoreWork<S, M> + ?Sized>
24where
25    S: Send + 'static,
26    M: Send + 'static,
27{
28    backend: &'a B,
29    active: bool,
30    _marker: PhantomData<(S, M)>,
31}
32
33impl<'a, S, M, B> StoreWorkUnwindGuard<'a, S, M, B>
34where
35    S: Send + 'static,
36    M: Send + 'static,
37    B: StoreWork<S, M> + ?Sized,
38{
39    pub(crate) fn new(backend: &'a B) -> Self {
40        Self {
41            backend,
42            active: true,
43            _marker: PhantomData,
44        }
45    }
46
47    pub(crate) fn disarm(&mut self) {
48        self.active = false;
49    }
50}
51
52impl<S, M, B> Drop for StoreWorkUnwindGuard<'_, S, M, B>
53where
54    S: Send + 'static,
55    M: Send + 'static,
56    B: StoreWork<S, M> + ?Sized,
57{
58    fn drop(&mut self) {
59        if self.active {
60            self.backend.end_store_work();
61        }
62    }
63}
64
65/// Cloneable dispatch handle for a running [`Runtime`](crate::Runtime).
66///
67/// `update` runs on the runtime thread — actions are never applied synchronously inside
68/// `send`, so reducers cannot re-enter themselves from the caller's stack (UDF parity).
69///
70/// See also [`RwStore`](crate::runtime::rw_store::RwStore) and, with the `arc-swap` feature,
71/// [`SwapStore`](crate::runtime::swap_store::SwapStore).
72pub struct Store<S, M> {
73    pub(crate) backend: StoreBackend<S, M>,
74}
75
76impl<S, M> Clone for Store<S, M>
77where
78    S: Send + 'static,
79    M: Send + 'static,
80{
81    fn clone(&self) -> Self {
82        Self {
83            backend: self.backend.clone(),
84        }
85    }
86}
87
88pub(crate) struct StoreHub<S, M> {
89    pub sender: BusSender<M>,
90    pub state_listeners: Arc<Mutex<Vec<Sender<()>>>>,
91    pub effect_done: Arc<Mutex<VecDeque<Sender<()>>>>,
92    pub in_flight: Arc<AtomicUsize>,
93    pub interpreter: Arc<InterpreterState<M>>,
94    _marker: PhantomData<S>,
95}
96
97/// Shared store-work accounting (in-flight dispatches, listener notify, effect-done signals).
98pub(crate) trait StoreWork<S, M>
99where
100    S: Send + 'static,
101    M: Send + 'static,
102{
103    fn hub(&self) -> &Arc<StoreHub<S, M>>;
104
105    fn begin_store_work(&self) {
106        self.hub().in_flight.fetch_add(1, Ordering::SeqCst);
107    }
108
109    fn end_store_work(&self) {
110        let prev = self.hub().in_flight.fetch_sub(1, Ordering::SeqCst);
111        if prev == 1 {
112            self.hub().notify_state();
113            self.hub().signal_effect_done();
114        }
115    }
116
117    fn notify_state(&self) {
118        self.hub().notify_state();
119    }
120}
121
122impl<S, M> StoreHub<S, M>
123where
124    S: Send + 'static,
125    M: Send + 'static,
126{
127    pub fn new(sender: BusSender<M>, interpreter: Arc<InterpreterState<M>>) -> Arc<Self> {
128        Arc::new(Self {
129            sender,
130            state_listeners: Arc::new(Mutex::new(Vec::new())),
131            effect_done: Arc::new(Mutex::new(VecDeque::new())),
132            in_flight: Arc::new(AtomicUsize::new(0)),
133            interpreter,
134            _marker: PhantomData,
135        })
136    }
137
138    pub fn notify_state(&self) {
139        self.state_listeners
140            .lock()
141            .retain(|tx| tx.send(()).is_ok());
142    }
143
144    pub fn pop_effect_done(&self) -> Option<Sender<()>> {
145        self.effect_done.lock().pop_front()
146    }
147
148    pub fn signal_effect_done(&self) {
149        if let Some(tx) = self.pop_effect_done() {
150            let _ = tx.send(());
151        }
152    }
153}
154
155pub(crate) struct StoreBackend<S, M> {
156    pub state: Arc<Mutex<S>>,
157    pub hub: Arc<StoreHub<S, M>>,
158}
159
160impl<S, M> Clone for StoreBackend<S, M>
161where
162    S: Send + 'static,
163    M: Send + 'static,
164{
165    fn clone(&self) -> Self {
166        Self {
167            state: self.state.clone(),
168            hub: self.hub.clone(),
169        }
170    }
171}
172
173impl<S, M> StoreWork<S, M> for StoreBackend<S, M>
174where
175    S: Send + 'static,
176    M: Send + 'static,
177{
178    fn hub(&self) -> &Arc<StoreHub<S, M>> {
179        &self.hub
180    }
181}
182
183impl<S, M> StoreBackend<S, M>
184where
185    S: Send + 'static,
186    M: Send + 'static,
187{
188    pub fn new(
189        state: Arc<Mutex<S>>,
190        sender: BusSender<M>,
191        interpreter: Arc<InterpreterState<M>>,
192    ) -> Self {
193        Self {
194            state,
195            hub: StoreHub::new(sender, interpreter),
196        }
197    }
198
199    pub fn store(&self) -> Store<S, M> {
200        Store {
201            backend: self.clone(),
202        }
203    }
204}
205
206impl<S, M> Store<S, M>
207where
208    S: Send + Sync + Clone + 'static,
209    M: Send + 'static,
210{
211    /// Dispatch an action and return a handle that completes when its effects finish.
212    pub fn send(&self, action: M) -> StoreTask {
213        let (tx, rx) = crossbeam_channel::bounded(1);
214        self.backend.hub.effect_done.lock().push_back(tx);
215        self.backend.begin_store_work();
216        let _ = self.backend.hub.sender.send_blocking(action);
217        StoreTask::from_rx(rx)
218    }
219
220    /// Fire-and-forget dispatch.
221    pub fn dispatch(&self, action: M) {
222        let _ = self.send(action);
223    }
224
225    pub fn state(&self) -> S {
226        self.backend.state.lock().clone()
227    }
228
229    /// Zero-copy read/write handle to store state (no clone of `S`).
230    pub fn binding(&self) -> StateBinding<S> {
231        StateBinding::new(Arc::clone(&self.backend.state))
232    }
233
234    pub fn cancel(&self, id: EffectId) {
235        if let Some(handle) = self.backend.hub.interpreter.cancel_tokens.lock().remove(&id) {
236            handle.abort();
237        }
238    }
239
240    /// Subscribe to deduplicated `Arc` state snapshots (skips consecutive equal states).
241    pub fn subscribe_state(&self) -> StateSubscriber<S>
242    where
243        S: Clone + PartialEq,
244    {
245        let (tx, rx) = crossbeam_channel::unbounded();
246        self.backend.hub.state_listeners.lock().push(tx);
247        StateSubscriber {
248            rx,
249            state: self.backend.state.clone(),
250            last: Some(Arc::new(self.backend.state.lock().clone())),
251        }
252    }
253
254    /// Subscribe to field-level change signals without cloning whole state.
255    ///
256    /// Dedup uses root + per-field hashes ([`FieldDiff`]); subscribers read changed
257    /// fields on demand via keypaths / [`StateBinding`].
258    pub fn subscribe_changes(&self) -> ChangeSubscriber<S>
259    where
260        S: FieldDiff + Hash,
261    {
262        let (tx, rx) = crossbeam_channel::unbounded();
263        self.backend.hub.state_listeners.lock().push(tx);
264        let (last_hash, last_fields) = {
265            let guard = self.backend.state.lock();
266            let root = hash_value(&*guard);
267            let mut fields = Vec::new();
268            guard.field_hashes(&mut fields);
269            (Some(root), fields)
270        };
271        ChangeSubscriber {
272            rx,
273            state: self.backend.state.clone(),
274            last_hash,
275            last_fields,
276            _marker: PhantomData,
277        }
278    }
279
280    /// Focus a child store via state and action casepaths (see [`crate::optics`]).
281    pub fn scope<CS, CM, AK, SK>(
282        &self,
283        state_kp: SK,
284        action_kp: AK,
285    ) -> ScopedStore<S, M, CS, CM, AK, SK>
286    where
287        CS: Clone + PartialEq + Send + Sync + 'static,
288        CM: Clone + Send + 'static,
289        S: 'static,
290        M: 'static,
291        AK: Casepath<M, CM> + Clone + Send + Sync + 'static,
292        SK: RefKpTrait<S, CS> + Clone,
293    {
294        ScopedStore {
295            store: self.clone(),
296            state_kp,
297            action_kp,
298            _marker: PhantomData,
299        }
300    }
301}
302
303/// Awaitable handle for in-flight effects from a single [`Store::send`].
304pub struct StoreTask {
305    rx: Receiver<()>,
306}
307
308impl StoreTask {
309    pub(crate) fn from_rx(rx: Receiver<()>) -> Self {
310        Self { rx }
311    }
312
313    pub fn finish(self) -> Result<(), StoreTaskError> {
314        self.finish_with_timeout(Duration::from_secs(5))
315    }
316
317    pub fn finish_with_timeout(self, timeout: Duration) -> Result<(), StoreTaskError> {
318        self.rx
319            .recv_timeout(timeout)
320            .map_err(|_| StoreTaskError::Timeout)
321    }
322}
323
324#[derive(Debug, Clone, PartialEq, Eq)]
325pub enum StoreTaskError {
326    Timeout,
327}
328
329impl std::fmt::Display for StoreTaskError {
330    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
331        match self {
332            Self::Timeout => write!(f, "timed out waiting for store effects to finish"),
333        }
334    }
335}
336
337impl std::error::Error for StoreTaskError {}
338
339/// Which top-level (or derived) fields changed since the last signal.
340#[derive(Debug, Clone, PartialEq, Eq)]
341pub struct ChangeSet<P> {
342    pub paths: Vec<P>,
343}
344
345/// Iterator-like subscription to field change signals (no full-state clone).
346pub struct ChangeSubscriber<S>
347where
348    S: FieldDiff + Hash,
349{
350    rx: Receiver<()>,
351    state: Arc<Mutex<S>>,
352    last_hash: Option<u64>,
353    last_fields: Vec<(S::Path, u64)>,
354    _marker: PhantomData<S>,
355}
356
357impl<S> ChangeSubscriber<S>
358where
359    S: FieldDiff + Hash,
360{
361    fn snapshot_fields(&self) -> (u64, Vec<(S::Path, u64)>) {
362        let guard = self.state.lock();
363        let root = hash_value(&*guard);
364        let mut fields = Vec::new();
365        guard.field_hashes(&mut fields);
366        (root, fields)
367    }
368
369    fn diff_fields(&self, fields: &[(S::Path, u64)]) -> Vec<S::Path> {
370        fields
371            .iter()
372            .filter(|(path, hash)| {
373                self.last_fields
374                    .iter()
375                    .find(|(prev_path, _)| prev_path == path)
376                    .is_none_or(|(_, prev_hash)| prev_hash != hash)
377            })
378            .map(|(path, _)| *path)
379            .collect()
380    }
381
382    /// Zero-copy binding to read changed fields after a signal.
383    pub fn binding(&self) -> StateBinding<S> {
384        StateBinding::new(Arc::clone(&self.state))
385    }
386
387    #[allow(clippy::should_implement_trait)]
388    pub fn next(&mut self) -> Option<ChangeSet<S::Path>> {
389        loop {
390            match self.rx.try_recv() {
391                Ok(()) => {
392                    while self.rx.try_recv().is_ok() {}
393                    let (root_hash, fields) = self.snapshot_fields();
394                    if Some(root_hash) == self.last_hash {
395                        continue;
396                    }
397                    let paths = self.diff_fields(&fields);
398                    self.last_hash = Some(root_hash);
399                    self.last_fields = fields;
400                    if paths.is_empty() {
401                        continue;
402                    }
403                    return Some(ChangeSet { paths });
404                }
405                Err(crossbeam_channel::TryRecvError::Empty) => return None,
406                Err(crossbeam_channel::TryRecvError::Disconnected) => return None,
407            }
408        }
409    }
410
411    pub fn wait_next(&mut self, timeout: Duration) -> Option<ChangeSet<S::Path>> {
412        let deadline = std::time::Instant::now() + timeout;
413        while std::time::Instant::now() < deadline {
414            if let Some(change) = self.next() {
415                return Some(change);
416            }
417            std::thread::sleep(Duration::from_millis(5));
418        }
419        None
420    }
421}
422
423/// Iterator-like subscription to state snapshots.
424pub struct StateSubscriber<S> {
425    rx: Receiver<()>,
426    state: Arc<Mutex<S>>,
427    last: Option<Arc<S>>,
428}
429
430impl<S: PartialEq + Clone> StateSubscriber<S> {
431    #[allow(clippy::should_implement_trait)]
432    pub fn next(&mut self) -> Option<Arc<S>> {
433        loop {
434            match self.rx.try_recv() {
435                Ok(()) => {
436                    while self.rx.try_recv().is_ok() {}
437                    let snapshot = Arc::new(self.state.lock().clone());
438                    if self
439                        .last
440                        .as_ref()
441                        .is_some_and(|prev| prev.as_ref() == snapshot.as_ref())
442                    {
443                        continue;
444                    }
445                    self.last = Some(snapshot.clone());
446                    return Some(snapshot);
447                }
448                Err(crossbeam_channel::TryRecvError::Empty) => return None,
449                Err(crossbeam_channel::TryRecvError::Disconnected) => return None,
450            }
451        }
452    }
453
454    pub fn wait_next(&mut self, timeout: Duration) -> Option<Arc<S>> {
455        let deadline = std::time::Instant::now() + timeout;
456        while std::time::Instant::now() < deadline {
457            if let Some(s) = self.next() {
458                return Some(s);
459            }
460            std::thread::sleep(Duration::from_millis(5));
461        }
462        None
463    }
464
465    pub fn latest(&self) -> Option<Arc<S>> {
466        self.last.clone()
467    }
468}
469
470/// Child store routing actions through a parent action casepath.
471pub struct ScopedStore<S: 'static, M: 'static, CS: 'static, CM: 'static, AK: 'static, SK>
472where
473    SK: RefKpTrait<S, CS> + Clone,
474{
475    store: Store<S, M>,
476    state_kp: SK,
477    action_kp: AK,
478    _marker: PhantomData<(M, CM, S, CS)>,
479}
480
481impl<S: 'static, M: 'static, CS: 'static, CM: 'static, AK: 'static, SK> Clone
482    for ScopedStore<S, M, CS, CM, AK, SK>
483where
484    S: Send,
485    M: Send,
486    AK: Clone,
487    SK: RefKpTrait<S, CS> + Clone,
488{
489    fn clone(&self) -> Self {
490        Self {
491            store: self.store.clone(),
492            state_kp: self.state_kp.clone(),
493            action_kp: self.action_kp.clone(),
494            _marker: PhantomData,
495        }
496    }
497}
498
499impl<S: 'static, M: 'static, CS: 'static, CM: 'static, AK: 'static, SK>
500    ScopedStore<S, M, CS, CM, AK, SK>
501where
502    S: Send + Sync + Clone,
503    M: Send,
504    CS: Clone + PartialEq + Send + Sync,
505    CM: Clone + Send,
506    AK: Casepath<M, CM> + Clone + Send + Sync + 'static,
507    SK: RefKpTrait<S, CS> + Clone,
508{
509    pub fn send(&self, action: CM) -> StoreTask {
510        self.store.send(self.action_kp.wrap(action))
511    }
512
513    pub fn dispatch(&self, action: CM) {
514        self.store.dispatch(self.action_kp.wrap(action));
515    }
516
517    pub fn child_state(&self) -> Option<CS> {
518        self.state_kp.focus(&self.store.state()).cloned()
519    }
520
521    /// Zero-copy binding projected to the scoped child via keypath.
522    pub fn binding(&self) -> crate::runtime::binding::ProjectedBinding<S, CS, SK> {
523        self.store.binding().project(self.state_kp.clone())
524    }
525
526    pub fn subscribe_state(&self) -> ScopedStateSubscriber<S, CS, SK>
527    where
528        S: Clone + PartialEq,
529        SK: Clone,
530    {
531        ScopedStateSubscriber {
532            inner: self.store.subscribe_state(),
533            state_kp: self.state_kp.clone(),
534            _marker: PhantomData,
535        }
536    }
537
538    /// Field-level changes for the scoped child (hashes computed under lock, no clone of `S`).
539    pub fn subscribe_changes(&self) -> ScopedChangeSubscriber<S, CS, SK>
540    where
541        CS: FieldDiff + Hash,
542    {
543        let (tx, rx) = crossbeam_channel::unbounded();
544        self.store.backend.hub.state_listeners.lock().push(tx);
545        let (last_hash, last_fields) = {
546            let guard = self.store.backend.state.lock();
547            if let Some(child) = self.state_kp.focus(&*guard) {
548                let root = hash_value(child);
549                let mut fields = Vec::new();
550                child.field_hashes(&mut fields);
551                (Some(root), fields)
552            } else {
553                (None, Vec::new())
554            }
555        };
556        ScopedChangeSubscriber {
557            rx,
558            state: self.store.backend.state.clone(),
559            state_kp: self.state_kp.clone(),
560            last_hash,
561            last_fields,
562            _marker: PhantomData,
563        }
564    }
565}
566
567pub struct ScopedStateSubscriber<S: 'static, CS: 'static, SK>
568where
569    SK: RefKpTrait<S, CS> + Clone,
570{
571    inner: StateSubscriber<S>,
572    state_kp: SK,
573    _marker: PhantomData<(S, CS)>,
574}
575
576impl<S, CS, SK> ScopedStateSubscriber<S, CS, SK>
577where
578    S: PartialEq + Clone,
579    CS: Clone + PartialEq,
580    SK: RefKpTrait<S, CS> + Clone,
581{
582    #[allow(clippy::should_implement_trait)]
583    pub fn next(&mut self) -> Option<CS> {
584        loop {
585            let parent = self.inner.next()?;
586            let Some(child) = self.state_kp.focus(parent.as_ref()).cloned() else {
587                continue;
588            };
589            return Some(child);
590        }
591    }
592}
593
594/// Field change subscription focused to a scoped child store.
595pub struct ScopedChangeSubscriber<S: 'static, CS: 'static, SK>
596where
597    CS: FieldDiff,
598    SK: RefKpTrait<S, CS> + Clone,
599{
600    rx: Receiver<()>,
601    state: Arc<Mutex<S>>,
602    state_kp: SK,
603    last_hash: Option<u64>,
604    last_fields: Vec<(CS::Path, u64)>,
605    _marker: PhantomData<(S, CS)>,
606}
607
608impl<S, CS, SK> ScopedChangeSubscriber<S, CS, SK>
609where
610    S: Send + Sync,
611    CS: FieldDiff + Hash + Send + Sync,
612    SK: RefKpTrait<S, CS> + Clone,
613{
614    fn snapshot_fields(&self) -> Option<(u64, Vec<(CS::Path, u64)>)> {
615        let guard = self.state.lock();
616        let child = self.state_kp.focus(&*guard)?;
617        let root = hash_value(child);
618        let mut fields = Vec::new();
619        child.field_hashes(&mut fields);
620        Some((root, fields))
621    }
622
623    fn diff_fields(&self, fields: &[(CS::Path, u64)]) -> Vec<CS::Path> {
624        fields
625            .iter()
626            .filter(|(path, hash)| {
627                self.last_fields
628                    .iter()
629                    .find(|(prev_path, _)| prev_path == path)
630                    .is_none_or(|(_, prev_hash)| prev_hash != hash)
631            })
632            .map(|(path, _)| *path)
633            .collect()
634    }
635
636    pub fn binding(&self) -> crate::runtime::binding::ProjectedBinding<S, CS, SK> {
637        crate::runtime::binding::StateBinding::new(Arc::clone(&self.state))
638            .project(self.state_kp.clone())
639    }
640
641    #[allow(clippy::should_implement_trait)]
642    pub fn next(&mut self) -> Option<ChangeSet<CS::Path>> {
643        loop {
644            match self.rx.try_recv() {
645                Ok(()) => {
646                    while self.rx.try_recv().is_ok() {}
647                    let Some((root_hash, fields)) = self.snapshot_fields() else {
648                        continue;
649                    };
650                    if Some(root_hash) == self.last_hash {
651                        continue;
652                    }
653                    let paths = self.diff_fields(&fields);
654                    self.last_hash = Some(root_hash);
655                    self.last_fields = fields;
656                    if paths.is_empty() {
657                        continue;
658                    }
659                    return Some(ChangeSet { paths });
660                }
661                Err(crossbeam_channel::TryRecvError::Empty) => return None,
662                Err(crossbeam_channel::TryRecvError::Disconnected) => return None,
663            }
664        }
665    }
666
667    pub fn wait_next(&mut self, timeout: Duration) -> Option<ChangeSet<CS::Path>> {
668        let deadline = std::time::Instant::now() + timeout;
669        while std::time::Instant::now() < deadline {
670            if let Some(change) = self.next() {
671                return Some(change);
672            }
673            std::thread::sleep(Duration::from_millis(5));
674        }
675        None
676    }
677}
rust_elm/runtime/store.rs

rust_elm/runtime/
store.rs
