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(mut self) {
234        super::lifecycle::join_reducer_thread(&self.shutdown, &self._subscriptions, &mut self._thread);
235    }
236}
237
238impl<S, M> Drop for TeaRuntime<S, M> {
239    fn drop(&mut self) {
240        super::lifecycle::join_reducer_thread(&self.shutdown, &self._subscriptions, &mut self._thread);
241    }
242}
243
244enum LoopMsg<S, M> {
245    Action(M),
246    Control(TeaControl<S>),
247}
248
249fn recv_action_or_control<S, M>(
250    action_rx: &Receiver<M>,
251    control_rx: &Receiver<TeaControl<S>>,
252    timeout: Duration,
253) -> Option<LoopMsg<S, M>>
254where
255    S: Send + 'static,
256    M: Send + 'static,
257{
258    crossbeam_channel::select! {
259        recv(action_rx) -> msg => msg.ok().map(LoopMsg::Action),
260        recv(control_rx) -> ctrl => ctrl.ok().map(LoopMsg::Control),
261        default(timeout) => None,
262    }
263}
264
265fn publish_snapshots<S>(state: &S, subscribers: &mut Vec<Sender<Arc<S>>>)
266where
267    S: Clone,
268{
269    let snap = Arc::new(state.clone());
270    subscribers.retain(|tx| tx.send(Arc::clone(&snap)).is_ok());
271}
272
273#[cfg(test)]
274mod tests {
275    use super::*;
276
277    #[derive(Default, Clone, PartialEq)]
278    struct Counter {
279        n: i32,
280    }
281
282    fn init() -> (Counter, Cmd<i32>) {
283        (Counter::default(), Cmd::none())
284    }
285
286    fn update(s: &mut Counter, msg: i32) -> Cmd<i32> {
287        s.n += msg;
288        Cmd::none()
289    }
290
291    fn subs(_: &Counter) -> Sub<i32> {
292        Sub::none()
293    }
294
295    fn boot(program: Program<Counter, i32>) -> TeaRuntime<Counter, i32> {
296        match TeaRuntime::from_program(program, Environment::new(), RuntimeConfig::new(16)) {
297            Ok(runtime) => runtime,
298            Err(err) => panic!("test runtime bootstrap failed: {err}"),
299        }
300    }
301
302    #[test]
303    fn tea_runtime_dispatches_and_pushes_snapshots() {
304        let runtime = boot(Program::new(init, update, subs));
305        let store = runtime.tea_store();
306        let Ok(mut sub) = store.subscribe_state() else {
307            panic!("expected subscribe_state to succeed in test");
308        };
309        store.dispatch(3);
310        store.dispatch(4);
311        std::thread::sleep(Duration::from_millis(200));
312        while sub.next().is_some() {}
313        assert_eq!(sub.latest().map(|s| s.n), Some(7));
314    }
315
316    #[test]
317    fn tea_view_store_requests_snapshot() {
318        let runtime = boot(Program::new(init, update, subs));
319        let store = runtime.tea_store();
320        store.dispatch(5);
321        std::thread::sleep(Duration::from_millis(100));
322        let n = store
323            .view_store()
324            .try_with_snapshot(|s| s.n)
325            .map_err(|e| format!("{e}"))
326            .ok();
327        assert_eq!(n, Some(5));
328        runtime.shutdown();
329    }
330}
rust_elm/runtime/tea_engine.rs

rust_elm/runtime/
tea_engine.rs
