rust_elm/runtime/
rw_store.rs

1//! [`RwLock`](parking_lot::RwLock)-backed store for read-heavy workloads.
2//!
3//! Use one [`RwStore`] handle from [`RwRuntime::rw_store`](crate::RwRuntime::rw_store).
4//! Call [`RwStore::read_store`] when you need concurrent readers — you do not need a
5//! separate runtime accessor.
6
7use std::hash::Hash;
8use std::marker::PhantomData;
9use std::sync::Arc;
10use std::time::Duration;
11
12use crossbeam_channel::Receiver;
13use key_paths_core::{FieldDiff, hash_value};
14use parking_lot::RwLock;
15
16use crate::effect::EffectId;
17use crate::optics::Casepath;
18use crate::runtime::binding::{ReadStateBinding, RwStateBinding};
19use crate::runtime::state_access::StateRead;
20use crate::runtime::store::{ChangeSet, StoreHub, StoreTask, StoreWork};
21use key_paths_core::RefKpTrait;
22
23pub(crate) struct RwStoreBackend<S, M> {
24    pub state: Arc<RwLock<S>>,
25    pub hub: Arc<StoreHub<S, M>>,
26}
27
28impl<S, M> Clone for RwStoreBackend<S, M>
29where
30    S: Send + 'static,
31    M: Send + 'static,
32{
33    fn clone(&self) -> Self {
34        Self {
35            state: self.state.clone(),
36            hub: self.hub.clone(),
37        }
38    }
39}
40
41impl<S, M> StoreWork<S, M> for RwStoreBackend<S, M>
42where
43    S: Send + 'static,
44    M: Send + 'static,
45{
46    fn hub(&self) -> &Arc<StoreHub<S, M>> {
47        &self.hub
48    }
49}
50
51impl<S, M> RwStoreBackend<S, M>
52where
53    S: Send + 'static,
54    M: Send + 'static,
55{
56    pub fn new(state: Arc<RwLock<S>>, hub: Arc<StoreHub<S, M>>) -> Self {
57        Self { state, hub }
58    }
59
60    pub fn rw_store(&self) -> RwStore<S, M> {
61        RwStore {
62            backend: self.clone(),
63        }
64    }
65
66    pub fn read_store(&self) -> ReadStore<S, M> {
67        ReadStore {
68            backend: self.clone(),
69        }
70    }
71}
72
73/// Writable store backed by [`RwLock`]. Readers use [`Self::read_store`].
74pub struct RwStore<S, M> {
75    pub(crate) backend: RwStoreBackend<S, M>,
76}
77
78impl<S, M> Clone for RwStore<S, M>
79where
80    S: Send + 'static,
81    M: Send + 'static,
82{
83    fn clone(&self) -> Self {
84        Self {
85            backend: self.backend.clone(),
86        }
87    }
88}
89
90impl<S, M> RwStore<S, M>
91where
92    S: Send + Sync + Clone + 'static,
93    M: Send + 'static,
94{
95    pub fn send(&self, action: M) -> StoreTask {
96        let (tx, rx) = crossbeam_channel::bounded(1);
97        self.backend.hub.effect_done.lock().push_back(tx);
98        self.backend.begin_store_work();
99        let _ = self.backend.hub.sender.send_blocking(action);
100        StoreTask::from_rx(rx)
101    }
102
103    pub fn dispatch(&self, action: M) {
104        let _ = self.send(action);
105    }
106
107    pub fn state(&self) -> S {
108        self.backend.state.with_read(|s| s.clone())
109    }
110
111    pub fn binding(&self) -> RwStateBinding<S> {
112        RwStateBinding::new(Arc::clone(&self.backend.state))
113    }
114
115    /// Read-only view for concurrent `read()` — obtain per reader thread from one [`RwStore`].
116    pub fn read_store(&self) -> ReadStore<S, M> {
117        self.backend.read_store()
118    }
119
120    pub fn cancel(&self, id: EffectId) {
121        if let Some(handle) = self
122            .backend
123            .hub
124            .interpreter
125            .cancel_tokens
126            .lock()
127            .remove(&id)
128        {
129            handle.abort();
130        }
131    }
132
133    pub fn subscribe_state(&self) -> RwStateSubscriber<S>
134    where
135        S: Clone + PartialEq,
136    {
137        let (tx, rx) = crossbeam_channel::unbounded();
138        self.backend.hub.state_listeners.lock().push(tx);
139        RwStateSubscriber {
140            rx,
141            state: self.backend.state.clone(),
142            last: Some(Arc::new(self.backend.state.with_read(|s| s.clone()))),
143        }
144    }
145
146    pub fn subscribe_changes(&self) -> RwChangeSubscriber<S>
147    where
148        S: FieldDiff + Hash,
149    {
150        let (tx, rx) = crossbeam_channel::unbounded();
151        self.backend.hub.state_listeners.lock().push(tx);
152        let (last_hash, last_fields) = self.backend.state.with_read(|guard| {
153            let root = hash_value(guard);
154            let mut fields = Vec::new();
155            guard.field_hashes(&mut fields);
156            (Some(root), fields)
157        });
158        RwChangeSubscriber {
159            rx,
160            state: self.backend.state.clone(),
161            last_hash,
162            last_fields,
163            _marker: PhantomData,
164        }
165    }
166
167    pub fn scope<CS, CM, AK, SK>(
168        &self,
169        state_kp: SK,
170        action_kp: AK,
171    ) -> ScopedRwStore<S, M, CS, CM, AK, SK>
172    where
173        CS: Clone + PartialEq + Send + Sync + 'static,
174        CM: Clone + Send + 'static,
175        S: 'static,
176        M: 'static,
177        AK: Casepath<M, CM> + Clone + Send + Sync + 'static,
178        SK: RefKpTrait<S, CS> + Clone,
179    {
180        ScopedRwStore {
181            store: self.clone(),
182            state_kp,
183            action_kp,
184            _marker: PhantomData,
185        }
186    }
187}
188
189/// Read-only store handle — many threads can [`read`](parking_lot::RwLock::read) concurrently.
190pub struct ReadStore<S, M> {
191    pub(crate) backend: RwStoreBackend<S, M>,
192}
193
194impl<S, M> Clone for ReadStore<S, M>
195where
196    S: Send + 'static,
197    M: Send + 'static,
198{
199    fn clone(&self) -> Self {
200        Self {
201            backend: self.backend.clone(),
202        }
203    }
204}
205
206impl<S, M> ReadStore<S, M>
207where
208    S: Send + Sync + Clone + 'static,
209    M: Send + 'static,
210{
211    pub fn with_read<R>(&self, f: impl FnOnce(&S) -> R) -> R {
212        self.backend.state.with_read(f)
213    }
214
215    pub fn state(&self) -> S {
216        self.with_read(|s| s.clone())
217    }
218
219    pub fn read_binding(&self) -> ReadStateBinding<S> {
220        ReadStateBinding::new(Arc::clone(&self.backend.state))
221    }
222
223    pub fn subscribe_state(&self) -> RwStateSubscriber<S>
224    where
225        S: Clone + PartialEq,
226    {
227        self.backend.rw_store().subscribe_state()
228    }
229
230    pub fn subscribe_changes(&self) -> RwChangeSubscriber<S>
231    where
232        S: FieldDiff + Hash,
233    {
234        self.backend.rw_store().subscribe_changes()
235    }
236}
237
238pub struct RwStateSubscriber<S> {
239    rx: Receiver<()>,
240    state: Arc<RwLock<S>>,
241    last: Option<Arc<S>>,
242}
243
244impl<S: PartialEq + Clone> RwStateSubscriber<S> {
245    #[allow(clippy::should_implement_trait)]
246    pub fn next(&mut self) -> Option<Arc<S>> {
247        loop {
248            match self.rx.try_recv() {
249                Ok(()) => {
250                    while self.rx.try_recv().is_ok() {}
251                    let snapshot = Arc::new(self.state.with_read(|s| s.clone()));
252                    if self
253                        .last
254                        .as_ref()
255                        .is_some_and(|prev| prev.as_ref() == snapshot.as_ref())
256                    {
257                        continue;
258                    }
259                    self.last = Some(snapshot.clone());
260                    return Some(snapshot);
261                }
262                Err(crossbeam_channel::TryRecvError::Empty) => return None,
263                Err(crossbeam_channel::TryRecvError::Disconnected) => return None,
264            }
265        }
266    }
267
268    pub fn wait_next(&mut self, timeout: Duration) -> Option<Arc<S>> {
269        let deadline = std::time::Instant::now() + timeout;
270        while std::time::Instant::now() < deadline {
271            if let Some(s) = self.next() {
272                return Some(s);
273            }
274            std::thread::sleep(Duration::from_millis(5));
275        }
276        None
277    }
278
279    pub fn latest(&self) -> Option<Arc<S>> {
280        self.last.clone()
281    }
282
283    pub fn read_binding(&self) -> ReadStateBinding<S> {
284        ReadStateBinding::new(Arc::clone(&self.state))
285    }
286}
287
288pub struct RwChangeSubscriber<S>
289where
290    S: FieldDiff + Hash,
291{
292    rx: Receiver<()>,
293    state: Arc<RwLock<S>>,
294    last_hash: Option<u64>,
295    last_fields: Vec<(S::Path, u64)>,
296    _marker: PhantomData<S>,
297}
298
299impl<S> RwChangeSubscriber<S>
300where
301    S: FieldDiff + Hash,
302{
303    fn snapshot_fields(&self) -> (u64, Vec<(S::Path, u64)>) {
304        self.state.with_read(|guard| {
305            let root = hash_value(guard);
306            let mut fields = Vec::new();
307            guard.field_hashes(&mut fields);
308            (root, fields)
309        })
310    }
311
312    fn diff_fields(&self, fields: &[(S::Path, u64)]) -> Vec<S::Path> {
313        fields
314            .iter()
315            .filter(|(path, hash)| {
316                self.last_fields
317                    .iter()
318                    .find(|(prev_path, _)| prev_path == path)
319                    .is_none_or(|(_, prev_hash)| prev_hash != hash)
320            })
321            .map(|(path, _)| *path)
322            .collect()
323    }
324
325    pub fn read_binding(&self) -> ReadStateBinding<S> {
326        ReadStateBinding::new(Arc::clone(&self.state))
327    }
328
329    #[allow(clippy::should_implement_trait)]
330    pub fn next(&mut self) -> Option<ChangeSet<S::Path>> {
331        loop {
332            match self.rx.try_recv() {
333                Ok(()) => {
334                    while self.rx.try_recv().is_ok() {}
335                    let (root_hash, fields) = self.snapshot_fields();
336                    if Some(root_hash) == self.last_hash {
337                        continue;
338                    }
339                    let paths = self.diff_fields(&fields);
340                    self.last_hash = Some(root_hash);
341                    self.last_fields = fields;
342                    if paths.is_empty() {
343                        continue;
344                    }
345                    return Some(ChangeSet { paths });
346                }
347                Err(crossbeam_channel::TryRecvError::Empty) => return None,
348                Err(crossbeam_channel::TryRecvError::Disconnected) => return None,
349            }
350        }
351    }
352
353    pub fn wait_next(&mut self, timeout: Duration) -> Option<ChangeSet<S::Path>> {
354        let deadline = std::time::Instant::now() + timeout;
355        while std::time::Instant::now() < deadline {
356            if let Some(change) = self.next() {
357                return Some(change);
358            }
359            std::thread::sleep(Duration::from_millis(5));
360        }
361        None
362    }
363}
364
365/// Child [`RwStore`] routing actions through a parent action casepath.
366pub struct ScopedRwStore<S: 'static, M: 'static, CS: 'static, CM: 'static, AK: 'static, SK>
367where
368    SK: RefKpTrait<S, CS> + Clone,
369{
370    store: RwStore<S, M>,
371    state_kp: SK,
372    action_kp: AK,
373    _marker: PhantomData<(M, CM, S, CS)>,
374}
375
376impl<S: 'static, M: 'static, CS: 'static, CM: 'static, AK: 'static, SK> Clone
377    for ScopedRwStore<S, M, CS, CM, AK, SK>
378where
379    S: Send,
380    M: Send,
381    AK: Clone,
382    SK: RefKpTrait<S, CS> + Clone,
383{
384    fn clone(&self) -> Self {
385        Self {
386            store: self.store.clone(),
387            state_kp: self.state_kp.clone(),
388            action_kp: self.action_kp.clone(),
389            _marker: PhantomData,
390        }
391    }
392}
393
394impl<S: 'static, M: 'static, CS: 'static, CM: 'static, AK: 'static, SK>
395    ScopedRwStore<S, M, CS, CM, AK, SK>
396where
397    S: Send + Sync + Clone,
398    M: Send,
399    CS: Clone + PartialEq + Send + Sync,
400    CM: Clone + Send,
401    AK: Casepath<M, CM> + Clone + Send + Sync + 'static,
402    SK: RefKpTrait<S, CS> + Clone,
403{
404    pub fn send(&self, action: CM) -> StoreTask {
405        self.store.send(self.action_kp.wrap(action))
406    }
407
408    pub fn dispatch(&self, action: CM) {
409        self.store.dispatch(self.action_kp.wrap(action));
410    }
411
412    pub fn child_state(&self) -> Option<CS> {
413        self.state_kp.focus(&self.store.state()).cloned()
414    }
415
416    pub fn read_binding(&self) -> crate::runtime::binding::ReadProjectedBinding<S, CS, SK> {
417        self.store
418            .read_store()
419            .read_binding()
420            .project(self.state_kp.clone())
421    }
422
423    pub fn rw_binding(&self) -> crate::runtime::binding::RwProjectedBinding<S, CS, SK> {
424        self.store.binding().project(self.state_kp.clone())
425    }
426}
rust_elm/runtime/rw_store.rs

rust_elm/runtime/
rw_store.rs
