store effects in reactive system

TODO.md

@@ -12,7 +12,6 @@
 - ErrorBoundary
 - ssr examples
 - reactivity 
-    - Effects need to be stored (and not mem::forget)
     - Signal wrappers
     - SignalDispose implementations on all Copy types
     - untracked access warnings

examples/todomvc/src/lib.rs

@@ -207,9 +207,7 @@ pub fn TodoMVC() -> impl IntoView {
     // this is the main point of effects: to synchronize reactive state
     // with something outside the reactive system (like localStorage)
 
-    // TODO: should be a stored value instead of leaked
-    std::mem::forget(Effect::new(move |_| {
-        leptos::tachys::log("saving todos to localStorage...");
+    Effect::new(move |_| {
         if let Ok(Some(storage)) = window().local_storage() {
             let json = serde_json::to_string(&todos)
                 .expect("couldn't serialize Todos");
@@ -217,16 +215,14 @@ pub fn TodoMVC() -> impl IntoView {
                 log::error!("error while trying to set item in localStorage");
             }
         }
-    }));
+    });
 
     // focus the main input on load
-    // TODO: should be a stored value instead of leaked
-    std::mem::forget(Effect::new(move |_| {
-        leptos::tachys::log("focusing...");
+    Effect::new(move |_| {
         if let Some(input) = input_ref.get() {
             let _ = input.focus();
         }
-    }));
+    });
 
     view! {
         <main>

reactive_graph/src/effect/effect.rs

@@ -2,7 +2,7 @@ use crate::{
     channel::{channel, Receiver},
     effect::inner::EffectInner,
     graph::{AnySubscriber, SourceSet, Subscriber, ToAnySubscriber},
-    owner::Owner,
+    owner::{Owner, StoredValue},
 };
 use any_spawner::Executor;
 use futures::StreamExt;
@@ -12,21 +12,8 @@ use std::{
     sync::{Arc, RwLock},
 };
 
-pub struct Effect<T>
-where
-    T: 'static,
-{
-    value: Arc<RwLock<Option<T>>>,
-    inner: Arc<RwLock<EffectInner>>,
-}
-
-impl<T> Clone for Effect<T> {
-    fn clone(&self) -> Self {
-        Self {
-            value: Arc::clone(&self.value),
-            inner: Arc::clone(&self.inner),
-        }
-    }
+pub struct Effect {
+    inner: StoredValue<Option<Arc<RwLock<EffectInner>>>>,
 }
 
 fn effect_base() -> (Receiver, Owner, Arc<RwLock<EffectInner>>) {
@@ -46,18 +33,15 @@ fn effect_base() -> (Receiver, Owner, Arc<RwLock<EffectInner>>) {
     (rx, owner, inner)
 }
 
-impl<T> Effect<T>
-where
-    T: 'static,
-{
-    pub fn with_value_mut<U>(
-        &self,
-        fun: impl FnOnce(&mut T) -> U,
-    ) -> Option<U> {
-        self.value.write().or_poisoned().as_mut().map(fun)
+impl Effect {
+    pub fn stop(self) {
+        drop(self.inner.update_value(|inner| inner.take()));
     }
 
-    pub fn new(mut fun: impl FnMut(Option<T>) -> T + 'static) -> Self {
+    pub fn new<T>(mut fun: impl FnMut(Option<T>) -> T + 'static) -> Self
+    where
+        T: 'static,
+    {
         let (mut rx, owner, inner) = effect_base();
         let value = Arc::new(RwLock::new(None));
 
@@ -79,17 +63,17 @@ where
             }
         });
 
-        Self { value, inner }
+        Self {
+            inner: StoredValue::new(Some(inner)),
+        }
     }
-}
 
-impl<T> Effect<T>
-where
-    T: Send + Sync + 'static,
-{
-    pub fn new_sync(
+    pub fn new_sync<T>(
         mut fun: impl FnMut(Option<T>) -> T + Send + Sync + 'static,
-    ) -> Self {
+    ) -> Self
+    where
+        T: Send + Sync + 'static,
+    {
         let (mut rx, owner, inner) = effect_base();
         let value = Arc::new(RwLock::new(None));
 
@@ -110,12 +94,19 @@ where
                 }
             }
         });
-        Self { value, inner }
+        Self {
+            inner: StoredValue::new(Some(inner)),
+        }
     }
 }
 
-impl<T> ToAnySubscriber for Effect<T> {
+impl ToAnySubscriber for Effect {
     fn to_any_subscriber(&self) -> AnySubscriber {
-        self.inner.to_any_subscriber()
+        self.inner
+            .with_value(|inner| {
+                inner.as_ref().map(|inner| inner.to_any_subscriber())
+            })
+            .flatten()
+            .expect("tried to subscribe to effect that has been stopped")
     }
 }