dioxus_core/
reactive_context.rs

1use crate::{current_scope_id, scope_context::Scope, tasks::SchedulerMsg, Runtime, ScopeId};
2use futures_channel::mpsc::UnboundedReceiver;
3use generational_box::{BorrowMutError, GenerationalBox, SyncStorage};
4use std::{
5    cell::RefCell,
6    collections::HashSet,
7    hash::Hash,
8    sync::{Arc, Mutex},
9};
10
11#[doc = include_str!("../docs/reactivity.md")]
12#[derive(Clone, Copy)]
13pub struct ReactiveContext {
14    scope: ScopeId,
15    inner: GenerationalBox<Inner, SyncStorage>,
16}
17
18impl PartialEq for ReactiveContext {
19    fn eq(&self, other: &Self) -> bool {
20        self.inner.ptr_eq(&other.inner)
21    }
22}
23
24impl Eq for ReactiveContext {}
25
26thread_local! {
27    static CURRENT: RefCell<Vec<ReactiveContext>> = const { RefCell::new(vec![]) };
28}
29
30impl std::fmt::Display for ReactiveContext {
31    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
32        #[cfg(debug_assertions)]
33        {
34            if let Ok(read) = self.inner.try_read() {
35                if let Some(scope) = read.scope {
36                    return write!(f, "ReactiveContext(for scope: {:?})", scope);
37                }
38                return write!(f, "ReactiveContext created at {}", read.origin);
39            }
40        }
41        write!(f, "ReactiveContext")
42    }
43}
44
45impl ReactiveContext {
46    /// Create a new reactive context
47    #[track_caller]
48    pub fn new() -> (Self, UnboundedReceiver<()>) {
49        Self::new_with_origin(std::panic::Location::caller())
50    }
51
52    /// Create a new reactive context with a location for debugging purposes
53    /// This is useful for reactive contexts created within closures
54    pub fn new_with_origin(
55        origin: &'static std::panic::Location<'static>,
56    ) -> (Self, UnboundedReceiver<()>) {
57        let (tx, rx) = futures_channel::mpsc::unbounded();
58        let callback = move || {
59            // If there is already an update queued, we don't need to queue another
60            if !tx.is_empty() {
61                return;
62            }
63            let _ = tx.unbounded_send(());
64        };
65        let _self = Self::new_with_callback(
66            callback,
67            current_scope_id().unwrap_or_else(|e| panic!("{}", e)),
68            origin,
69        );
70        (_self, rx)
71    }
72
73    /// Create a new reactive context that may update a scope. When any signal that this context subscribes to changes, the callback will be run
74    pub fn new_with_callback(
75        callback: impl FnMut() + Send + Sync + 'static,
76        scope: ScopeId,
77        #[allow(unused)] origin: &'static std::panic::Location<'static>,
78    ) -> Self {
79        let inner = Inner {
80            self_: None,
81            update: Box::new(callback),
82            subscribers: Default::default(),
83            #[cfg(debug_assertions)]
84            origin,
85            #[cfg(debug_assertions)]
86            scope: None,
87        };
88
89        let owner = scope.owner();
90
91        let self_ = Self {
92            scope,
93            inner: owner.insert(inner),
94        };
95
96        self_.inner.write().self_ = Some(self_);
97
98        self_
99    }
100
101    /// Get the current reactive context from the nearest reactive hook or scope
102    pub fn current() -> Option<Self> {
103        CURRENT.with(|current| current.borrow().last().cloned())
104    }
105
106    /// Create a reactive context for a scope id
107    pub(crate) fn new_for_scope(scope: &Scope, runtime: &Runtime) -> Self {
108        let id = scope.id;
109        let sender = runtime.sender.clone();
110        let update_scope = move || {
111            tracing::trace!("Marking scope {:?} as dirty", id);
112            sender.unbounded_send(SchedulerMsg::Immediate(id)).unwrap();
113        };
114
115        // Otherwise, create a new context at the current scope
116        let inner = Inner {
117            self_: None,
118            update: Box::new(update_scope),
119            subscribers: Default::default(),
120            #[cfg(debug_assertions)]
121            origin: std::panic::Location::caller(),
122            #[cfg(debug_assertions)]
123            scope: Some(id),
124        };
125
126        let owner = scope.owner();
127
128        let self_ = Self {
129            scope: id,
130            inner: owner.insert(inner),
131        };
132
133        self_.inner.write().self_ = Some(self_);
134
135        self_
136    }
137
138    /// Clear all subscribers to this context
139    pub fn clear_subscribers(&self) {
140        // The key type is mutable, but the hash is stable through mutations because we hash by pointer
141        #[allow(clippy::mutable_key_type)]
142        let old_subscribers = std::mem::take(&mut self.inner.write().subscribers);
143        for subscriber in old_subscribers {
144            subscriber.0.remove(self);
145        }
146    }
147
148    /// Update the subscribers
149    pub(crate) fn update_subscribers(&self) {
150        #[allow(clippy::mutable_key_type)]
151        let subscribers = &self.inner.read().subscribers;
152        for subscriber in subscribers.iter() {
153            subscriber.0.add(*self);
154        }
155    }
156
157    /// Reset the reactive context and then run the callback in the context. This can be used to create custom reactive hooks like `use_memo`.
158    ///
159    /// ```rust, no_run
160    /// # use dioxus::prelude::*;
161    /// # use dioxus_core::ReactiveContext;
162    /// # use futures_util::StreamExt;
163    /// fn use_simplified_memo(mut closure: impl FnMut() -> i32 + 'static) -> Signal<i32> {
164    ///     use_hook(|| {
165    ///         // Create a new reactive context and channel that will receive a value every time a value the reactive context subscribes to changes
166    ///         let (reactive_context, mut changed) = ReactiveContext::new();
167    ///         // Compute the value of the memo inside the reactive context. This will subscribe the reactive context to any values you read inside the closure
168    ///         let value = reactive_context.reset_and_run_in(&mut closure);
169    ///         // Create a new signal with the value of the memo
170    ///         let mut signal = Signal::new(value);
171    ///         // Create a task that reruns the closure when the reactive context changes
172    ///         spawn(async move {
173    ///             while changed.next().await.is_some() {
174    ///                 // Since we reset the reactive context as we run the closure, our memo will only subscribe to the new values that are read in the closure
175    ///                 let new_value = reactive_context.run_in(&mut closure);
176    ///                 if new_value != value {
177    ///                     signal.set(new_value);
178    ///                 }
179    ///             }
180    ///         });
181    ///         signal
182    ///     })
183    /// }
184    ///
185    /// let mut boolean = use_signal(|| false);
186    /// let mut count = use_signal(|| 0);
187    /// // Because we use `reset_and_run_in` instead of just `run_in`, our memo will only subscribe to the signals that are read this run of the closure (initially just the boolean)
188    /// let memo = use_simplified_memo(move || if boolean() { count() } else { 0 });
189    /// println!("{memo}");
190    /// // Because the count signal is not read in this run of the closure, the memo will not rerun
191    /// count += 1;
192    /// println!("{memo}");
193    /// // Because the boolean signal is read in this run of the closure, the memo will rerun
194    /// boolean.toggle();
195    /// println!("{memo}");
196    /// // If we toggle the boolean again, and the memo unsubscribes from the count signal
197    /// boolean.toggle();
198    /// println!("{memo}");
199    /// ```
200    pub fn reset_and_run_in<O>(&self, f: impl FnOnce() -> O) -> O {
201        self.clear_subscribers();
202        self.run_in(f)
203    }
204
205    /// Run this function in the context of this reactive context
206    ///
207    /// This will set the current reactive context to this context for the duration of the function.
208    /// You can then get information about the current subscriptions.
209    pub fn run_in<O>(&self, f: impl FnOnce() -> O) -> O {
210        CURRENT.with(|current| current.borrow_mut().push(*self));
211        let out = f();
212        CURRENT.with(|current| current.borrow_mut().pop());
213        self.update_subscribers();
214        out
215    }
216
217    /// Marks this reactive context as dirty
218    ///
219    /// If there's a scope associated with this context, then it will be marked as dirty too
220    ///
221    /// Returns true if the context was marked as dirty, or false if the context has been dropped
222    pub fn mark_dirty(&self) -> bool {
223        if let Ok(mut self_write) = self.inner.try_write() {
224            #[cfg(debug_assertions)]
225            {
226                tracing::trace!(
227                    "Marking reactive context created at {} as dirty",
228                    self_write.origin
229                );
230            }
231
232            (self_write.update)();
233
234            true
235        } else {
236            false
237        }
238    }
239
240    /// Subscribe to this context. The reactive context will automatically remove itself from the subscriptions when it is reset.
241    pub fn subscribe(&self, subscriptions: impl Into<Subscribers>) {
242        match self.inner.try_write() {
243            Ok(mut inner) => {
244                let subscriptions = subscriptions.into();
245                subscriptions.add(*self);
246                inner
247                    .subscribers
248                    .insert(PointerHash(subscriptions.inner.clone()));
249            }
250            // If the context was dropped, we don't need to subscribe to it anymore
251            Err(BorrowMutError::Dropped(_)) => {}
252            Err(expect) => {
253                panic!(
254                    "Expected to be able to write to reactive context to subscribe, but it failed with: {expect:?}"
255                );
256            }
257        }
258    }
259
260    /// Get the scope that inner CopyValue is associated with
261    pub fn origin_scope(&self) -> ScopeId {
262        self.scope
263    }
264}
265
266impl Hash for ReactiveContext {
267    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
268        self.inner.id().hash(state);
269    }
270}
271
272struct PointerHash<T: ?Sized>(Arc<T>);
273
274impl<T: ?Sized> Hash for PointerHash<T> {
275    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
276        std::sync::Arc::<T>::as_ptr(&self.0).hash(state);
277    }
278}
279
280impl<T: ?Sized> PartialEq for PointerHash<T> {
281    fn eq(&self, other: &Self) -> bool {
282        std::sync::Arc::ptr_eq(&self.0, &other.0)
283    }
284}
285
286impl<T: ?Sized> Eq for PointerHash<T> {}
287
288impl<T: ?Sized> Clone for PointerHash<T> {
289    fn clone(&self) -> Self {
290        Self(self.0.clone())
291    }
292}
293
294struct Inner {
295    self_: Option<ReactiveContext>,
296
297    // Futures will call .changed().await
298    update: Box<dyn FnMut() + Send + Sync>,
299
300    // Subscribers to this context
301    subscribers: HashSet<PointerHash<dyn SubscriberList + Send + Sync>>,
302
303    // Debug information for signal subscriptions
304    #[cfg(debug_assertions)]
305    origin: &'static std::panic::Location<'static>,
306
307    #[cfg(debug_assertions)]
308    // The scope that this reactive context is associated with
309    scope: Option<ScopeId>,
310}
311
312impl Drop for Inner {
313    fn drop(&mut self) {
314        let Some(self_) = self.self_.take() else {
315            return;
316        };
317
318        for subscriber in std::mem::take(&mut self.subscribers) {
319            subscriber.0.remove(&self_);
320        }
321    }
322}
323
324/// A list of [ReactiveContext]s that are subscribed. This is used to notify subscribers when the value changes.
325#[derive(Clone)]
326pub struct Subscribers {
327    /// The list of subscribers.
328    pub(crate) inner: Arc<dyn SubscriberList + Send + Sync>,
329}
330
331impl Default for Subscribers {
332    fn default() -> Self {
333        Self::new()
334    }
335}
336
337impl Subscribers {
338    /// Create a new no-op list of subscribers.
339    pub fn new_noop() -> Self {
340        struct NoopSubscribers;
341        impl SubscriberList for NoopSubscribers {
342            fn add(&self, _subscriber: ReactiveContext) {}
343
344            fn remove(&self, _subscriber: &ReactiveContext) {}
345
346            fn visit(&self, _f: &mut dyn FnMut(&ReactiveContext)) {}
347        }
348        Subscribers {
349            inner: Arc::new(NoopSubscribers),
350        }
351    }
352
353    /// Create a new list of subscribers.
354    pub fn new() -> Self {
355        Subscribers {
356            inner: Arc::new(Mutex::new(HashSet::new())),
357        }
358    }
359
360    /// Add a subscriber to the list.
361    pub fn add(&self, subscriber: ReactiveContext) {
362        self.inner.add(subscriber);
363    }
364
365    /// Remove a subscriber from the list.
366    pub fn remove(&self, subscriber: &ReactiveContext) {
367        self.inner.remove(subscriber);
368    }
369
370    /// Visit all subscribers in the list.
371    pub fn visit(&self, mut f: impl FnMut(&ReactiveContext)) {
372        self.inner.visit(&mut f);
373    }
374}
375
376impl<S: SubscriberList + Send + Sync + 'static> From<Arc<S>> for Subscribers {
377    fn from(inner: Arc<S>) -> Self {
378        Subscribers { inner }
379    }
380}
381
382/// A list of subscribers that can be notified when the value changes. This is used to track when the value changes and notify subscribers.
383pub trait SubscriberList: Send + Sync {
384    /// Add a subscriber to the list.
385    fn add(&self, subscriber: ReactiveContext);
386
387    /// Remove a subscriber from the list.
388    fn remove(&self, subscriber: &ReactiveContext);
389
390    /// Visit all subscribers in the list.
391    fn visit(&self, f: &mut dyn FnMut(&ReactiveContext));
392}
393
394impl SubscriberList for Mutex<HashSet<ReactiveContext>> {
395    fn add(&self, subscriber: ReactiveContext) {
396        if let Ok(mut lock) = self.lock() {
397            lock.insert(subscriber);
398        } else {
399            tracing::warn!("Failed to lock subscriber list to add subscriber: {subscriber}");
400        }
401    }
402
403    fn remove(&self, subscriber: &ReactiveContext) {
404        if let Ok(mut lock) = self.lock() {
405            lock.remove(subscriber);
406        } else {
407            tracing::warn!("Failed to lock subscriber list to remove subscriber: {subscriber}");
408        }
409    }
410
411    fn visit(&self, f: &mut dyn FnMut(&ReactiveContext)) {
412        if let Ok(lock) = self.lock() {
413            lock.iter().for_each(f);
414        } else {
415            tracing::warn!("Failed to lock subscriber list to visit subscribers");
416        }
417    }
418}
dioxus_core/reactive_context.rs

dioxus_core/
reactive_context.rs
