neon: cleanup StreamListenable

packages/neon/neon/lib/src/router.dart

@@ -23,7 +23,7 @@ class AppRouter extends GoRouter {
     required final AccountsBloc accountsBloc,
   }) : super(
           debugLogDiagnostics: kDebugMode,
-          refreshListenable: StreamListenable.behaviorSubject(accountsBloc.activeAccount),
+          refreshListenable: StreamListenable(accountsBloc.activeAccount),
           navigatorKey: navigatorKey,
           initialLocation: const HomeRoute().location,
           redirect: (final context, final state) {

packages/neon/neon/lib/src/utils/stream_listenable.dart

@@ -9,32 +9,36 @@ import 'package:rxdart/rxdart.dart';
 /// Objects need to be manually disposed.
 class StreamListenable extends ChangeNotifier {
   /// Listenable Stream
-  ///
-  /// Implementation for all types of [Stream]s.
-  /// For an implementation tailored towards [BehaviorSubject] have a look at [StreamListenable.behaviorSubject].
   StreamListenable(final Stream<dynamic> stream) {
-    notifyListeners();
+    if (stream is! BehaviorSubject) {
-
-    _subscription = stream.asBroadcastStream().listen((final value) {
       notifyListeners();
-    });
+    }
+
+    addSubscription(stream);
   }
 
-  /// Listenable BehaviorSubject
+  /// Listenable for multiple Streams.
   ///
-  /// Implementation for a [BehaviorSubject]. It ensures to not unececcary notify listeners.
+  /// Notifies it's listeners on every event emitted by any of the streams.
-  /// For an implementation tailored towards otnher kinds of [Stream] have a look at [StreamListenable].
+  StreamListenable.multiListenable(final Iterable<Stream<dynamic>> streams) {
-  StreamListenable.behaviorSubject(final BehaviorSubject<dynamic> subject) {
+    streams.forEach(addSubscription);
-    _subscription = subject.listen((final value) {
+  }
-      notifyListeners();
+
-    });
+  void addSubscription(final Stream<dynamic> stream) {
+    _subscriptions.add(
+      stream.asBroadcastStream().listen((final _) {
+        notifyListeners();
+      }),
+    );
   }
 
-  late final StreamSubscription<dynamic> _subscription;
+  final List<StreamSubscription<dynamic>> _subscriptions = [];
 
   @override
   void dispose() {
-    unawaited(_subscription.cancel());
+    for (final subscription in _subscriptions) {
+      unawaited(subscription.cancel());
+    }
 
     super.dispose();
   }