Re-implementing Riverpod

!Pods.md

Hey 👋

This is an attempt to recreate to a degree the essence of Riverpod. There is also a Flutter version https://gist.github.com/jogboms/f7f44f11e2d9aa30d355f94f5257281d.

Warning

As you would later notice, there are two major missing pieces: A robust error handling mechanism & Tests. The goal is educational and not a replacement of the said library.

See the hosted version on DartPad.

main.dart

import 'dart:async';
import 'dart:math';

import 'package:meta/meta.dart';

/* Start Example */

final FuturePod<int> randomIntPod = FuturePod((_) => Random().nextInt(1000));

final ValuePod<String> subjectPod = ValuePod((_) => 'World');

final ValuePod<int> countPod = ValuePod((_) => 3);

final ValuePod<String> greetingPod = ValuePod((Ref ref) => 'Hello (${ref.read(countPod)})');

final StatePod<bool> isLoggedInPod = StatePod((_) => false);

final FuturePod<int> scorePod = FuturePod(
  (Ref ref) async => Future<int>.delayed(
    Duration(seconds: ref.read(countPod)),
    () => 1000,
  ),
);

final FuturePod<int> doubleScorePod = FuturePod(
  (Ref ref) async {
    final int score = await ref.watch(scorePod.future);

    return score * 2;
  },
);

final StreamPod<double> tenthScorePod = StreamPod(
  (Ref ref) async* {
    final int score = await ref.watch(doubleScorePod.future);
    for (int i = 1; i < 5; i++) {
      await Future<void>.delayed(Duration(seconds: 1));
      yield score * i.toDouble();
    }
  },
);

final StreamPod<int> finalScorePod = StreamPod(
  (Ref ref) async* {
    final bool isLoggedIn = await ref.watch(isLoggedInPod);

    final int i = switch (ref.watch(tenthScorePod)) {
      PendingAsyncValue() => 0,
      SuccessfulAsyncValue(:final double data) => data.toInt(),
      FailedAsyncValue() => -1,
    };

    yield (isLoggedIn ? 2000 : 1000) * i;
  },
);

void main() async {
  final Pods ref = Pods();

  ref.listen(isLoggedInPod, (bool? previous, bool next) {
    print(('is logged in?', previous, next));
  }, fireImmediately: true);

  print(('double score', await ref.read(doubleScorePod.future)));

  ref.listen(finalScorePod, (_, AsyncValue<int> next) {
    final int? data = switch (next) {
      PendingAsyncValue() => null,
      SuccessfulAsyncValue(:final int data) => data,
      FailedAsyncValue() => null,
    };
    print(('final score', data ?? 'loading...'));
  });
  
  ref.read(isLoggedInPod.notifier).update((_) => true);
}

/* End Example */

/* Start Pods */

typedef VoidCallback = void Function();
typedef ValueCallback<T> = T Function();
typedef ValueMapper<U, T> = T Function(U);

@optionalTypeArgs
typedef PodOverride<T> = ({Pod<T> origin, Pod<T> override});
@optionalTypeArgs
typedef PodSubscription<U> = ({ValueCallback<U> read, VoidCallback close});
typedef PodListener<U> = void Function(U? previous, U next);
typedef PodFactory<T> = ValueMapper<Ref, T>;
typedef PodUpdater<T> = ValueMapper<T?, T>;
typedef FamilyPodFactory<T, U> = T Function(Ref ref, U args);

abstract interface class Ref {
  U read<U>(Pod<U> pod);

  U watch<U>(WatchablePod<U> pod);

  PodSubscription<U> listen<U>(
    ListenablePod<U> pod,
    PodListener<U> listener, {
    bool fireImmediately = false,
  });

  void invalidate<U>(Pod<U> pod);

  void onDispose(VoidCallback callback);
}

/* End Pods */

/* Start Pod Element */

@optionalTypeArgs
abstract base class PodElement<T> {
  PodElement([this._state]);

  String? _debugLabel;
  late Ref _ref;

  final Set<PodListener<T>> _listeners = <PodListener<T>>{};
  final Set<PodElement> _references = <PodElement>{};
  final Set<PodElement> _dependents = <PodElement>{};
  final Set<VoidCallback> _disposeListener = <VoidCallback>{};

  @protected
  T get state => _state!;
  T? _state;

  @protected
  set state(T newState) {
    final T? oldState = _state;
    _state = newState;
    _notifyListeners(oldState);
  }

  @protected
  bool get mounted => _state != null;

  @override
  bool operator ==(Object other) =>
      identical(this, other) || other is PodElement<T> && runtimeType == other.runtimeType;

  @override
  int get hashCode => identityHashCode(this);

  @mustCallSuper
  void mount(covariant Ref ref) {
    assert(state != null, 'Set the state after mount');
    _ref = ref;
  }

  @mustCallSuper
  void dispose() {
    _notifyDisposeCallbacks();

    _state = null;
    _debugLabel = null;
    _listeners.clear();
    _dependents.clear();
    _references.clear();
  }

  @override
  String toString() => (StringBuffer()
        ..write(_debugLabel != null ? '${_debugLabel}Element' : '$runtimeType')
        ..write('($hashCode)'))
      .toString();

  // todo: maybe do not mount immediately
  // todo: maybe introduce a scheduler
  void _invalidate() {
    _notifyDisposeCallbacks();
    Future<void>(() {
      if (_ref case final Ref ref) {
        final T? oldState = _state;
        mount(ref);
        _notifyListeners(oldState);
      }
    });
  }

  void _attachTo(PodElement element) => element._references.add(this);

  void _dependsOn(PodElement element) => element._dependents.add(this);

  VoidCallback _addListener(PodListener<T> listener) {
    _listeners.add(listener);
    return () => _listeners.remove(listener);
  }

  void _addDisposeListener(VoidCallback callback) => _disposeListener.add(callback);

  @protected
  void _notifyListeners(T? oldState) {
    if (oldState == state) {
      return;
    }

    for (int i = 0; i < _listeners.length; i++) {
      _listeners.elementAt(i)(oldState, state);
    }

    for (int i = _dependents.length - 1; i >= 0; i--) {
      final PodElement dependent = _dependents.elementAt(i);
      dependent.mount(dependent._ref);
    }
  }

  void _notifyDisposeCallbacks() {
    for (final VoidCallback callback in _disposeListener) {
      callback();
    }
    _disposeListener.clear();
  }
}

/* End Pod Element */

/* Start Pod */

@optionalTypeArgs
abstract base class Pod<T> {
  const Pod({this.debugLabel});

  final String? debugLabel;

  static ValueMapper<U, V> _family<U, V>(ValueMapper<U, V> factory) => (U args) => factory(args);

  @factory
  PodElement<T> createElement();

  @override
  bool operator ==(Object other) =>
      identical(this, other) || other is Pod<T> && runtimeType == other.runtimeType && debugLabel == other.debugLabel;

  @override
  int get hashCode => debugLabel?.hashCode ?? identityHashCode(this);

  @override
  String toString() => '${debugLabel ?? runtimeType}($hashCode)';
}

@optionalTypeArgs
base mixin ListenablePod<T> on Pod<T> {}

@optionalTypeArgs
base mixin WatchablePod<T> on Pod<T> {}

@optionalTypeArgs
base mixin NotifiablePod<T> implements WatchablePod<T>, ListenablePod<T> {}

@optionalTypeArgs
base mixin AsyncPod<T, U> implements NotifiablePod<AsyncValue<T>> {
  PodFactory<U> get _factory;

  WatchablePod<FutureOr<T>> get future;
}

/* End Pod */

/* Start Pods Container */

class Pods implements Ref {
  Pods({
    Pods? parent,
    PodElement? owner,
    List<PodOverride> overrides = const <PodOverride>[],
  }) : this._(
          owner: owner ?? (_AnonymousPodElement().._debugLabel = 'RootElement'),
          pods: <Pod, PodElement>{
            ...?parent?._pods,
          },
          overrides: <Pod, Pod>{
            ...?parent?._overrides,
            for (final PodOverride entry in overrides) entry.origin: entry.override,
          },
        );

  const Pods._({
    required PodElement owner,
    required Map<Pod, PodElement> pods,
    required Map<Pod, Pod> overrides,
  })  : _owner = owner,
        _pods = pods,
        _overrides = overrides;

  final PodElement _owner;
  final Map<Pod, PodElement> _pods;
  final Map<Pod, Pod> _overrides;

  @override
  U read<U>(Pod<U> pod) => _resolvePodElement(pod).state;

  @Deprecated('.watch is not supported in CLI mode, instead use .listen')
  @override
  U watch<U>(WatchablePod<U> pod) {
    final PodElement<U> element = _resolvePodElement(pod);
    _owner._dependsOn(element);
    return element.state;
  }

  @override
  PodSubscription<U> listen<U>(
    ListenablePod<U> pod,
    PodListener<U> listener, {
    bool fireImmediately = false,
  }) {
    final PodElement<U> element = _resolvePodElement(pod, shouldMount: false);

    final VoidCallback dispose = element._addListener(listener);
    if (!element.mounted) {
      _mountPodElement(element);
    }
    if (fireImmediately) {
      listener(null, element.state);
    }

    return (
      read: () => element.state,
      close: dispose,
    );
  }

  @override
  void invalidate<U>(Pod<U> pod) {
    return switch ((_pods[pod], _overrides[pod])) {
      (_, Pod<U> podOverride) => invalidate(podOverride),
      (PodElement<U> element, _) => element._invalidate(),
      _ => null,
    };
  }

  @Deprecated('not supposed to use this directly')
  @override
  void onDispose(VoidCallback callback) => _owner._addDisposeListener(callback);

  void updateOverrides(List<PodOverride> overrides, Pods? parent) {
    Map<Pod, Pod>? parentOverrides = parent?._overrides;
    final Set<PodOverride> combinedOverrides = <PodOverride>{
      if (parentOverrides != null)
        for (final MapEntry<Pod, Pod> entry in parentOverrides.entries) (origin: entry.key, override: entry.value),
      ...overrides,
    };
    for (final PodOverride entry in combinedOverrides) {
      final Pod? previousOverride = _overrides[entry.origin];
      _pods
        ..[previousOverride]?.dispose()
        ..remove(previousOverride);
      _overrides[entry.origin] = entry.override;
    }
  }

  void dispose() {
    Future<void>(() {
      _pods
        ..forEach((_, PodElement pod) => pod.dispose())
        ..clear();
      _overrides.clear();
      _owner.dispose();
    });
  }

  @override
  bool operator ==(Object other) =>
      identical(this, other) ||
      other is Pods &&
          runtimeType == other.runtimeType &&
          _owner == other._owner &&
          _pods == other._pods &&
          _overrides == other._overrides;

  @override
  int get hashCode => _owner.hashCode ^ _pods.hashCode ^ _overrides.hashCode;

  @override
  String toString() => 'Ref($hashCode)';

  @visibleForTesting
  PodElement<U> element<U>(Pod<U> pod) => _resolvePodElement(pod);

  @visibleForTesting
  Future<void> pump() => Future<void>.microtask(() {});

  PodElement<U> _resolvePodElement<U>(Pod<U> pod, {bool shouldMount = true}) {
    switch ((_pods[pod], _overrides[pod])) {
      case (_, Pod<U> podOverride):
        return _resolvePodElement(podOverride, shouldMount: shouldMount);
      case (PodElement<U> element, _):
        return element.._attachTo(_owner);
      case _:
        final PodElement<U> element = pod.createElement()
          .._debugLabel = pod.debugLabel
          .._attachTo(_owner);
        _pods[pod] = element;
        if (shouldMount) {
          _mountPodElement(element);
        }
        return element;
    }
  }

  void _mountPodElement(PodElement element) => element.mount(
        Pods._(
          owner: element,
          pods: _pods,
          overrides: _overrides,
        ),
      );
}

/* End Pods Container */

/* Start Pod Types */

extension PodOverrideWith<T> on Pod<T> {
  PodOverride<T> overrideWith(PodFactory<T> factory) => (
        origin: this,
        override: _ProxyPod(factory, debugLabel: 'Override<$this>'),
      );
}

extension PodSelector<T> on NotifiablePod<T> {
  NotifiablePod<U> select<U>(ValueMapper<T, U> selector) => _SelectorPod(
        (Ref ref) => selector(ref.watch(this)),
        debugLabel: 'Selector<$this>',
      );
}

extension AsyncPodSelector<T, V> on AsyncPod<T, V> {
  NotifiablePod<Future<U>> selectAsync<U>(ValueMapper<T, U> selector) => _AsyncSelectorPod<T, U, V>(
        _factory,
        selector,
        debugLabel: 'AsyncSelector<$this>',
      );
}

@optionalTypeArgs
final class ValuePod<T> extends Pod<T> {
  const ValuePod(this._factory, {super.debugLabel});

  static ValueMapper<U, ValuePod<T>> family<T, U>(
    FamilyPodFactory<T, U> factory, {
    String? debugLabel,
  }) {
    return Pod._family(
      (U args) => ValuePod<T>(
        (Ref ref) => factory(ref, args),
        debugLabel: '${debugLabel ?? 'ValueFamily'}<$args>',
      ),
    );
  }

  final PodFactory<T> _factory;

  @override
  PodElement<T> createElement() => _ValuePodElement<T>(this);
}

@optionalTypeArgs
final class StatePod<T> extends ValuePod<T> with NotifiablePod<T> {
  StatePod(super._factory, {super.debugLabel});

  static ValueMapper<U, StatePod<T>> family<T, U>(
    FamilyPodFactory<T, U> factory, {
    String? debugLabel,
  }) {
    return Pod._family(
      (U args) => StatePod<T>(
        (Ref ref) => factory(ref, args),
        debugLabel: '${debugLabel ?? 'StateFamily'}<$args>',
      ),
    );
  }

  late final Pod<Notifier<T>> notifier = _ProxyPod(
    (Ref ref) {
      Pods pods = (ref as Pods);
      return Notifier(
        () => pods._resolvePodElement(this).state,
        (T state) => pods._resolvePodElement(this, shouldMount: false).state = state,
      );
    },
    debugLabel: 'Notifier<$this>',
  );
}

@optionalTypeArgs
final class FuturePod<T> extends _AsyncValuePod<T, FutureOr<T>> {
  FuturePod(super._factory, {super.debugLabel});

  static ValueMapper<U, FuturePod<T>> family<T, U>(
    FamilyPodFactory<FutureOr<T>, U> factory, {
    String? debugLabel,
  }) {
    return Pod._family(
      (U args) => FuturePod<T>(
        (Ref ref) => factory(ref, args),
        debugLabel: '${debugLabel ?? 'FutureFamily'}<$args>',
      ),
    );
  }

  @override
  PodElement<AsyncValue<T>> createElement() => _FuturePodElement(this);
}

@optionalTypeArgs
final class StreamPod<T> extends _AsyncValuePod<T, Stream<T>> {
  StreamPod(super._factory, {super.debugLabel});

  static ValueMapper<U, StreamPod<T>> family<T, U>(
    FamilyPodFactory<Stream<T>, U> factory, {
    String? debugLabel,
  }) {
    return Pod._family(
      (U args) => StreamPod<T>(
        (Ref ref) => factory(ref, args),
        debugLabel: '${debugLabel ?? 'StreamFamily'}<$args>',
      ),
    );
  }

  @override
  PodElement<AsyncValue<T>> createElement() => _StreamPodElement(this);
}

/* End Pod Types */

/* Start Internal Pod Types */

@optionalTypeArgs
final class _ProxyPod<T> extends Pod<T> with WatchablePod<T> {
  const _ProxyPod(this._builder, {required super.debugLabel});

  final PodFactory<T> _builder;

  @override
  PodElement<T> createElement() => _ProxyPodElement(_builder);
}

@optionalTypeArgs
final class _SelectorPod<T> extends _ProxyPod<T> with NotifiablePod<T> {
  const _SelectorPod(super._builder, {super.debugLabel});
}

@optionalTypeArgs
final class _AsyncSelectorPod<T, U, V> extends Pod<Future<U>> with NotifiablePod<Future<U>> {
  const _AsyncSelectorPod(this._factory, this._selector, {super.debugLabel});

  final PodFactory<V> _factory;
  final ValueMapper<T, U> _selector;

  @override
  PodElement<Future<U>> createElement() => _AsyncSelectorPodElement<T, U, V>(_factory, _selector);
}

@optionalTypeArgs
abstract base class _AsyncValuePod<T, U> extends Pod<AsyncValue<T>> implements AsyncPod<T, U> {
  _AsyncValuePod(this._factory, {super.debugLabel});

  @override
  final PodFactory<U> _factory;
  Completer<T>? _completer;

  @override
  late final WatchablePod<Future<T>> future = _AsyncSelectorPod<T, T, Future<T>>(
    (Ref ref) {
      if (_completer?.isCompleted == true) {
        _completer = null;
      }
      _completer ??= Completer<T>.sync();

      final T? value = ref.watch(this).previousValue;
      if (value != null) {
        _completer?.complete(value);
      }

      (ref as Pods)._owner._dependents.whereType<_AsyncValuePodElement>().forEach(
            (_) => _._refreshState(),
          );

      return _completer!.future;
    },
    (T value) => value,
    debugLabel: 'Future<$this>',
  );
}

/* End Internal Pod Types */

/* Start Pod Elements */

final class _AnonymousPodElement extends PodElement<Object> {
  _AnonymousPodElement() : super(Object());
}

@optionalTypeArgs
final class _ValuePodElement<T> extends PodElement<T> {
  _ValuePodElement(this._pod);

  final ValuePod<T> _pod;

  @override
  void mount(Ref ref) {
    _state = _pod._factory(ref);
    super.mount(ref);
  }
}

@optionalTypeArgs
base class _AsyncValuePodElement<T, U> extends PodElement<AsyncValue<T>> {
  _AsyncValuePodElement(_AsyncValuePod<T, U> pod) : _factory = pod._factory;

  final PodFactory<U> _factory;

  @override
  void mount(Ref ref) {
    _refreshState();
    super.mount(ref);
  }

  void _refreshState() => state = PendingAsyncValue(_state?.previousValue);
}

@optionalTypeArgs
final class _FuturePodElement<T> extends _AsyncValuePodElement<T, FutureOr<T>> {
  _FuturePodElement(super._pod);

  @override
  void mount(Ref ref) {
    Future<T>.value(_factory(ref)).then((T value) {
      state = SuccessfulAsyncValue(value);
    }).catchError((Object error, StackTrace stackTrace) {
      state = FailedAsyncValue(error, stackTrace);
    });
    super.mount(ref);
  }
}

@optionalTypeArgs
final class _StreamPodElement<T> extends _AsyncValuePodElement<T, Stream<T>> {
  _StreamPodElement(super._pod);

  StreamSubscription<T>? _subscription;

  @override
  void mount(Ref ref) {
    _subscription?.cancel();

    _subscription = _factory(ref).listen((T value) {
      state = SuccessfulAsyncValue(value);
    }, onError: (Object error, StackTrace stackTrace) {
      state = FailedAsyncValue(error, stackTrace);
    });
    super.mount(ref);
  }

  @override
  void dispose() {
    _subscription?.cancel();
    super.dispose();
  }
}

@optionalTypeArgs
final class _ProxyPodElement<T> extends PodElement<T> {
  _ProxyPodElement(this._builder);

  final PodFactory<T> _builder;

  @override
  void mount(Pods ref) {
    state = _builder(ref);
    super.mount(ref);
  }
}

@optionalTypeArgs
final class _AsyncSelectorPodElement<T, U, V> extends PodElement<Future<U>> {
  _AsyncSelectorPodElement(this._factory, this._selector);

  final PodFactory<V> _factory;
  final ValueMapper<T, U> _selector;

  final StreamController<T> _controller = StreamController<T>.broadcast();
  final Completer<U> _completer = Completer<U>();

  StreamSubscription<U>? _subscription;
  StreamSubscription<T>? _innerSubscription;

  @override
  void mount(Ref ref) {
    _subscription ??= _controller.stream.map(_selector).distinct().listen((U value) {
      state = Future<U>.value(value);
      if (!_completer.isCompleted) {
        _completer.complete(value);
      }
    });

    switch (_factory(ref)) {
      case T value:
        _controller.add(value);
      case Future<T> future:
        future.then(_controller.add);
      case Stream<T> stream:
        _innerSubscription?.cancel();
        _innerSubscription = stream.listen(_controller.add);
    }

    _state = _completer.future;
    super.mount(ref);
  }

  @override
  void dispose() {
    _state?.ignore();
    _innerSubscription?.cancel();
    _subscription?.cancel();
    _controller.close();
    super.dispose();
  }
}

/* End Pod Elements */

/* Start Data Classes */

@optionalTypeArgs
class Notifier<T> {
  Notifier(this._stateFn, this._listener);

  final ValueCallback<T> _stateFn;
  final ValueMapper<T, void> _listener;

  @protected
  T get state => _stateFn();

  set state(T value) => _listener(value);

  T update(ValueMapper<T, T> cb) => state = cb(_stateFn());

  @override
  String toString() => 'Notifier<${_stateFn()}>';
}

@optionalTypeArgs
sealed class AsyncValue<T> {}

@optionalTypeArgs
class PendingAsyncValue<T> implements AsyncValue<T> {
  const PendingAsyncValue(this.data);

  final T? data;

  @override
  bool operator ==(Object other) =>
      identical(this, other) || other is PendingAsyncValue<T> && runtimeType == other.runtimeType && data == other.data;

  @override
  int get hashCode => data.hashCode;

  @override
  String toString() => 'PendingAsyncValue${data != null ? '<$data>' : ''}';
}

@optionalTypeArgs
class SuccessfulAsyncValue<T> implements AsyncValue<T> {
  const SuccessfulAsyncValue(this.data);

  final T data;

  @override
  bool operator ==(Object other) =>
      identical(this, other) ||
      other is SuccessfulAsyncValue<T> && runtimeType == other.runtimeType && data == other.data;

  @override
  int get hashCode => data.hashCode;

  @override
  String toString() => 'SuccessfulAsyncValue<$data>';
}

@optionalTypeArgs
class FailedAsyncValue<T> implements AsyncValue<T> {
  const FailedAsyncValue(this.error, this.stackTrace);

  final Object error;
  final StackTrace stackTrace;

  @override
  bool operator ==(Object other) =>
      identical(this, other) ||
      other is FailedAsyncValue<T> && runtimeType == other.runtimeType && error == other.error;

  @override
  int get hashCode => error.hashCode;

  @override
  String toString() => 'FailedAsyncValue<$error>';
}

extension<T> on AsyncValue<T> {
  T? get previousValue => switch (this) {
        PendingAsyncValue<T> state => state.data,
        SuccessfulAsyncValue<T> state => state.data,
        FailedAsyncValue<T> _ => null,
      };
}

/* End Data Classes */