fusor_async/
value.rs

1//! Read-only declarations evaluated by a coherent renderer, including while a
2//! descendant's ordinary DOM owner is still prepared.
3use crate::{CancellationSource, RequestContext};
4use fusor::{
5    OwnerHandle, Registration,
6    coherence::{Attempt, BoundaryLifetime, ReadLease},
7    versions::Versions,
8};
9use futures_util::future::{AbortHandle, Abortable, LocalBoxFuture};
10use std::{
11    cell::{Cell, RefCell},
12    future::Future,
13    rc::{Rc, Weak},
14};
15
16thread_local! { static NEXT: Cell<u64> = const { Cell::new(0) }; }
17
18/// A scoped, typed result. Pending is an ordinary value, never control flow via
19/// panic. The HTML compiler evaluates an `Await` subtree for `Ready`.
20pub enum AsyncRead<T> {
21    Pending,
22    Ready(Rc<T>),
23}
24
25enum State<T, E> {
26    Idle,
27    Pending,
28    Ready(Rc<T>),
29    Error(Rc<E>),
30}
31struct Work {
32    abort: AbortHandle,
33    cancellation: Option<CancellationSource>,
34}
35impl Drop for Work {
36    fn drop(&mut self) {
37        self.abort.abort();
38    }
39}
40type Load<K, T, E> = dyn Fn(K, RequestContext) -> LocalBoxFuture<'static, Result<T, E>>;
41
42struct Inner<K, T, E> {
43    id: u64,
44    owner: OwnerHandle,
45    boundary: RefCell<Option<(u64, BoundaryLifetime)>>,
46    key: Box<dyn Fn() -> K>,
47    selected: RefCell<Option<(K, Versions)>>,
48    state: RefCell<State<T, E>>,
49    generation: Cell<u64>,
50    retry: Cell<u64>,
51    request: RefCell<Option<Work>>,
52    cleanup: RefCell<Option<Registration>>,
53    load: Rc<Load<K, T, E>>,
54    spawn: Box<dyn Fn(LocalBoxFuture<'static, ()>)>,
55}
56
57impl<K, T, E> Inner<K, T, E> {
58    fn cancel_work(&self) {
59        self.generation.set(
60            self.generation
61                .get()
62                .checked_add(1)
63                .expect("read generation overflow"),
64        );
65        let work = self.request.take();
66        if matches!(*self.state.borrow(), State::Pending) {
67            *self.state.borrow_mut() = State::Idle;
68        }
69        drop(work);
70    }
71}
72impl<K, T, E> Drop for Inner<K, T, E> {
73    fn drop(&mut self) {
74        self.request.get_mut().take();
75    }
76}
77
78struct Lease<K, T, E>(Weak<Inner<K, T, E>>);
79impl<K, T, E> ReadLease for Lease<K, T, E> {
80    fn cancel(&self) {
81        if let Some(inner) = self.0.upgrade() {
82            inner.cancel_work();
83        }
84    }
85}
86
87/// A declaration of one read. All changing request inputs belong in `key`.
88/// Loader execution is untracked and uses the supplied local executor. Reads
89/// participate when reached by a boundary, not merely when declared.
90pub struct AsyncValue<K, T, E>(Rc<Inner<K, T, E>>);
91impl<K, T, E> Clone for AsyncValue<K, T, E> {
92    fn clone(&self) -> Self {
93        Self(self.0.clone())
94    }
95}
96
97impl<K: Clone + PartialEq + 'static, T: 'static, E: std::fmt::Display + 'static>
98    AsyncValue<K, T, E>
99{
100    pub fn new<F: Future<Output = Result<T, E>> + 'static>(
101        owner: &OwnerHandle,
102        key: impl Fn() -> K + 'static,
103        load: impl Fn(K, RequestContext) -> F + 'static,
104        spawn: impl Fn(LocalBoxFuture<'static, ()>) + 'static,
105    ) -> Self {
106        let id = NEXT.with(|next| {
107            let id = next.get().checked_add(1).expect("read id overflow");
108            next.set(id);
109            id
110        });
111        let inner = Rc::new(Inner {
112            id,
113            owner: owner.clone(),
114            boundary: RefCell::new(None),
115            key: Box::new(key),
116            selected: RefCell::new(None),
117            state: RefCell::new(State::Idle),
118            generation: Cell::new(0),
119            retry: Cell::new(0),
120            request: RefCell::new(None),
121            cleanup: RefCell::new(None),
122            load: Rc::new(move |key, context| Box::pin(load(key, context))),
123            spawn: Box::new(spawn),
124        });
125        let weak = Rc::downgrade(&inner);
126        let cleanup = owner.on_cleanup(move || {
127            if let Some(inner) = weak.upgrade() {
128                inner.cancel_work();
129                let old = inner.state.replace(State::Idle);
130                drop(old);
131            }
132        });
133        *inner.cleanup.borrow_mut() = Some(cleanup);
134        Self(inner)
135    }
136
137    /// Renderer integration. Calling this does not commit the component owner.
138    #[doc(hidden)]
139    pub fn read(&self, attempt: &Attempt) -> Result<AsyncRead<T>, String> {
140        let inner = &self.0;
141        if inner.owner.is_disposed() {
142            return Err("async read owner was disposed".into());
143        }
144        let boundary = attempt.boundary_id();
145        let (changed, live) = inner
146            .boundary
147            .borrow()
148            .as_ref()
149            .map_or((true, false), |(id, lifetime)| {
150                (*id != boundary, lifetime.is_live())
151            });
152        if changed && live {
153            return Err(
154                "one AsyncValue cannot participate in different live async boundaries".into(),
155            );
156        }
157        if changed {
158            // A parent may retain a declaration while a conditional Await is
159            // removed and remounted. Never retain the disposed boundary or let
160            // its request publish into the new view.
161            inner.cancel_work();
162            let old = inner.state.replace(State::Idle);
163            inner.selected.take();
164            inner
165                .boundary
166                .replace(Some((boundary, attempt.boundary_lifetime())));
167            drop(old);
168        }
169        let (key, versions) = Versions::capture(|| (inner.key)());
170        let selected = inner.selected.borrow().clone();
171        let compatible = selected
172            .as_ref()
173            .is_some_and(|(old, inputs)| old == &key && inputs.same(&versions));
174        if !compatible {
175            inner.cancel_work();
176            let old = inner.state.replace(State::Idle);
177            inner.selected.replace(Some((key.clone(), versions)));
178            drop(old);
179        }
180        if inner.retry.get() != attempt.retry_generation() {
181            inner.retry.set(attempt.retry_generation());
182            if matches!(*inner.state.borrow(), State::Error(_)) {
183                let old = inner.state.replace(State::Idle);
184                drop(old);
185            }
186        }
187        attempt.register(inner.id, Rc::new(Lease(Rc::downgrade(inner))));
188        if matches!(*inner.state.borrow(), State::Idle) {
189            let generation = inner.generation.get();
190            let cancellation = CancellationSource::default();
191            let context = cancellation.context();
192            let (abort, registration) = AbortHandle::new_pair();
193            *inner.request.borrow_mut() = Some(Work {
194                abort,
195                cancellation: Some(cancellation),
196            });
197            *inner.state.borrow_mut() = State::Pending;
198            let weak = Rc::downgrade(inner);
199            let load = inner.load.clone();
200            let notify = attempt.notifier();
201            (inner.spawn)(Box::pin(async move {
202                let work = async move {
203                    let result = load(key, context).await;
204                    if let Some(inner) = weak.upgrade().filter(|inner| {
205                        !inner.owner.is_disposed() && inner.generation.get() == generation
206                    }) {
207                        if let Some(mut work) = inner.request.take() {
208                            work.cancellation
209                                .take()
210                                .expect("pending request")
211                                .complete();
212                        }
213                        let old = inner.state.replace(match result {
214                            Ok(value) => State::Ready(Rc::new(value)),
215                            Err(error) => State::Error(Rc::new(error)),
216                        });
217                        drop(old);
218                        notify();
219                    }
220                };
221                let _ = Abortable::new(work, registration).await;
222            }));
223        }
224        let error = match &*inner.state.borrow() {
225            State::Ready(value) => return Ok(AsyncRead::Ready(value.clone())),
226            State::Error(error) => Some(error.clone()),
227            State::Idle | State::Pending => {
228                attempt.pending();
229                return Ok(AsyncRead::Pending);
230            }
231        };
232        Err(error.expect("error state").to_string())
233    }
234}
fusor_async/value.rs

fusor_async/
value.rs
