Avoid missing streams when relistening to broadcast StreamGroup (dart…

…-archive/async#184)

In a broadcast `StreamGroup` when all listeners have canceled the inner
subscriptions are kept only for single subscriber streams. When a new
listener is added and subscriptions need to be recreated for inner
broadcast streams the presence of a single subscriber subscription in
the collection of subscriptions would cause the listen to bail out early
and fail to subscribe to later broadcast streams. Change `return` to
`continue` to keep going with the rest of the streams.

pkgs/async/CHANGELOG.md

@@ -4,6 +4,9 @@
   easier to implement custom sinks.
 * Improve performance for `ChunkedStreamReader` by creating fewer internal
   sublists and specializing to create views for `Uint8List` chunks.
+* Don't ignore broadcast streams added to a `StreamGroup` that doesn't have an
+  active listener but previously had listeners and contains a single
+  subscription inner stream.
 
 ## 2.7.0
 

pkgs/async/lib/src/stream_group.dart

@@ -192,7 +192,7 @@ class StreamGroup<T> implements Sink<Stream<T>> {
       // 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 (entry.value != null) return;
+      if (entry.value != null) continue;
 
       var stream = entry.key;
       try {

pkgs/async/test/stream_group_test.dart

@@ -417,6 +417,24 @@ void main() {
       expect(controller.hasListener, isTrue);
     });
 
+    test(
+        'listens on streams that follow single-subscription streams when '
+        'relistening after a cancel', () async {
+      var controller1 = StreamController<String>();
+      streamGroup.add(controller1.stream);
+      streamGroup.stream.listen(null).cancel();
+
+      var controller2 = StreamController<String>();
+      streamGroup.add(controller2.stream);
+
+      var emitted = <String>[];
+      streamGroup.stream.listen(emitted.add);
+      controller1.add('one');
+      controller2.add('two');
+      await flushMicrotasks();
+      expect(emitted, ['one', 'two']);
+    });
+
     test('never cancels single-subscription streams', () async {
       var subscription = streamGroup.stream.listen(null);