Fix a StreamGroup bug when a component stream's listen() throws (#173)

nex3 · web-flow · commit cde00b89bd3c · 2021-04-29T14:49:04.000-07:00
This would put the StreamGroup into an inconsistent state where it would believe itself to be active, but only some streams would have subscriptions. This was exacerbated by dart-lang/sdk#45815, which meant that even though _onListen threw an error a StreamSubscription was created and returned, so further callbacks could still be called. Now instead of going into an inconsistent state, the StreamGroup simply cancels itself.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,3 +1,8 @@
+## 2.6.1
+
+* When `StreamGroup.stream.listen()` is called, gracefully handle component
+  streams throwing errors when their `Stream.listen()` methods are called.
+
 ## 2.6.0
 
 * Add a `StreamCloser` class, which is a `StreamTransformer` that allows the
diff --git a/lib/src/stream_group.dart b/lib/src/stream_group.dart
@@ -4,6 +4,8 @@
 
 import 'dart:async';
 
+import 'package:collection/collection.dart';
+
 /// A collection of streams whose events are unified and sent through a central
 /// stream.
 ///
@@ -185,13 +187,24 @@ class StreamGroup<T> implements Sink<Stream<T>> {
   /// This is called for both single-subscription and broadcast groups.
   void _onListen() {
     _state = _StreamGroupState.listening;
-    _subscriptions.forEach((stream, subscription) {
+
+    for (var entry in _subscriptions.entries.toList()) {
       // If this is a broadcast group and this isn't the first time it's been
       // listened to, there may still be some subscriptions to
       // single-subscription streams.
-      if (subscription != null) return;
-      _subscriptions[stream] = _listenToStream(stream);
-    });
+      if (entry.value != null) return;
+
+      var stream = entry.key;
+      try {
+        _subscriptions[stream] = _listenToStream(stream);
+      } catch (error) {
+        // If [Stream.listen] throws a synchronous error (for example because
+        // the stream has already been listened to), cancel all subscriptions
+        // and rethrow the error.
+        _onCancel()?.catchError((_) {});
+        rethrow;
+      }
+    }
   }
 
   /// A callback called when [stream] is paused.
@@ -216,8 +229,17 @@ class StreamGroup<T> implements Sink<Stream<T>> {
   Future? _onCancel() {
     _state = _StreamGroupState.canceled;
 
-    var futures = _subscriptions.values
-        .map((subscription) => subscription!.cancel())
+    var futures = _subscriptions.entries
+        .map((entry) {
+          var subscription = entry.value;
+          if (subscription != null) return subscription.cancel();
+          try {
+            return entry.key.listen(null).cancel();
+          } catch (_) {
+            return null;
+          }
+        })
+        .whereNotNull()
         .toList();
 
     _subscriptions.clear();
diff --git a/pubspec.yaml b/pubspec.yaml
@@ -1,5 +1,5 @@
 name: async
-version: 2.6.0
+version: 2.6.1
 
 description: Utility functions and classes related to the 'dart:async' library.
 repository: https://github.com/dart-lang/async
diff --git a/test/stream_group_test.dart b/test/stream_group_test.dart
@@ -246,6 +246,69 @@ void main() {
         expect(fired, isTrue);
       });
     });
+
+    group('when listen() throws an error', () {
+      late Stream<String> alreadyListened;
+      setUp(() {
+        alreadyListened = Stream.value('foo')..listen(null);
+      });
+
+      group('listen()', () {
+        test('rethrows that error', () {
+          streamGroup.add(alreadyListened);
+
+          // We can't use expect(..., throwsStateError) here bceause of
+          // dart-lang/sdk#45815.
+          runZonedGuarded(
+              () => streamGroup.stream.listen(expectAsync1((_) {}, count: 0)),
+              expectAsync2((error, _) => expect(error, isStateError)));
+        });
+
+        test('cancels other subscriptions', () async {
+          var firstCancelled = false;
+          var first =
+              StreamController<String>(onCancel: () => firstCancelled = true);
+          streamGroup.add(first.stream);
+
+          streamGroup.add(alreadyListened);
+
+          var lastCancelled = false;
+          var last =
+              StreamController<String>(onCancel: () => lastCancelled = true);
+          streamGroup.add(last.stream);
+
+          runZonedGuarded(() => streamGroup.stream.listen(null), (_, __) {});
+
+          expect(firstCancelled, isTrue);
+          expect(lastCancelled, isTrue);
+        });
+
+        // There really shouldn't even be a subscription here, but due to
+        // dart-lang/sdk#45815 there is.
+        group('canceling the subscription is a no-op', () {
+          test('synchronously', () {
+            streamGroup.add(alreadyListened);
+
+            var subscription = runZonedGuarded(
+                () => streamGroup.stream.listen(null),
+                expectAsync2((_, __) {}, count: 1));
+
+            expect(subscription!.cancel(), completes);
+          });
+
+          test('asynchronously', () async {
+            streamGroup.add(alreadyListened);
+
+            var subscription = runZonedGuarded(
+                () => streamGroup.stream.listen(null),
+                expectAsync2((_, __) {}, count: 1));
+
+            await pumpEventQueue();
+            expect(subscription!.cancel(), completes);
+          });
+        });
+      });
+    });
   });
 
   group('broadcast', () {
