rust_elm/runtime/
tea_engine.rs

1use std::sync::atomic::{AtomicBool, Ordering};
2use std::sync::Arc;
3use std::thread::{self, JoinHandle};
4use std::time::Duration;
5
6use crossbeam_channel::{Receiver, Sender};
7use tokio::runtime::Runtime as TokioRuntime;
8
9use crate::bus::{Bus, BusSender};
10use crate::cmd::Cmd;
11use crate::effect::EffectId;
12use crate::env::Environment;
13use crate::program::{Program, ReducerProgram};
14use crate::reducer::Reducer;
15use crate::safe_reducer::safe_reduce_update;
16use crate::sub::Sub;
17
18use super::config::RuntimeConfig;
19use super::error::{build_tokio, RuntimeError};
20use super::interpreter::{interpret_effects_async, InterpreterState};
21use super::store::{StoreHub, StoreWork, StoreWorkUnwindGuard};
22use super::subscription;
23use super::tea_store::{TeaControl, TeaStore, TeaStoreBackend};
24
25/// Live runtime with **channel-delivered model** — true Elm Architecture.
26///
27/// `S` is owned exclusively by the reducer thread. After each `update`, the runtime
28/// pushes `Arc<S>` to subscribers. Readers never lock shared state.
29pub struct TeaRuntime<S, M> {
30    pub bus: Bus<M>,
31    pub env: Environment,
32    backend: TeaStoreBackend<S, M>,
33    shutdown: Arc<AtomicBool>,
34    _thread: Option<JoinHandle<()>>,
35    _tokio: Arc<TokioRuntime>,
36    _subscriptions: Arc<subscription::SubscriptionHandles>,
37}
38
39impl<S, M> TeaRuntime<S, M>
40where
41    S: Send + Sync + Clone + 'static,
42    M: Send + 'static,
43{
44    pub fn from_program(
45        program: Program<S, M>,
46        env: Environment,
47        config: RuntimeConfig,
48    ) -> Result<Self, RuntimeError> {
49        Self::bootstrap(
50            program.init,
51            program.update,
52            program.subscriptions,
53            env,
54            config,
55        )
56    }
57
58    pub fn from_reducer_program<R>(
59        program: ReducerProgram<R>,
60        env: Environment,
61        config: RuntimeConfig,
62    ) -> Result<Self, RuntimeError>
63    where
64        R: Reducer<State = S, Action = M> + Send + Sync + 'static,
65    {
66        let reducer = Arc::new(program.reducer);
67        let init = program.init;
68        let subscriptions = program.subscriptions;
69        Self::bootstrap(
70            init,
71            move |state, msg| reducer.reduce(state, msg),
72            subscriptions,
73            env,
74            config,
75        )
76    }
77
78    fn bootstrap(
79        init: fn() -> (S, Cmd<M>),
80        update: impl Fn(&mut S, M) -> Cmd<M> + Send + Sync + 'static,
81        subscriptions: fn(&S) -> Sub<M>,
82        env: Environment,
83        config: RuntimeConfig,
84    ) -> Result<Self, RuntimeError> {
85        let RuntimeConfig {
86            bus_capacity,
87            worker_threads,
88            thread_name,
89        } = config;
90        let (mut state, init_cmd) = init();
91        let bus = Bus::new(bus_capacity);
92        let shutdown = Arc::new(AtomicBool::new(false));
93        let tokio = build_tokio(worker_threads, thread_name)?;
94        let interpreter = InterpreterState::new();
95        let sub_handles = Arc::new(subscription::SubscriptionHandles::new());
96        let hub = StoreHub::new(bus.sender(), interpreter);
97        let (control_tx, control_rx) = crossbeam_channel::unbounded();
98        let backend = TeaStoreBackend::new(hub, control_tx);
99        let tx = bus.sender();
100        let init_handles = tokio.block_on(interpret_effects_async(
101            init_cmd.into_effects(),
102            tx.clone(),
103            env.clone(),
104            tokio.handle().clone(),
105            backend.clone(),
106        ));
107        tokio.handle().spawn(async move {
108            for join in init_handles {
109                let _ = join.await;
110            }
111        });
112
113        let action_rx = bus.receiver().clone();
114        let subs_fn = subscriptions;
115        let env_for_thread = env.clone();
116        let shutdown_for_thread = shutdown.clone();
117        let tokio_for_thread = tokio.clone();
118        let backend_for_thread = backend.clone();
119        let tx_for_thread = tx;
120        let subs_registry = sub_handles.clone();
121
122        let thread = thread::Builder::new()
123            .name(thread_name.to_string())
124            .spawn(move || {
125                let rt = tokio_for_thread;
126                let mut snapshot_subscribers: Vec<Sender<Arc<S>>> = Vec::new();
127
128                let sync_subs = |state: &S| {
129                    let sub = subs_fn(state);
130                    subscription::sync_subscriptions(
131                        &sub,
132                        &subs_registry,
133                        rt.handle().clone(),
134                        tx_for_thread.clone(),
135                        backend_for_thread.clone(),
136                        shutdown_for_thread.clone(),
137                    );
138                };
139
140                sync_subs(&state);
141
142                while !shutdown_for_thread.load(Ordering::Relaxed) {
143                    if let Some(msg) = recv_action_or_control(
144                        &action_rx,
145                        &control_rx,
146                        Duration::from_millis(50),
147                    ) {
148                        match msg {
149                            LoopMsg::Action(msg) => {
150                                let mut unwind =
151                                    StoreWorkUnwindGuard::new(&backend_for_thread);
152                                let cmd = match safe_reduce_update(
153                                    &mut state,
154                                    |s, a| update(s, a),
155                                    msg,
156                                ) {
157                                    Ok(cmd) => cmd,
158                                    Err(_) => {
159                                        publish_snapshots(&state, &mut snapshot_subscribers);
160                                        continue;
161                                    }
162                                };
163                                publish_snapshots(&state, &mut snapshot_subscribers);
164                                unwind.disarm();
165                                let handles = rt.block_on(interpret_effects_async(
166                                    cmd.into_effects(),
167                                    tx_for_thread.clone(),
168                                    env_for_thread.clone(),
169                                    rt.handle().clone(),
170                                    backend_for_thread.clone(),
171                                ));
172                                if handles.is_empty() {
173                                    backend_for_thread.end_store_work();
174                                } else {
175                                    let backend_wait = backend_for_thread.clone();
176                                    rt.handle().spawn(async move {
177                                        for join in handles {
178                                            let _ = join.await;
179                                        }
180                                        backend_wait.end_store_work();
181                                    });
182                                }
183
184                                sync_subs(&state);
185                            }
186                            LoopMsg::Control(TeaControl::Subscribe {
187                                snapshot_tx,
188                                ready_tx,
189                            }) => {
190                                let snap = Arc::new(state.clone());
191                                let _ = snapshot_tx.send(snap);
192                                snapshot_subscribers.push(snapshot_tx);
193                                let _ = ready_tx.send(());
194                            }
195                            LoopMsg::Control(TeaControl::RequestSnapshot { reply }) => {
196                                let _ = reply.send(Arc::new(state.clone()));
197                            }
198                        }
199                    }
200                }
201            })
202            .map_err(RuntimeError::ThreadSpawn)?;
203
204        Ok(Self {
205            bus,
206            env,
207            backend,
208            shutdown,
209            _thread: Some(thread),
210            _tokio: tokio,
211            _subscriptions: sub_handles,
212        })
213    }
214
215    pub fn tea_store(&self) -> TeaStore<S, M> {
216        self.backend.tea_store()
217    }
218
219    pub fn dispatch(&self, msg: M) {
220        let _ = self.bus.sender().send_blocking(msg);
221    }
222
223    pub fn sender(&self) -> BusSender<M> {
224        self.bus.sender()
225    }
226
227    pub fn cancel(&self, id: EffectId) {
228        if let Some(handle) = self.backend.hub.interpreter.cancel_tokens.lock().remove(&id) {
229            handle.abort();
230        }
231    }
232
233    pub fn shutdown(self) {
234        self.shutdown.store(true, Ordering::Relaxed);
235        self._subscriptions.abort_all();
236        if let Some(handle) = self._thread {
237            let _ = handle.join();
238        }
239    }
240}
241
242enum LoopMsg<S, M> {
243    Action(M),
244    Control(TeaControl<S>),
245}
246
247fn recv_action_or_control<S, M>(
248    action_rx: &Receiver<M>,
249    control_rx: &Receiver<TeaControl<S>>,
250    timeout: Duration,
251) -> Option<LoopMsg<S, M>>
252where
253    S: Send + 'static,
254    M: Send + 'static,
255{
256    crossbeam_channel::select! {
257        recv(action_rx) -> msg => msg.ok().map(LoopMsg::Action),
258        recv(control_rx) -> ctrl => ctrl.ok().map(LoopMsg::Control),
259        default(timeout) => None,
260    }
261}
262
263fn publish_snapshots<S>(state: &S, subscribers: &mut Vec<Sender<Arc<S>>>)
264where
265    S: Clone,
266{
267    let snap = Arc::new(state.clone());
268    subscribers.retain(|tx| tx.send(Arc::clone(&snap)).is_ok());
269}
270
271#[cfg(test)]
272mod tests {
273    use super::*;
274
275    #[derive(Default, Clone, PartialEq)]
276    struct Counter {
277        n: i32,
278    }
279
280    fn init() -> (Counter, Cmd<i32>) {
281        (Counter::default(), Cmd::none())
282    }
283
284    fn update(s: &mut Counter, msg: i32) -> Cmd<i32> {
285        s.n += msg;
286        Cmd::none()
287    }
288
289    fn subs(_: &Counter) -> Sub<i32> {
290        Sub::none()
291    }
292
293    fn boot(program: Program<Counter, i32>) -> TeaRuntime<Counter, i32> {
294        match TeaRuntime::from_program(program, Environment::new(), RuntimeConfig::new(16)) {
295            Ok(runtime) => runtime,
296            Err(err) => panic!("test runtime bootstrap failed: {err}"),
297        }
298    }
299
300    #[test]
301    fn tea_runtime_dispatches_and_pushes_snapshots() {
302        let store = boot(Program::new(init, update, subs)).tea_store();
303        let Ok(mut sub) = store.subscribe_state() else {
304            panic!("expected subscribe_state to succeed in test");
305        };
306        store.dispatch(3);
307        store.dispatch(4);
308        std::thread::sleep(Duration::from_millis(200));
309        while sub.next().is_some() {}
310        assert_eq!(sub.latest().map(|s| s.n), Some(7));
311        boot(Program::new(init, update, subs)).shutdown();
312    }
313
314    #[test]
315    fn tea_view_store_requests_snapshot() {
316        let runtime = boot(Program::new(init, update, subs));
317        let store = runtime.tea_store();
318        store.dispatch(5);
319        std::thread::sleep(Duration::from_millis(100));
320        let n = store
321            .view_store()
322            .try_with_snapshot(|s| s.n)
323            .map_err(|e| format!("{e}"))
324            .ok();
325        assert_eq!(n, Some(5));
326        runtime.shutdown();
327    }
328}
rust_elm/runtime/tea_engine.rs

rust_elm/runtime/
tea_engine.rs
