support for lazy hydration

reactive_graph/src/effect/render_effect.rs

@@ -11,7 +11,9 @@ use futures::StreamExt;
 use or_poisoned::OrPoisoned;
 use std::{
     fmt::Debug,
+    future::{Future, IntoFuture},
     mem,
+    pin::Pin,
     sync::{Arc, RwLock, Weak},
 };
 
@@ -64,6 +66,18 @@ where
         Self::new_with_value_erased(Box::new(fun), initial_value)
     }
 
+    /// Creates a new render effect, which immediately runs `fun`.
+    pub async fn new_with_async_value(
+        fun: impl FnMut(Option<T>) -> T + 'static,
+        value: impl IntoFuture<Output = T> + 'static,
+    ) -> Self {
+        Self::new_with_async_value_erased(
+            Box::new(fun),
+            Box::pin(value.into_future()),
+        )
+        .await
+    }
+
     fn new_with_value_erased(
         mut fun: Box<dyn FnMut(Option<T>) -> T + 'static>,
         initial_value: Option<T>,
@@ -127,6 +141,73 @@ where
         RenderEffect { value, inner }
     }
 
+    async fn new_with_async_value_erased(
+        mut fun: Box<dyn FnMut(Option<T>) -> T + 'static>,
+        initial_value: Pin<Box<dyn Future<Output = T>>>,
+    ) -> Self {
+        // codegen optimisation:
+        fn prep() -> (Owner, Arc<RwLock<EffectInner>>, crate::channel::Receiver)
+        {
+            let (observer, rx) = channel();
+            let owner = Owner::new();
+            let inner = Arc::new(RwLock::new(EffectInner {
+                dirty: false,
+                observer,
+                sources: SourceSet::new(),
+            }));
+            (owner, inner, rx)
+        }
+
+        let (owner, inner, mut rx) = prep();
+
+        let value = Arc::new(RwLock::new(None::<T>));
+
+        #[cfg(not(feature = "effects"))]
+        {
+            let _ = initial_value;
+            let _ = owner;
+            let _ = &mut rx;
+            let _ = &mut fun;
+        }
+
+        #[cfg(feature = "effects")]
+        {
+            use crate::computed::ScopedFuture;
+
+            let subscriber = inner.to_any_subscriber();
+
+            let initial = subscriber
+                .with_observer(|| ScopedFuture::new(initial_value))
+                .await;
+            *value.write().or_poisoned() = Some(initial);
+
+            any_spawner::Executor::spawn_local({
+                let value = Arc::clone(&value);
+
+                async move {
+                    while rx.next().await.is_some() {
+                        if !owner.paused()
+                            && subscriber.with_observer(|| {
+                                subscriber.update_if_necessary()
+                            })
+                        {
+                            subscriber.clear_sources(&subscriber);
+
+                            let old_value =
+                                mem::take(&mut *value.write().or_poisoned());
+                            let new_value = owner.with_cleanup(|| {
+                                subscriber.with_observer(|| fun(old_value))
+                            });
+                            *value.write().or_poisoned() = Some(new_value);
+                        }
+                    }
+                }
+            });
+        }
+
+        RenderEffect { value, inner }
+    }
+
     /// Mutably accesses the current value.
     pub fn with_value_mut<U>(
         &self,