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

rust_elm/runtime/
store.rs
