rust_elm/
store.rs

1use std::any::Any;
2use std::collections::VecDeque;
3use std::marker::PhantomData;
4use std::panic::{catch_unwind, AssertUnwindSafe};
5use std::sync::atomic::{AtomicUsize, Ordering};
6use std::sync::Arc;
7use std::time::Duration;
8
9use crossbeam_channel::{Receiver, Sender};
10use parking_lot::Mutex;
11
12use crate::bus::BusSender;
13use crate::cmd::Cmd;
14use crate::effect::EffectId;
15use crate::optics::{Casepath, StateLens};
16use crate::runtime::InterpreterState;
17
18/// A reducer panic caught by [`catch_reduce`] or [`CatchReducer`](crate::reducer::CatchReducer).
19#[derive(Debug, Clone, PartialEq, Eq)]
20pub struct ReducePanic {
21    message: Option<String>,
22}
23
24impl ReducePanic {
25    pub fn message(&self) -> Option<&str> {
26        self.message.as_deref()
27    }
28
29    fn from_payload(payload: Box<dyn Any + Send>) -> Self {
30        let message = payload
31            .downcast_ref::<&str>()
32            .map(|s| (*s).to_string())
33            .or_else(|| payload.downcast_ref::<String>().cloned());
34        Self { message }
35    }
36}
37
38/// Run `reduce` inside [`catch_unwind`]. This is the default for [`Runtime`] and [`CatchReducer`]:
39/// panics are caught but `state` is not reverted. Use [`catch_reduce`] only when you want rollback.
40pub fn catch_reduce_panic<S, M, F>(state: &mut S, reduce: F, action: M) -> Result<Cmd<M>, ReducePanic>
41where
42    F: FnOnce(&mut S, M) -> Cmd<M>,
43{
44    match catch_unwind(AssertUnwindSafe(|| reduce(state, action))) {
45        Ok(cmd) => Ok(cmd),
46        Err(payload) => Err(ReducePanic::from_payload(payload)),
47    }
48}
49
50/// Run `reduce` inside [`catch_unwind`], rolling back via `checkpoint` on panic.
51///
52/// `checkpoint` must hold the last committed state (see [`RollbackCatchReducer`](crate::reducer::RollbackCatchReducer)).
53/// On panic, `state` and `checkpoint` are swapped — no clone on the failure path. After a
54/// successful reduce, `checkpoint` is updated from `state` (one clone on the success path only).
55pub fn catch_reduce<S, M, F>(
56    state: &mut S,
57    checkpoint: &mut S,
58    reduce: F,
59    action: M,
60) -> Result<Cmd<M>, ReducePanic>
61where
62    S: Clone,
63    F: FnOnce(&mut S, M) -> Cmd<M>,
64{
65    match catch_unwind(AssertUnwindSafe(|| reduce(state, action))) {
66        Ok(cmd) => {
67            checkpoint.clone_from(state);
68            Ok(cmd)
69        }
70        Err(payload) => {
71            std::mem::swap(state, checkpoint);
72            Err(ReducePanic::from_payload(payload))
73        }
74    }
75}
76
77/// Calls [`StoreBackend::end_store_work`] on drop unless [`Self::disarm`]d.
78///
79/// Mirrors stack-unwind / `finally` semantics so [`StoreTask`] waiters are not left
80/// hanging when a dispatch aborts (e.g. reducer panic).
81pub(crate) struct StoreWorkUnwindGuard<'a, S, M>
82where
83    S: Send + 'static,
84    M: Send + 'static,
85{
86    backend: &'a StoreBackend<S, M>,
87    active: bool,
88}
89
90impl<'a, S, M> StoreWorkUnwindGuard<'a, S, M>
91where
92    S: Send + 'static,
93    M: Send + 'static,
94{
95    pub(crate) fn new(backend: &'a StoreBackend<S, M>) -> Self {
96        Self {
97            backend,
98            active: true,
99        }
100    }
101
102    pub(crate) fn disarm(&mut self) {
103        self.active = false;
104    }
105}
106
107impl<S, M> Drop for StoreWorkUnwindGuard<'_, S, M>
108where
109    S: Send + 'static,
110    M: Send + 'static,
111{
112    fn drop(&mut self) {
113        if self.active {
114            self.backend.end_store_work();
115        }
116    }
117}
118
119/// Cloneable dispatch handle for a running [`Runtime`](crate::Runtime).
120///
121/// `update` runs on the runtime thread — actions are never applied synchronously inside
122/// `send`, so reducers cannot re-enter themselves from the caller's stack (UDF parity).
123pub struct Store<S, M> {
124    pub(crate) backend: StoreBackend<S, M>,
125}
126
127impl<S, M> Clone for Store<S, M>
128where
129    S: Send + 'static,
130    M: Send + 'static,
131{
132    fn clone(&self) -> Self {
133        Self {
134            backend: self.backend.clone(),
135        }
136    }
137}
138
139pub(crate) struct StoreBackend<S, M> {
140    pub state: Arc<Mutex<S>>,
141    pub sender: BusSender<M>,
142    state_listeners: Arc<Mutex<Vec<Sender<()>>>>,
143    effect_done: Arc<Mutex<VecDeque<Sender<()>>>>,
144    in_flight: Arc<AtomicUsize>,
145    pub interpreter: Arc<InterpreterState<M>>,
146}
147
148impl<S, M> Clone for StoreBackend<S, M>
149where
150    S: Send + 'static,
151    M: Send + 'static,
152{
153    fn clone(&self) -> Self {
154        Self {
155            state: self.state.clone(),
156            sender: self.sender.clone(),
157            state_listeners: self.state_listeners.clone(),
158            effect_done: self.effect_done.clone(),
159            in_flight: self.in_flight.clone(),
160            interpreter: self.interpreter.clone(),
161        }
162    }
163}
164
165impl<S, M> StoreBackend<S, M>
166where
167    S: Send + 'static,
168    M: Send + 'static,
169{
170    pub fn new(
171        state: Arc<Mutex<S>>,
172        sender: BusSender<M>,
173        interpreter: Arc<InterpreterState<M>>,
174    ) -> Self {
175        Self {
176            state,
177            sender,
178            state_listeners: Arc::new(Mutex::new(Vec::new())),
179            effect_done: Arc::new(Mutex::new(VecDeque::new())),
180            in_flight: Arc::new(AtomicUsize::new(0)),
181            interpreter,
182        }
183    }
184
185    pub fn begin_store_work(&self) {
186        self.in_flight.fetch_add(1, Ordering::SeqCst);
187    }
188
189    pub fn end_store_work(&self) {
190        let prev = self.in_flight.fetch_sub(1, Ordering::SeqCst);
191        if prev == 1 {
192            self.notify_state();
193            self.signal_effect_done();
194        }
195    }
196
197    pub fn store(&self) -> Store<S, M> {
198        Store {
199            backend: self.clone(),
200        }
201    }
202
203    pub fn notify_state(&self) {
204        self.state_listeners
205            .lock()
206            .retain(|tx| tx.send(()).is_ok());
207    }
208
209    pub fn pop_effect_done(&self) -> Option<Sender<()>> {
210        self.effect_done.lock().pop_front()
211    }
212
213    pub fn signal_effect_done(&self) {
214        if let Some(tx) = self.pop_effect_done() {
215            let _ = tx.send(());
216        }
217    }
218}
219
220impl<S, M> Store<S, M>
221where
222    S: Send + Sync + Clone + 'static,
223    M: Send + 'static,
224{
225    /// Dispatch an action and return a handle that completes when its effects finish.
226    pub fn send(&self, action: M) -> StoreTask {
227        let (tx, rx) = crossbeam_channel::bounded(1);
228        self.backend.effect_done.lock().push_back(tx);
229        self.backend.begin_store_work();
230        let _ = self.backend.sender.send_blocking(action);
231        StoreTask { rx }
232    }
233
234    /// Fire-and-forget dispatch.
235    pub fn dispatch(&self, action: M) {
236        let _ = self.send(action);
237    }
238
239    pub fn state(&self) -> S {
240        self.backend.state.lock().clone()
241    }
242
243    pub fn cancel(&self, id: EffectId) {
244        if let Some(handle) = self.backend.interpreter.cancel_tokens.lock().remove(&id) {
245            handle.abort();
246        }
247    }
248
249    /// Subscribe to deduplicated `Arc` state snapshots (skips consecutive equal states).
250    pub fn subscribe_state(&self) -> StateSubscriber<S>
251    where
252        S: Clone + PartialEq,
253    {
254        let (tx, rx) = crossbeam_channel::unbounded();
255        self.backend.state_listeners.lock().push(tx);
256        StateSubscriber {
257            rx,
258            state: self.backend.state.clone(),
259            last: Some(Arc::new(self.backend.state.lock().clone())),
260        }
261    }
262
263    /// Focus a child store via state and action casepaths (see [`crate::optics`]).
264    pub fn scope<CS, CM, AK, SK>(
265        &self,
266        state_kp: SK,
267        action_kp: AK,
268    ) -> ScopedStore<S, M, CS, CM, AK, SK>
269    where
270        CS: Clone + PartialEq + Send + Sync + 'static,
271        CM: Clone + Send + 'static,
272        S: 'static,
273        M: 'static,
274        AK: Casepath<M, CM> + Clone + Send + Sync + 'static,
275        SK: StateLens<S, CS> + Clone,
276    {
277        ScopedStore {
278            store: self.clone(),
279            state_kp,
280            action_kp,
281            _marker: PhantomData,
282        }
283    }
284}
285
286/// Awaitable handle for in-flight effects from a single [`Store::send`].
287pub struct StoreTask {
288    rx: Receiver<()>,
289}
290
291impl StoreTask {
292    pub fn finish(self) -> Result<(), StoreTaskError> {
293        self.rx
294            .recv_timeout(Duration::from_secs(5))
295            .map_err(|_| StoreTaskError::Timeout)
296    }
297}
298
299#[derive(Debug, Clone, PartialEq, Eq)]
300pub enum StoreTaskError {
301    Timeout,
302}
303
304impl std::fmt::Display for StoreTaskError {
305    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
306        match self {
307            Self::Timeout => write!(f, "timed out waiting for store effects to finish"),
308        }
309    }
310}
311
312impl std::error::Error for StoreTaskError {}
313
314/// Iterator-like subscription to state snapshots.
315pub struct StateSubscriber<S> {
316    rx: Receiver<()>,
317    state: Arc<Mutex<S>>,
318    last: Option<Arc<S>>,
319}
320
321impl<S: PartialEq + Clone> StateSubscriber<S> {
322    pub fn next(&mut self) -> Option<Arc<S>> {
323        loop {
324            match self.rx.try_recv() {
325                Ok(()) => {
326                    while self.rx.try_recv().is_ok() {}
327                    let snapshot = Arc::new(self.state.lock().clone());
328                    if self
329                        .last
330                        .as_ref()
331                        .is_some_and(|prev| prev.as_ref() == snapshot.as_ref())
332                    {
333                        continue;
334                    }
335                    self.last = Some(snapshot.clone());
336                    return Some(snapshot);
337                }
338                Err(crossbeam_channel::TryRecvError::Empty) => return None,
339                Err(crossbeam_channel::TryRecvError::Disconnected) => return None,
340            }
341        }
342    }
343
344    pub fn wait_next(&mut self, timeout: Duration) -> Option<Arc<S>> {
345        let deadline = std::time::Instant::now() + timeout;
346        while std::time::Instant::now() < deadline {
347            if let Some(s) = self.next() {
348                return Some(s);
349            }
350            std::thread::sleep(Duration::from_millis(5));
351        }
352        None
353    }
354
355    pub fn latest(&self) -> Option<Arc<S>> {
356        self.last.clone()
357    }
358}
359
360/// Child store routing actions through a parent action casepath.
361pub struct ScopedStore<S: 'static, M: 'static, CS: 'static, CM: 'static, AK: 'static, SK>
362where
363    SK: StateLens<S, CS> + Clone,
364{
365    store: Store<S, M>,
366    state_kp: SK,
367    action_kp: AK,
368    _marker: PhantomData<(M, CM, S, CS)>,
369}
370
371impl<S: 'static, M: 'static, CS: 'static, CM: 'static, AK: 'static, SK> Clone
372    for ScopedStore<S, M, CS, CM, AK, SK>
373where
374    S: Send,
375    M: Send,
376    AK: Clone,
377    SK: StateLens<S, CS> + Clone,
378{
379    fn clone(&self) -> Self {
380        Self {
381            store: self.store.clone(),
382            state_kp: self.state_kp.clone(),
383            action_kp: self.action_kp.clone(),
384            _marker: PhantomData,
385        }
386    }
387}
388
389impl<S: 'static, M: 'static, CS: 'static, CM: 'static, AK: 'static, SK>
390    ScopedStore<S, M, CS, CM, AK, SK>
391where
392    S: Send + Sync + Clone,
393    M: Send,
394    CS: Clone + PartialEq + Send + Sync,
395    CM: Clone + Send,
396    AK: Casepath<M, CM> + Clone + Send + Sync + 'static,
397    SK: StateLens<S, CS> + Clone,
398{
399    pub fn send(&self, action: CM) -> StoreTask {
400        self.store.send(self.action_kp.wrap(action))
401    }
402
403    pub fn dispatch(&self, action: CM) {
404        self.store.dispatch(self.action_kp.wrap(action));
405    }
406
407    pub fn child_state(&self) -> Option<CS> {
408        self.state_kp.focus(&self.store.state()).cloned()
409    }
410
411    pub fn subscribe_state(&self) -> ScopedStateSubscriber<S, CS, SK>
412    where
413        S: Clone + PartialEq,
414        SK: Clone,
415    {
416        ScopedStateSubscriber {
417            inner: self.store.subscribe_state(),
418            state_kp: self.state_kp.clone(),
419            _marker: PhantomData,
420        }
421    }
422}
423
424pub struct ScopedStateSubscriber<S: 'static, CS: 'static, SK>
425where
426    SK: StateLens<S, CS> + Clone,
427{
428    inner: StateSubscriber<S>,
429    state_kp: SK,
430    _marker: PhantomData<(S, CS)>,
431}
432
433impl<S, CS, SK> ScopedStateSubscriber<S, CS, SK>
434where
435    S: PartialEq + Clone,
436    CS: Clone + PartialEq,
437    SK: StateLens<S, CS> + Clone,
438{
439    pub fn next(&mut self) -> Option<CS> {
440        loop {
441            let parent = self.inner.next()?;
442            let Some(child) = self.state_kp.focus(parent.as_ref()).cloned() else {
443                continue;
444            };
445            return Some(child);
446        }
447    }
448}
rust_elm/store.rs

rust_elm/
store.rs
