feat: minimal support for subsecond and an example (#4307)

reactive_graph/Cargo.toml

@@ -27,6 +27,7 @@ async-lock = { workspace = true, default-features = true }
 send_wrapper = { features = [
   "futures",
 ], workspace = true, default-features = true }
+subsecond = { workspace = true, default-features = true, optional = true }
 indexmap = { workspace = true, default-features = true }
 
 [target.'cfg(all(target_arch = "wasm32", target_os = "unknown"))'.dependencies]
@@ -51,6 +52,7 @@ hydration = ["dep:hydration_context"]
 effects = [
 ] # whether to run effects: should be disabled for something like server rendering
 sandboxed-arenas = []
+subsecond = ["dep:subsecond"]
 
 [package.metadata.docs.rs]
 all-features = true

reactive_graph/src/effect/render_effect.rs

@@ -9,6 +9,8 @@ use crate::{
 };
 use futures::StreamExt;
 use or_poisoned::OrPoisoned;
+#[cfg(feature = "subsecond")]
+use std::sync::Mutex;
 use std::{
     fmt::Debug,
     future::{Future, IntoFuture},
@@ -49,13 +51,39 @@ impl<T> Debug for RenderEffect<T> {
     }
 }
 
+#[cfg(feature = "subsecond")]
+type CurrentHotPtr = Box<dyn Fn() -> Option<subsecond::HotFnPtr> + Send + Sync>;
+
 impl<T> RenderEffect<T>
 where
     T: 'static,
 {
     /// Creates a new render effect, which immediately runs `fun`.
     pub fn new(fun: impl FnMut(Option<T>) -> T + 'static) -> Self {
-        Self::new_with_value_erased(Box::new(fun), None)
+        #[cfg(feature = "subsecond")]
+        let (hot_fn_ptr, fun) = {
+            let fun = Arc::new(Mutex::new(subsecond::HotFn::current(fun)));
+            (
+                {
+                    let fun = Arc::downgrade(&fun);
+                    let wrapped = send_wrapper::SendWrapper::new(move || {
+                        fun.upgrade()
+                            .map(|n| n.lock().or_poisoned().ptr_address())
+                    });
+                    // it's not redundant, it's due to the SendWrapper deref
+                    #[allow(clippy::redundant_closure)]
+                    Box::new(move || wrapped())
+                },
+                move |prev| fun.lock().or_poisoned().call((prev,)),
+            )
+        };
+
+        Self::new_with_value_erased(
+            Box::new(fun),
+            None,
+            #[cfg(feature = "subsecond")]
+            hot_fn_ptr,
+        )
     }
 
     /// Creates a new render effect with an initial value.
@@ -63,7 +91,30 @@ where
         fun: impl FnMut(Option<T>) -> T + 'static,
         initial_value: Option<T>,
     ) -> Self {
-        Self::new_with_value_erased(Box::new(fun), initial_value)
+        #[cfg(feature = "subsecond")]
+        let (hot_fn_ptr, fun) = {
+            let fun = Arc::new(Mutex::new(subsecond::HotFn::current(fun)));
+            (
+                {
+                    let fun = Arc::downgrade(&fun);
+                    let wrapped = send_wrapper::SendWrapper::new(move || {
+                        fun.upgrade()
+                            .map(|n| n.lock().or_poisoned().ptr_address())
+                    });
+                    // it's not redundant, it's due to the SendWrapper deref
+                    #[allow(clippy::redundant_closure)]
+                    Box::new(move || wrapped())
+                },
+                move |prev| fun.lock().or_poisoned().call((prev,)),
+            )
+        };
+
+        Self::new_with_value_erased(
+            Box::new(fun),
+            initial_value,
+            #[cfg(feature = "subsecond")]
+            hot_fn_ptr,
+        )
     }
 
     /// Creates a new render effect, which immediately runs `fun`.
@@ -71,6 +122,11 @@ where
         fun: impl FnMut(Option<T>) -> T + 'static,
         value: impl IntoFuture<Output = T> + 'static,
     ) -> Self {
+        #[cfg(feature = "subsecond")]
+        let mut fun = subsecond::HotFn::current(fun);
+        #[cfg(feature = "subsecond")]
+        let fun = move |prev| fun.call((prev,));
+
         Self::new_with_async_value_erased(
             Box::new(fun),
             Box::pin(value.into_future()),
@@ -79,8 +135,13 @@ where
     }
 
     fn new_with_value_erased(
-        mut fun: Box<dyn FnMut(Option<T>) -> T + 'static>,
+        #[allow(unused_mut)] mut fun: Box<dyn FnMut(Option<T>) -> T + 'static>,
         initial_value: Option<T>,
+        // this argument can be used to invalidate individual effects in the future
+        // in present experiments, I have found that it is not actually granular enough to make a difference
+        #[allow(unused)]
+        #[cfg(feature = "subsecond")]
+        hot_fn_ptr: CurrentHotPtr,
     ) -> Self {
         // codegen optimisation:
         fn prep() -> (Owner, Arc<RwLock<EffectInner>>, crate::channel::Receiver)
@@ -104,12 +165,56 @@ where
             let _ = initial_value;
             let _ = owner;
             let _ = &mut rx;
-            let _ = &mut fun;
+            let _ = fun;
         }
 
         #[cfg(feature = "effects")]
         {
             let subscriber = inner.to_any_subscriber();
+
+            #[cfg(all(feature = "subsecond", debug_assertions))]
+            let mut fun = {
+                use crate::graph::ReactiveNode;
+                use rustc_hash::FxHashMap;
+                use std::sync::{Arc, LazyLock, Mutex};
+                use subsecond::HotFnPtr;
+
+                static HOT_RELOAD_SUBSCRIBERS: LazyLock<
+                    Mutex<FxHashMap<AnySubscriber, (HotFnPtr, CurrentHotPtr)>>,
+                > = LazyLock::new(|| {
+                    subsecond::register_handler(Arc::new(|| {
+                        HOT_RELOAD_SUBSCRIBERS.lock().or_poisoned().retain(
+                            |subscriber, (prev_ptr, hot_fn_ptr)| {
+                                match hot_fn_ptr() {
+                                    None => false,
+                                    Some(curr_hot_ptr) => {
+                                        if curr_hot_ptr != *prev_ptr {
+                                            crate::log_warning(format_args!(
+                                                "{prev_ptr:?} <> \
+                                                 {curr_hot_ptr:?}",
+                                            ));
+                                            *prev_ptr = curr_hot_ptr;
+
+                                            subscriber.mark_dirty();
+                                        }
+                                        true
+                                    }
+                                }
+                            },
+                        );
+                    }));
+                    Default::default()
+                });
+
+                let mut fun = subsecond::HotFn::current(fun);
+                let initial_ptr = hot_fn_ptr().unwrap();
+                HOT_RELOAD_SUBSCRIBERS
+                    .lock()
+                    .or_poisoned()
+                    .insert(subscriber.clone(), (initial_ptr, hot_fn_ptr));
+                move |prev| fun.call((prev,))
+            };
+
             *value.write().or_poisoned() = Some(
                 owner.with(|| subscriber.with_observer(|| fun(initial_value))),
             );
@@ -230,6 +335,11 @@ where
     pub fn new_isomorphic(
         fun: impl FnMut(Option<T>) -> T + Send + Sync + 'static,
     ) -> Self {
+        #[cfg(feature = "subsecond")]
+        let mut fun = subsecond::HotFn::current(fun);
+        #[cfg(feature = "subsecond")]
+        let fun = move |prev| fun.call((prev,));
+
         fn erased<T: Send + Sync + 'static>(
             mut fun: Box<dyn FnMut(Option<T>) -> T + Send + Sync + 'static>,
         ) -> RenderEffect<T> {