feat: add remaining side effects

packages/rearch/lib/rearch.dart

@@ -4,8 +4,6 @@ import 'package:rearch/src/node.dart';
 export 'src/side_effects.dart';
 export 'src/types.dart';
 
-// TODO(GregoryConrad): eager garbage collection mode
-
 /// Represents a disposable object.
 // ignore: one_member_abstracts
 abstract interface class Disposable {
@@ -37,15 +35,12 @@ abstract interface class SideEffectRegistrar {
 }
 
 /// The handle given to a [Capsule] to build its data.
+/// See [CapsuleReader] and [SideEffectRegistrar].
 abstract interface class CapsuleHandle
     implements CapsuleReader, SideEffectRegistrar {}
 
 /// Defines what a [SideEffect] should look like (a [Function]
 /// that consumes a [SideEffectApi] and returns something).
-///
-/// If your side effect is more advanced or requires parameters,
-/// simply make a callable class instead of just a regular [Function]!
-///
 /// See the documentation for more.
 typedef SideEffect<T> = T Function(SideEffectApi);
 

packages/rearch/lib/src/side_effects.dart

@@ -1,3 +1,5 @@
+import 'dart:async';
+
 import 'package:rearch/rearch.dart';
 
 /// A collection of builtin side effects.
@@ -17,7 +19,7 @@ extension BuiltinSideEffects on SideEffectRegistrar {
   /// if it was provided. Otherwise, you must manually set it
   /// via the setter before ever calling the getter.
   (T Function(), void Function(T)) rawValueWrapper<T>([T Function()? init]) {
-    return register((api) {
+    return callonce(() {
       late T state;
       if (init != null) state = init();
       return (() => state, (T newState) => state = newState);
@@ -100,7 +102,155 @@ extension BuiltinSideEffects on SideEffectRegistrar {
     }
   }
 
-  // TODO(GregoryConrad): other side effects
+  /// A simple implementation of the reducer pattern as a side effect.
+  ///
+  /// The React docs do a great job of explaining the reducer pattern more.
+  /// See https://react.dev/reference/react/useReducer
+  (State, void Function(Action)) reducer<State, Action>(
+    Reducer<State, Action> reducer,
+    State initialState,
+  ) {
+    final (currState, setState) = state(initialState);
+    return (
+      currState,
+      (action) => setState(reducer(currState, action)),
+    );
+  }
+
+  /// Consumes a [Future] and watches the given [future].
+  ///
+  /// Implemented by calling [Future.asStream] and forwarding calls
+  /// onto [stream].
+  ///
+  /// If the given future changes, then the current [StreamSubscription]
+  /// will be disposed and recreated for the new future.
+  /// Thus, it is important that the future instance only changes when needed.
+  /// It is incorrect to create a future in the same build as the [future],
+  /// unless you use something like [memo] to limit changes.
+  /// Or, if possible, it is even better to wrap the future in an entirely
+  /// new capsule (although this is not always possible).
+  AsyncValue<T> future<T>(Future<T> future) {
+    final asStream = memo(future.asStream, [future]);
+    return stream(asStream);
+  }
+
+  /// Consumes a [Stream] and watches the given stream.
+  ///
+  /// If the given stream changes between build calls, then the current
+  /// [StreamSubscription] will be disposed and recreated for the new stream.
+  /// Thus, it is important that the stream instance only changes when needed.
+  /// It is incorrect to create a stream in the same build as the [stream],
+  /// unless you use something like [memo] to limit changes.
+  /// Or, if possible, it is even better to wrap the stream in an entirely
+  /// new capsule (although this is not always possible).
+  AsyncValue<T> stream<T>(Stream<T> stream) {
+    final rebuild = rebuilder();
+    final (getValue, setValue) = rawValueWrapper<AsyncValue<T>>(
+      () => AsyncLoading<T>(None<T>()),
+    );
+
+    final (getSubscription, setSubscription) =
+        rawValueWrapper<StreamSubscription<T>?>(() => null);
+    effect(() => getSubscription()?.cancel, [getSubscription()]);
+
+    final oldStream = previous(stream);
+    final needToInitializeState = stream != oldStream;
+
+    if (needToInitializeState) {
+      setValue(AsyncLoading(getValue().data));
+      setSubscription(
+        stream.listen(
+          (data) {
+            setValue(AsyncData(data));
+            rebuild();
+          },
+          onError: (Object error, StackTrace trace) {
+            setValue(AsyncError(error, trace, getValue().data));
+            rebuild();
+          },
+          cancelOnError: false,
+        ),
+      );
+    }
+
+    return getValue();
+  }
+
+  /// A mechanism to persist changes made in state.
+  /// See the docs for usage information.
+  ///
+  /// Defines the way to interact with a storage provider of your choice
+  /// through the [read] and [write] parameters.
+  ///
+  /// [read] is only called once; it is assumed that if [write] is successful,
+  /// then calling [read] again would reflect the new state that we already
+  /// have access to. Thus, calling [read] again is skipped as an optimization.
+  (AsyncValue<T>, void Function(T)) persist<T>({
+    required Future<T> Function() read,
+    required Future<void> Function(T) write,
+  }) {
+    final readFuture = callonce(read);
+    final readState = future(readFuture);
+    final (state: writeState, :mutate, clear: _) = mutation<T>();
+
+    final state = (writeState ?? readState).fillInPreviousData(readState.data);
+
+    void persist(T data) {
+      mutate(() async {
+        await write(data);
+        return data;
+      }());
+    }
+
+    return (state, persist);
+  }
+
+  /// Allows you to trigger and watch [Future]s
+  /// (called mutations, since they often mutate some state)
+  /// from within the build function.
+  /// See the documentation for more.
+  ///
+  /// Note: `mutate()` and `clear()` *should not* be called directly from
+  /// within build, but rather from within some callback.
+  Mutation<T> mutation<T>() {
+    final rebuild = rebuilder();
+    final (getValue, setValue) = rawValueWrapper<AsyncValue<T>?>(() => null);
+
+    // We convert to a stream here because we can cancel a stream subscription;
+    // there is no builtin way to cancel a future.
+    final (future, setFuture) = state<Future<T>?>(null);
+    final asStream = memo(() => future?.asStream(), [future]);
+
+    effect(
+      () {
+        setValue(
+          asStream == null ? null : AsyncLoading(getValue()?.data ?? None<T>()),
+        );
+
+        final subscription = asStream?.listen(
+          (data) {
+            setValue(AsyncData(data));
+            rebuild();
+          },
+          onError: (Object error, StackTrace trace) {
+            setValue(
+              AsyncError(error, trace, getValue()?.data ?? None<T>()),
+            );
+            rebuild();
+          },
+        );
+
+        return () => subscription?.cancel();
+      },
+      [asStream],
+    );
+
+    return (
+      state: getValue(),
+      mutate: setFuture,
+      clear: () => setFuture(null),
+    );
+  }
 }
 
 /// Checks to see whether [newDeps] has changed from [oldDeps]
@@ -111,3 +261,18 @@ bool _didDepsListChange(List<Object?> newDeps, List<Object?>? oldDeps) {
       Iterable<int>.generate(newDeps.length)
           .any((i) => newDeps[i] != oldDeps[i]);
 }
+
+/// A reducer [Function] that consumes some [State] and [Action] and returns
+/// a new, transformed [State].
+typedef Reducer<State, Action> = State Function(State, Action);
+
+/// Represents a mutation, with:
+/// - `state`, the current [AsyncValue]? state of the mutation
+/// (will be null when the mutation is inactive/cleared)
+/// - `mutate`, a [Function] that triggers the mutation
+/// - `clear`, a [Function] that stops and clears the mutation
+typedef Mutation<T> = ({
+  AsyncValue<T>? state,
+  void Function(Future<T> mutater) mutate,
+  void Function() clear,
+});