rust_elm/
store.rs

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

rust_elm/
store.rs
