1use std::sync::atomic::{AtomicBool, Ordering};
2use std::sync::Arc;
3use std::thread::{self, JoinHandle};
4use std::time::Duration;
5
6use crossbeam_channel::RecvTimeoutError;
7use parking_lot::RwLock;
8use tokio::runtime::{Builder as TokioRuntimeBuilder, Runtime as TokioRuntime};
9
10use crate::bus::{Bus, BusSender};
11use crate::cmd::Cmd;
12use crate::effect::EffectId;
13use crate::env::Environment;
14use crate::program::{Program, ReducerProgram};
15use crate::reducer::Reducer;
16use crate::safe_reducer::safe_reduce_update;
17use crate::sub::Sub;
18
19use super::config::RuntimeConfig;
20use super::interpreter::{interpret_effects_async, InterpreterState};
21use super::rw_store::{RwStore, RwStoreBackend};
22use super::store::{StoreHub, StoreWork, StoreWorkUnwindGuard};
23use super::subscription;
24
25pub struct RwRuntime<S, M> {
27 pub state: Arc<RwLock<S>>,
28 pub bus: Bus<M>,
29 pub env: Environment,
30 backend: RwStoreBackend<S, M>,
31 shutdown: Arc<AtomicBool>,
32 _thread: Option<JoinHandle<()>>,
33 _tokio: Arc<TokioRuntime>,
34 _subscriptions: Arc<subscription::SubscriptionHandles>,
35}
36
37impl<S, M> RwRuntime<S, M>
38where
39 S: Send + Sync + 'static,
40 M: Send + 'static,
41{
42 pub fn from_program(program: Program<S, M>, env: Environment, config: RuntimeConfig) -> Self {
43 Self::bootstrap(
44 program.init,
45 program.update,
46 program.subscriptions,
47 env,
48 config,
49 )
50 }
51
52 pub fn from_reducer_program<R>(
53 program: ReducerProgram<R>,
54 env: Environment,
55 config: RuntimeConfig,
56 ) -> Self
57 where
58 R: Reducer<State = S, Action = M> + Send + Sync + 'static,
59 {
60 let reducer = Arc::new(program.reducer);
61 let init = program.init;
62 let subscriptions = program.subscriptions;
63 Self::bootstrap(
64 init,
65 move |state, msg| reducer.reduce(state, msg),
66 subscriptions,
67 env,
68 config,
69 )
70 }
71
72 fn bootstrap(
73 init: fn() -> (S, Cmd<M>),
74 update: impl Fn(&mut S, M) -> Cmd<M> + Send + Sync + 'static,
75 subscriptions: fn(&S) -> Sub<M>,
76 env: Environment,
77 config: RuntimeConfig,
78 ) -> Self {
79 let RuntimeConfig {
80 bus_capacity,
81 worker_threads,
82 thread_name,
83 } = config;
84 let (initial_state, init_cmd) = init();
85 let state = Arc::new(RwLock::new(initial_state));
86 let bus = Bus::new(bus_capacity);
87 let shutdown = Arc::new(AtomicBool::new(false));
88 let tokio = Arc::new(
89 TokioRuntimeBuilder::new_multi_thread()
90 .worker_threads(worker_threads.max(1))
91 .thread_name(thread_name)
92 .enable_all()
93 .build()
94 .expect("failed to create Tokio runtime for rust-elm"),
95 );
96 let interpreter = InterpreterState::new();
97 let sub_handles = Arc::new(subscription::SubscriptionHandles::new());
98 let hub = StoreHub::new(bus.sender(), interpreter);
99 let backend = RwStoreBackend::new(state.clone(), hub);
100 let tx = bus.sender();
101 let init_handles = tokio.block_on(interpret_effects_async(
102 init_cmd.into_effects(),
103 tx.clone(),
104 env.clone(),
105 tokio.handle().clone(),
106 backend.clone(),
107 ));
108 tokio.handle().spawn(async move {
109 for join in init_handles {
110 let _ = join.await;
111 }
112 });
113
114 let receiver = bus.receiver().clone();
115 let subs_fn = subscriptions;
116 let state_for_thread = state.clone();
117 let env_for_thread = env.clone();
118 let shutdown_for_thread = shutdown.clone();
119 let tokio_for_thread = tokio.clone();
120 let backend_for_thread = backend.clone();
121 let tx_for_thread = tx;
122 let subs_registry = sub_handles.clone();
123
124 let thread = thread::Builder::new()
125 .name(thread_name.to_string())
126 .spawn(move || {
127 let rt = tokio_for_thread;
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_for_thread.read());
141
142 while !shutdown_for_thread.load(Ordering::Relaxed) {
143 match receiver.recv_timeout(Duration::from_millis(50)) {
144 Ok(msg) => {
145 let mut unwind = StoreWorkUnwindGuard::new(&backend_for_thread);
146 let cmd = {
147 let mut guard = state_for_thread.write();
148 match safe_reduce_update(&mut *guard, |s, a| update(s, a), msg) {
149 Ok(cmd) => cmd,
150 Err(_) => {
151 drop(guard);
152 backend_for_thread.notify_state();
153 continue;
154 }
155 }
156 };
157 unwind.disarm();
158 backend_for_thread.notify_state();
159 let handles = rt.block_on(interpret_effects_async(
160 cmd.into_effects(),
161 tx_for_thread.clone(),
162 env_for_thread.clone(),
163 rt.handle().clone(),
164 backend_for_thread.clone(),
165 ));
166 if handles.is_empty() {
167 backend_for_thread.end_store_work();
168 } else {
169 let backend_wait = backend_for_thread.clone();
170 rt.handle().spawn(async move {
171 for join in handles {
172 let _ = join.await;
173 }
174 backend_wait.end_store_work();
175 });
176 }
177
178 sync_subs(&state_for_thread.read());
179 }
180 Err(RecvTimeoutError::Timeout) => continue,
181 Err(RecvTimeoutError::Disconnected) => break,
182 }
183 }
184 })
185 .expect("failed to spawn rust-elm reducer thread");
186
187 Self {
188 state,
189 bus,
190 env,
191 backend,
192 shutdown,
193 _thread: Some(thread),
194 _tokio: tokio,
195 _subscriptions: sub_handles,
196 }
197 }
198
199 pub fn rw_store(&self) -> RwStore<S, M>
200 where
201 S: Clone + Send + Sync + 'static,
202 {
203 self.backend.rw_store()
204 }
205
206 pub fn dispatch(&self, msg: M) {
207 let _ = self.bus.sender().send_blocking(msg);
208 }
209
210 pub fn sender(&self) -> BusSender<M> {
211 self.bus.sender()
212 }
213
214 pub fn cancel(&self, id: EffectId) {
215 if let Some(handle) = self.backend.hub.interpreter.cancel_tokens.lock().remove(&id) {
216 handle.abort();
217 }
218 }
219
220 pub fn shutdown(self) {
221 self.shutdown.store(true, Ordering::Relaxed);
222 self._subscriptions.abort_all();
223 if let Some(handle) = self._thread {
224 let _ = handle.join();
225 }
226 }
227}
228
229#[cfg(test)]
230mod tests {
231 use super::*;
232 use crate::panic_on_state_clone;
233
234 panic_on_state_clone! {
235 #[derive(Default)]
236 struct Counter {
237 n: i32,
238 }
239 }
240
241 fn init() -> (Counter, Cmd<i32>) {
242 (Counter::default(), Cmd::none())
243 }
244
245 fn update(s: &mut Counter, msg: i32) -> Cmd<i32> {
246 s.n += msg;
247 Cmd::none()
248 }
249
250 fn subs(_: &Counter) -> Sub<i32> {
251 Sub::none()
252 }
253
254 #[test]
255 fn rw_runtime_applies_dispatched_messages() {
256 let program = Program::new(init, update, subs);
257 let runtime = RwRuntime::from_program(program, Environment::new(), RuntimeConfig::new(16));
258 runtime.dispatch(3);
259 runtime.dispatch(4);
260 std::thread::sleep(Duration::from_millis(200));
261 assert_eq!(runtime.state.read().n, 7);
262 runtime.shutdown();
263 }
264
265 #[test]
266 fn read_store_allows_concurrent_reads() {
267 let program = Program::new(init, update, subs);
268 let runtime = RwRuntime::from_program(program, Environment::new(), RuntimeConfig::new(16));
269 let store = runtime.rw_store();
270 store.dispatch(5);
271 std::thread::sleep(Duration::from_millis(100));
272 assert_eq!(store.read_store().with_read(|s| s.n), 5);
273 runtime.shutdown();
274 }
275}