cranpose_core/
launched_effect.rs

1#[cfg(not(target_arch = "wasm32"))]
2use std::cell::{Cell, RefCell};
3#[cfg(not(target_arch = "wasm32"))]
4use std::rc::Rc;
5use std::{
6    future::Future,
7    pin::Pin,
8    sync::{
9        Arc,
10        atomic::{AtomicBool, Ordering},
11    },
12};
13
14use crate::{Key, RuntimeHandle, TaskHandle, effect_key::EffectKey, with_current_composer};
15
16trait EffectKeySlot {
17    fn key(&self) -> &Option<EffectKey>;
18    fn key_mut(&mut self) -> &mut Option<EffectKey>;
19
20    fn should_run(&self, key: &EffectKey) -> bool {
21        match self.key() {
22            Some(current) => key.differs_from(current),
23            None => true,
24        }
25    }
26
27    fn set_key(&mut self, key: EffectKey) {
28        *self.key_mut() = Some(key);
29    }
30}
31
32#[derive(Default)]
33struct LaunchedEffectState {
34    key: Option<EffectKey>,
35    cancel: Option<LaunchedEffectCancellation>,
36}
37
38struct LaunchedEffectCancellation {
39    #[cfg(not(target_arch = "wasm32"))]
40    runtime: RuntimeHandle,
41    active: Arc<AtomicBool>,
42    #[cfg(not(target_arch = "wasm32"))]
43    continuations: Rc<RefCell<Vec<u64>>>,
44}
45
46/// Where an effect was written, carried onto the task it spawns.
47///
48/// A leak report that can only say "one task is still queued" names nothing:
49/// every run prints the same number and no run says which effect it was. This
50/// is what turns that count into a place to look, and it stays a pair of plain
51/// fields so carrying it costs nothing until a task is actually spawned.
52#[derive(Clone, Copy, Debug, PartialEq, Eq)]
53pub struct TaskSite {
54    pub file: &'static str,
55    pub line: u32,
56}
57
58impl TaskSite {
59    /// The site a macro captured with `file!()` and `line!()`.
60    pub const fn new(file: &'static str, line: u32) -> TaskSite {
61        TaskSite { file, line }
62    }
63}
64
65impl Default for TaskSite {
66    fn default() -> TaskSite {
67        TaskSite::new("unknown", 0)
68    }
69}
70
71impl std::fmt::Display for TaskSite {
72    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
73        write!(formatter, "{}:{}", self.file, self.line)
74    }
75}
76
77impl From<&'static std::panic::Location<'static>> for TaskSite {
78    fn from(location: &'static std::panic::Location<'static>) -> TaskSite {
79        TaskSite::new(location.file(), location.line())
80    }
81}
82
83#[derive(Default)]
84struct LaunchedEffectAsyncState {
85    key: Option<EffectKey>,
86    cancel: Option<LaunchedEffectCancellation>,
87    task: Option<TaskHandle>,
88    site: TaskSite,
89}
90
91impl EffectKeySlot for LaunchedEffectState {
92    fn key(&self) -> &Option<EffectKey> {
93        &self.key
94    }
95
96    fn key_mut(&mut self) -> &mut Option<EffectKey> {
97        &mut self.key
98    }
99}
100
101impl LaunchedEffectState {
102    fn launch(
103        &mut self,
104        runtime: RuntimeHandle,
105        effect: impl FnOnce(LaunchedEffectScope) + 'static,
106    ) {
107        self.cancel_current();
108        let active = Arc::new(AtomicBool::new(true));
109        #[cfg(not(target_arch = "wasm32"))]
110        let continuations = Rc::new(RefCell::new(Vec::new()));
111        self.cancel = Some(LaunchedEffectCancellation {
112            #[cfg(not(target_arch = "wasm32"))]
113            runtime: runtime.clone(),
114            active: Arc::clone(&active),
115            #[cfg(not(target_arch = "wasm32"))]
116            continuations: Rc::clone(&continuations),
117        });
118        let scope = LaunchedEffectScope {
119            active: Arc::clone(&active),
120            runtime: runtime.clone(),
121            #[cfg(not(target_arch = "wasm32"))]
122            continuations,
123        };
124        runtime.enqueue_ui_task(Box::new(move || effect(scope)));
125    }
126
127    fn cancel_current(&mut self) {
128        if let Some(cancel) = self.cancel.take() {
129            cancel.cancel();
130        }
131    }
132}
133
134impl LaunchedEffectCancellation {
135    fn cancel(&self) {
136        self.active.store(false, Ordering::SeqCst);
137        #[cfg(not(target_arch = "wasm32"))]
138        {
139            let mut pending = self.continuations.borrow_mut();
140            for id in pending.drain(..) {
141                self.runtime.cancel_ui_cont(id);
142            }
143        }
144    }
145}
146
147impl EffectKeySlot for LaunchedEffectAsyncState {
148    fn key(&self) -> &Option<EffectKey> {
149        &self.key
150    }
151
152    fn key_mut(&mut self) -> &mut Option<EffectKey> {
153        &mut self.key
154    }
155}
156
157impl LaunchedEffectAsyncState {
158    fn set_site(&mut self, site: TaskSite) {
159        self.site = site;
160    }
161
162    fn launch(
163        &mut self,
164        runtime: RuntimeHandle,
165        mk_future: impl FnOnce(LaunchedEffectScope) -> Pin<Box<dyn Future<Output = ()>>> + 'static,
166    ) {
167        self.cancel_current();
168        let active = Arc::new(AtomicBool::new(true));
169        #[cfg(not(target_arch = "wasm32"))]
170        let continuations = Rc::new(RefCell::new(Vec::new()));
171        self.cancel = Some(LaunchedEffectCancellation {
172            #[cfg(not(target_arch = "wasm32"))]
173            runtime: runtime.clone(),
174            active: Arc::clone(&active),
175            #[cfg(not(target_arch = "wasm32"))]
176            continuations: Rc::clone(&continuations),
177        });
178        let scope = LaunchedEffectScope {
179            active: Arc::clone(&active),
180            runtime: runtime.clone(),
181            #[cfg(not(target_arch = "wasm32"))]
182            continuations,
183        };
184        let future = mk_future(scope.clone());
185        let active_flag = Arc::clone(&scope.active);
186        crate::label_next_ui_task(self.site.to_string());
187        match runtime.spawn_ui(async move {
188            future.await;
189            active_flag.store(false, Ordering::SeqCst);
190        }) {
191            Some(handle) => {
192                self.task = Some(handle);
193            }
194            None => {
195                active.store(false, Ordering::SeqCst);
196                self.cancel = None;
197            }
198        }
199    }
200
201    fn cancel_current(&mut self) {
202        if let Some(handle) = self.task.take() {
203            handle.cancel();
204        }
205        if let Some(cancel) = self.cancel.take() {
206            cancel.cancel();
207        }
208    }
209}
210
211impl Drop for LaunchedEffectState {
212    fn drop(&mut self) {
213        self.cancel_current();
214    }
215}
216
217impl Drop for LaunchedEffectAsyncState {
218    fn drop(&mut self) {
219        self.cancel_current();
220    }
221}
222
223#[derive(Clone)]
224pub struct LaunchedEffectScope {
225    active: Arc<AtomicBool>,
226    runtime: RuntimeHandle,
227    #[cfg(not(target_arch = "wasm32"))]
228    continuations: Rc<RefCell<Vec<u64>>>,
229}
230
231impl LaunchedEffectScope {
232    #[cfg(not(target_arch = "wasm32"))]
233    fn track_continuation(&self, id: u64) {
234        self.continuations.borrow_mut().push(id);
235    }
236
237    #[cfg(not(target_arch = "wasm32"))]
238    fn release_continuation(&self, id: u64) {
239        let mut continuations = self.continuations.borrow_mut();
240        if let Some(index) = continuations.iter().position(|entry| *entry == id) {
241            continuations.remove(index);
242        }
243    }
244
245    pub fn is_active(&self) -> bool {
246        self.active.load(Ordering::SeqCst)
247    }
248
249    pub fn runtime(&self) -> RuntimeHandle {
250        self.runtime.clone()
251    }
252
253    /// Runs a follow-up `LaunchedEffect` task on the UI thread.
254    ///
255    /// The provided closure executes on the runtime thread and may freely
256    /// capture `Rc`/`RefCell` state. This must only be called from the UI
257    /// thread, typically inside another effect callback.
258    pub fn launch(&self, task: impl FnOnce(LaunchedEffectScope) + 'static) {
259        if !self.is_active() {
260            return;
261        }
262        let scope = self.clone();
263        self.runtime.enqueue_ui_task(Box::new(move || {
264            if scope.is_active() {
265                task(scope);
266            }
267        }));
268    }
269
270    /// Posts UI-only work that will execute on the runtime thread.
271    ///
272    /// The closure never crosses threads, so it may capture non-`Send` values.
273    /// Callers must invoke this from the UI thread.
274    pub fn post_ui(&self, task: impl FnOnce() + 'static) {
275        if !self.is_active() {
276            return;
277        }
278        let active = Arc::clone(&self.active);
279        self.runtime.enqueue_ui_task(Box::new(move || {
280            if active.load(Ordering::SeqCst) {
281                task();
282            }
283        }));
284    }
285
286    /// Runs background work and delivers results to the UI.
287    ///
288    /// On native targets, `work` runs on a worker thread and its future is
289    /// driven to completion there. On WASM, `work` runs as a task on the
290    /// browser event loop. The `on_ui` continuation always runs on the runtime
291    /// thread, so it may capture `Rc`/`RefCell` state safely.
292    #[cfg(not(target_arch = "wasm32"))]
293    pub fn launch_background<T, Work, Ui, Fut>(&self, work: Work, on_ui: Ui)
294    where
295        T: Send + 'static,
296        Work: FnOnce(CancelToken) -> Fut + Send + 'static,
297        Fut: Future<Output = T> + Send + 'static,
298        Ui: FnOnce(T) + 'static,
299    {
300        if !self.is_active() {
301            return;
302        }
303        let dispatcher = self.runtime.dispatcher();
304        let active_for_thread = Arc::clone(&self.active);
305        let continuation_scope = self.clone();
306        let continuation_active = Arc::clone(&self.active);
307        let id_cell = Rc::new(Cell::new(0));
308        let id_for_closure = Rc::clone(&id_cell);
309        let continuation = move |value: T| {
310            let id = id_for_closure.get();
311            continuation_scope.release_continuation(id);
312            if continuation_active.load(Ordering::SeqCst) {
313                on_ui(value);
314            }
315        };
316
317        let Some(cont_id) = self.runtime.register_ui_cont(continuation) else {
318            return;
319        };
320        id_cell.set(cont_id);
321        self.track_continuation(cont_id);
322
323        std::thread::spawn(move || {
324            let token = CancelToken::new(Arc::clone(&active_for_thread));
325            let value = pollster::block_on(work(token.clone()));
326            if token.is_cancelled() {
327                return;
328            }
329            dispatcher.post_invoke(cont_id, value);
330        });
331    }
332
333    /// Runs background work and delivers results to the UI.
334    ///
335    /// On native targets, `work` runs on a worker thread and its future is
336    /// driven to completion there. On WASM, `work` runs as a task on the
337    /// browser event loop. The `on_ui` continuation always runs on the runtime
338    /// thread, so it may capture `Rc`/`RefCell` state safely.
339    #[cfg(target_arch = "wasm32")]
340    pub fn launch_background<T, Work, Ui, Fut>(&self, work: Work, on_ui: Ui)
341    where
342        T: 'static,
343        Work: FnOnce(CancelToken) -> Fut + 'static,
344        Fut: Future<Output = T> + 'static,
345        Ui: FnOnce(T) + 'static,
346    {
347        if !self.is_active() {
348            return;
349        }
350        let active_for_task = Arc::clone(&self.active);
351        let scope = self.clone();
352        wasm_bindgen_futures::spawn_local(async move {
353            let token = CancelToken::new(Arc::clone(&active_for_task));
354            let value = work(token.clone()).await;
355            if token.is_cancelled() {
356                return;
357            }
358            scope.post_ui(move || {
359                if token.is_active() {
360                    on_ui(value);
361                }
362            });
363        });
364    }
365}
366
367#[derive(Clone)]
368/// Cooperative cancellation token passed into background `LaunchedEffect` work.
369///
370/// The token flips to "cancelled" when the associated scope leaves composition.
371/// Callers should periodically check [`CancelToken::is_cancelled`] in long-running
372/// operations and exit early; blocking I/O will not be interrupted automatically.
373pub struct CancelToken {
374    active: Arc<AtomicBool>,
375}
376
377impl CancelToken {
378    fn new(active: Arc<AtomicBool>) -> Self {
379        Self { active }
380    }
381
382    /// Returns `true` once the associated scope has been cancelled.
383    pub fn is_cancelled(&self) -> bool {
384        !self.active.load(Ordering::SeqCst)
385    }
386
387    /// Returns whether the scope is still active.
388    pub fn is_active(&self) -> bool {
389        self.active.load(Ordering::SeqCst)
390    }
391}
392
393pub fn __launched_effect_impl<K, F>(group_key: Key, keys: K, effect: F)
394where
395    K: PartialEq + 'static,
396    F: FnOnce(LaunchedEffectScope) + 'static,
397{
398    with_current_composer(|composer| {
399        composer.with_group(group_key, |composer| {
400            let key = EffectKey::new(keys);
401            let state = composer.remember_effect::<LaunchedEffectState>();
402            if state.with(|state| state.should_run(&key)) {
403                state.update(|state| state.set_key(key));
404                let runtime = composer.runtime_handle();
405                let state_for_effect = state.clone();
406                let mut effect_opt = Some(effect);
407                composer.register_side_effect(move || {
408                    if let Some(effect) = effect_opt.take() {
409                        state_for_effect.update(|state| state.launch(runtime.clone(), effect));
410                    }
411                });
412            }
413        });
414    });
415}
416
417/// Runs `effect` when this call site first enters composition, and again
418/// whenever `keys` no longer equals the value it ran with last time.
419///
420/// `keys` may be a tuple to depend on more than one value, matching Jetpack
421/// Compose's `LaunchedEffect(vararg keys)`.
422#[allow(non_snake_case)]
423#[track_caller]
424pub fn LaunchedEffect<K, F>(keys: K, effect: F)
425where
426    K: PartialEq + 'static,
427    F: FnOnce(LaunchedEffectScope) + 'static,
428{
429    __launched_effect_impl(crate::caller_location_key(), keys, effect);
430}
431
432/// `site` is where the effect was written. It travels onto the task the effect
433/// spawns, so a task the runtime is still holding says which effect started it.
434pub fn __launched_effect_async_impl<K, F>(group_key: Key, site: TaskSite, keys: K, mk_future: F)
435where
436    K: PartialEq + 'static,
437    F: FnOnce(LaunchedEffectScope) -> Pin<Box<dyn Future<Output = ()>>> + 'static,
438{
439    with_current_composer(|composer| {
440        composer.with_group(group_key, |composer| {
441            let key = EffectKey::new(keys);
442            let state = composer.remember_effect::<LaunchedEffectAsyncState>();
443            if state.with(|state| state.should_run(&key)) {
444                state.update(|state| {
445                    state.set_key(key);
446                    state.set_site(site);
447                });
448                let runtime = composer.runtime_handle();
449                let state_for_effect = state.clone();
450                let mut mk_future_opt = Some(mk_future);
451                composer.register_side_effect(move || {
452                    if let Some(mk_future) = mk_future_opt.take() {
453                        state_for_effect.update(|state| {
454                            state.launch(runtime.clone(), mk_future);
455                        });
456                    }
457                });
458            }
459        });
460    });
461}
462
463/// Like [`LaunchedEffect`], but `mk_future` builds the future to drive rather
464/// than receiving a scope to call back into synchronously.
465#[allow(non_snake_case)]
466#[track_caller]
467pub fn LaunchedEffectAsync<K, F>(keys: K, mk_future: F)
468where
469    K: PartialEq + 'static,
470    F: FnOnce(LaunchedEffectScope) -> Pin<Box<dyn Future<Output = ()>>> + 'static,
471{
472    let caller = std::panic::Location::caller();
473    let group_key = crate::location_key(caller.file(), caller.line(), caller.column());
474    __launched_effect_async_impl(group_key, TaskSite::from(caller), keys, mk_future);
475}
cranpose_core/launched_effect.rs

cranpose_core/
launched_effect.rs
