refactor bloc tools to its own package

packages/bloc_tools/.gitignore

@@ -0,0 +1,7 @@
+# https://dart.dev/guides/libraries/private-files
+# Created by `dart pub`
+.dart_tool/
+
+# Avoid committing pubspec.lock for library packages; see
+# https://dart.dev/guides/libraries/private-files#pubspeclock.
+pubspec.lock

packages/bloc_tools/analysis_options.yaml

@@ -0,0 +1,15 @@
+include: package:lint_hard/all.yaml
+analyzer:
+  errors:
+    invalid_annotation_target: ignore
+  exclude:
+    - '**/*.g.dart'
+    - '**/*.freezed.dart'
+    - '**/*.pb.dart'
+    - '**/*.pbenum.dart'
+    - '**/*.pbjson.dart'
+    - '**/*.pbserver.dart'
+linter:
+  rules:
+    unawaited_futures: true
+    avoid_positional_boolean_parameters: false

packages/bloc_tools/example/bloc_tools_example.dart

@@ -0,0 +1,6 @@
+// import 'package:bloc_tools/bloc_tools.dart';
+
+// void main() {
+//   var awesome = Awesome();
+//   print('awesome: ${awesome.isAwesome}');
+// }

packages/bloc_tools/lib/bloc_tools.dart

@@ -0,0 +1,11 @@
+/// BLoC Tools
+library;
+
+export 'src/async_transformer_cubit.dart';
+export 'src/bloc_busy_wrapper.dart';
+export 'src/bloc_map_cubit.dart';
+export 'src/bloc_tools_extension.dart';
+export 'src/future_cubit.dart';
+export 'src/state_follower.dart';
+export 'src/stream_wrapper_cubit.dart';
+export 'src/transformer_cubit.dart';

packages/bloc_tools/lib/src/async_transformer_cubit.dart

@@ -0,0 +1,42 @@
+import 'dart:async';
+
+import 'package:async_tools/async_tools.dart';
+import 'package:bloc/bloc.dart';
+
+class AsyncTransformerCubit<T, S> extends Cubit<AsyncValue<T>> {
+  AsyncTransformerCubit(this.input, {required this.transform})
+      : super(const AsyncValue.loading()) {
+    _asyncTransform(input.state);
+    _subscription = input.stream.listen(_asyncTransform);
+  }
+  void _asyncTransform(AsyncValue<S> newInputState) {
+    _singleStateProcessor.updateState(newInputState, closure: (newState) async {
+      // Emit the transformed state
+      try {
+        if (newState is AsyncLoading<S>) {
+          emit(const AsyncValue.loading());
+        } else if (newState is AsyncError<S>) {
+          emit(AsyncValue.error(newState.error, newState.stackTrace));
+        } else {
+          final transformedState = await transform(newState.data!.value);
+          emit(transformedState);
+        }
+      } on Exception catch (e, st) {
+        emit(AsyncValue.error(e, st));
+      }
+    });
+  }
+
+  @override
+  Future<void> close() async {
+    await _subscription.cancel();
+    await input.close();
+    await super.close();
+  }
+
+  Cubit<AsyncValue<S>> input;
+  final SingleStateProcessor<AsyncValue<S>> _singleStateProcessor =
+      SingleStateProcessor();
+  Future<AsyncValue<T>> Function(S) transform;
+  late final StreamSubscription<AsyncValue<S>> _subscription;
+}

packages/bloc_tools/lib/src/bloc_busy_wrapper.dart

@@ -0,0 +1,54 @@
+import 'dart:async';
+
+import 'package:bloc/bloc.dart';
+import 'package:equatable/equatable.dart';
+import 'package:meta/meta.dart';
+import 'package:mutex/mutex.dart';
+
+@immutable
+class BlocBusyState<S> extends Equatable {
+  const BlocBusyState(this.state) : busy = false;
+  const BlocBusyState._busy(this.state) : busy = true;
+  final bool busy;
+  final S state;
+
+  @override
+  List<Object?> get props => [busy, state];
+}
+
+mixin BlocBusyWrapper<S> on BlocBase<BlocBusyState<S>> {
+  Future<T> busy<T>(Future<T> Function(void Function(S) emit) closure) async =>
+      _mutex.protect(() async {
+        void busyemit(S state) {
+          changedState = state;
+        }
+
+        // Turn on busy state
+        emit(BlocBusyState._busy(state.state));
+
+        // Run the closure
+        final out = await closure(busyemit);
+
+        // If the closure did one or more 'busy emits' then
+        // take the most recent one and emit it for real
+        final finalState = changedState;
+        if (finalState != null && finalState != state.state) {
+          emit(BlocBusyState._busy(finalState));
+        } else {
+          emit(BlocBusyState._busy(state.state));
+        }
+
+        return out;
+      });
+
+  void changeState(S state) {
+    if (_mutex.isLocked) {
+      changedState = state;
+    } else {
+      emit(BlocBusyState(state));
+    }
+  }
+
+  final Mutex _mutex = Mutex();
+  S? changedState;
+}

packages/bloc_tools/lib/src/bloc_map_cubit.dart

@@ -0,0 +1,112 @@
+import 'dart:async';
+
+import 'package:async_tools/async_tools.dart';
+import 'package:bloc/bloc.dart';
+import 'package:fast_immutable_collections/fast_immutable_collections.dart';
+
+typedef BlocMapState<K, S> = IMap<K, S>;
+
+class _ItemEntry<S, B> {
+  _ItemEntry({required this.bloc, required this.subscription});
+  final B bloc;
+  final StreamSubscription<S> subscription;
+}
+
+// Streaming container cubit that is a map from some immutable key
+// to a some other cubit's output state. Output state for this container
+// cubit is an immutable map of the key to the output state of the contained
+// cubits.
+//
+// K = Key type for the bloc map, used to look up some mapped cubit
+// S = State type for the value, keys will look up values of this type
+// B = Bloc/cubit type for the value, output states of type S
+abstract class BlocMapCubit<K, S, B extends BlocBase<S>>
+    extends Cubit<BlocMapState<K, S>> {
+  BlocMapCubit()
+      : _entries = {},
+        _tagLock = AsyncTagLock(),
+        super(IMap<K, S>());
+
+  @override
+  Future<void> close() async {
+    await _entries.values.map((e) => e.subscription.cancel()).wait;
+    await _entries.values.map((e) => e.bloc.close()).wait;
+    await super.close();
+  }
+
+  Future<void> add(MapEntry<K, B> Function() create) {
+    // Create new element
+    final newElement = create();
+    final key = newElement.key;
+    final bloc = newElement.value;
+
+    return _tagLock.protect(key, closure: () async {
+      // Remove entry with the same key if it exists
+      await _internalRemove(key);
+
+      // Add entry with this key
+      _entries[key] = _ItemEntry(
+          bloc: bloc,
+          subscription: bloc.stream.listen((data) {
+            // Add sub-cubit's state to the map state
+            emit(state.add(key, data));
+          }));
+
+      emit(state.add(key, bloc.state));
+    });
+  }
+
+  Future<void> addState(K key, S value) =>
+      _tagLock.protect(key, closure: () async {
+        // Remove entry with the same key if it exists
+        await _internalRemove(key);
+
+        emit(state.add(key, value));
+      });
+
+  Future<void> _internalRemove(K key) async {
+    final sub = _entries.remove(key);
+    if (sub != null) {
+      await sub.subscription.cancel();
+      await sub.bloc.close();
+    }
+  }
+
+  Future<void> remove(K key) => _tagLock.protect(key, closure: () async {
+        await _internalRemove(key);
+        emit(state.remove(key));
+      });
+
+  R operate<R>(K key, {required R Function(B bloc) closure}) {
+    final bloc = _entries[key]!.bloc;
+    return closure(bloc);
+  }
+
+  R? tryOperate<R>(K key, {required R Function(B bloc) closure}) {
+    final entry = _entries[key];
+    if (entry == null) {
+      return null;
+    }
+    return closure(entry.bloc);
+  }
+
+  Future<R> operateAsync<R>(K key,
+          {required Future<R> Function(B bloc) closure}) =>
+      _tagLock.protect(key, closure: () async {
+        final bloc = _entries[key]!.bloc;
+        return closure(bloc);
+      });
+
+  Future<R?> tryOperateAsync<R>(K key,
+          {required Future<R> Function(B bloc) closure}) =>
+      _tagLock.protect(key, closure: () async {
+        final entry = _entries[key];
+        if (entry == null) {
+          return null;
+        }
+        return closure(entry.bloc);
+      });
+
+  final Map<K, _ItemEntry<S, B>> _entries;
+  final AsyncTagLock<K> _tagLock;
+}

packages/bloc_tools/lib/src/bloc_tools_extension.dart

@@ -0,0 +1,12 @@
+import 'package:bloc/bloc.dart';
+
+mixin BlocTools<State> on BlocBase<State> {
+  void withStateListen(void Function(State event)? onData,
+      {Function? onError, void Function()? onDone, bool? cancelOnError}) {
+    if (onData != null) {
+      onData(state);
+    }
+    stream.listen(onData,
+        onError: onError, onDone: onDone, cancelOnError: cancelOnError);
+  }
+}

packages/bloc_tools/lib/src/future_cubit.dart

@@ -0,0 +1,15 @@
+import 'dart:async';
+
+import 'package:async_tools/async_tools.dart';
+import 'package:bloc/bloc.dart';
+
+abstract class FutureCubit<State> extends Cubit<AsyncValue<State>> {
+  FutureCubit(Future<State> fut) : super(const AsyncValue.loading()) {
+    unawaited(fut.then((value) {
+      emit(AsyncValue.data(value));
+      // ignore: avoid_types_on_closure_parameters
+    }, onError: (Object e, StackTrace stackTrace) {
+      emit(AsyncValue.error(e, stackTrace));
+    }));
+  }
+}

packages/bloc_tools/lib/src/state_follower.dart

@@ -0,0 +1,58 @@
+import 'dart:async';
+
+import 'package:async_tools/async_tools.dart';
+import 'package:fast_immutable_collections/fast_immutable_collections.dart';
+
+// Mixin that automatically keeps two blocs/cubits in sync with each other
+// Useful for having a BlocMapCubit 'follow' the state of another input cubit.
+// As the state of the input cubit changes, the BlocMapCubit can add/remove
+// mapped Cubits that automatically process the input state reactively.
+//
+// S = Input state type
+// K = Key derived from elements of input state
+// V = Value derived from elements of input state
+abstract mixin class StateFollower<S extends Object, K, V> {
+  void follow({
+    required S initialInputState,
+    required Stream<S> stream,
+  }) {
+    //
+    _lastInputStateMap = getStateMap(initialInputState);
+    _subscription = stream.listen(_updateFollow);
+  }
+
+  Future<void> close() async {
+    await _subscription.cancel();
+  }
+
+  IMap<K, V> getStateMap(S state);
+  Future<void> removeFromState(K key);
+  Future<void> updateState(K key, V value);
+
+  void _updateFollow(S newInputState) {
+    _singleStateProcessor.updateState(getStateMap(newInputState),
+        closure: (newStateMap) async {
+      for (final k in _lastInputStateMap.keys) {
+        if (!newStateMap.containsKey(k)) {
+          // deleted
+          await removeFromState(k);
+        }
+      }
+      for (final newEntry in newStateMap.entries) {
+        final v = _lastInputStateMap.get(newEntry.key);
+        if (v == null || v != newEntry.value) {
+          // added or changed
+          await updateState(newEntry.key, newEntry.value);
+        }
+      }
+
+      // Keep this state map for the next time
+      _lastInputStateMap = newStateMap;
+    });
+  }
+
+  late IMap<K, V> _lastInputStateMap;
+  final SingleStateProcessor<IMap<K, V>> _singleStateProcessor =
+      SingleStateProcessor();
+  late final StreamSubscription<S> _subscription;
+}

packages/bloc_tools/lib/src/stream_wrapper_cubit.dart

@@ -0,0 +1,25 @@
+import 'dart:async';
+
+import 'package:async_tools/async_tools.dart';
+import 'package:bloc/bloc.dart';
+
+abstract class StreamWrapperCubit<State> extends Cubit<AsyncValue<State>> {
+  StreamWrapperCubit(Stream<State> stream, {State? defaultState})
+      : super(defaultState != null
+            ? AsyncValue.data(defaultState)
+            : const AsyncValue.loading()) {
+    _subscription = stream.listen((event) => emit(AsyncValue.data(event)),
+        // ignore: avoid_types_on_closure_parameters
+        onError: (Object error, StackTrace stackTrace) {
+      emit(AsyncValue.error(error, stackTrace));
+    });
+  }
+
+  @override
+  Future<void> close() async {
+    await _subscription.cancel();
+    await super.close();
+  }
+
+  late final StreamSubscription<State> _subscription;
+}

packages/bloc_tools/lib/src/transformer_cubit.dart

@@ -0,0 +1,21 @@
+import 'dart:async';
+
+import 'package:bloc/bloc.dart';
+
+class TransformerCubit<T, S> extends Cubit<T> {
+  TransformerCubit(this.input, {required this.transform})
+      : super(transform(input.state)) {
+    _subscription = input.stream.listen((event) => emit(transform(event)));
+  }
+
+  @override
+  Future<void> close() async {
+    await _subscription.cancel();
+    await input.close();
+    await super.close();
+  }
+
+  Cubit<S> input;
+  T Function(S) transform;
+  late final StreamSubscription<S> _subscription;
+}

packages/bloc_tools/pubspec.yaml

@@ -0,0 +1,24 @@
+name: bloc_tools
+description: A starting point for Dart libraries or applications.
+version: 1.0.0
+publish_to: none
+
+environment:
+  sdk: '>=3.2.0 <4.0.0'
+
+dependencies:
+  async_tools:
+    path: ../async_tools
+  bloc: ^8.1.3
+  equatable: ^2.0.5
+  fast_immutable_collections: ^10.1.1
+  freezed_annotation: ^2.4.1
+  meta: ^1.10.0
+  mutex:
+    path: ../mutex
+
+dev_dependencies:
+  build_runner: ^2.4.8
+  freezed: ^2.4.7
+  lint_hard: ^4.0.0
+  test: ^1.25.2

packages/bloc_tools/test/bloc_tools_test.dart

@@ -0,0 +1,16 @@
+// import 'package:bloc_tools/bloc_tools.dart';
+// import 'package:test/test.dart';
+
+// void main() {
+//   group('A group of tests', () {
+//     final awesome = Awesome();
+
+//     setUp(() {
+//       // Additional setup goes here.
+//     });
+
+//     test('First Test', () {
+//       expect(awesome.isAwesome, isTrue);
+//     });
+//   });
+// }