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

rust_elm/runtime/
store.rs
