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

cranpose_core/
launched_effect.rs
