Fix: StreamGroup close() completes when streams close without being removed (dart-lang#372)

Author: suojae

pkgs/async/CHANGELOG.md

@@ -1,6 +1,7 @@
 ## 2.13.0
 
 - Fix type check and cast in SubscriptionStream's cancelOnError wrapper
+- Fix `StreamGroup.broadcast().close()` to properly complete when all streams in the group close without being explicitly removed.
 
 ## 2.12.0
 

pkgs/async/lib/src/stream_group.dart

@@ -289,7 +289,23 @@ class StreamGroup<T> implements Sink<Stream<T>> {
     if (_closed) return _controller.done;
 
     _closed = true;
-    if (_subscriptions.isEmpty) _controller.close();
+
+    if (_subscriptions.isEmpty) {
+      _onIdleController?.add(null);
+      _onIdleController?.close();
+      _controller.close();
+      return _controller.done;
+    }
+
+    if (_controller.stream.isBroadcast) {
+      for (var entry in _subscriptions.entries.where((e) => e.value == null)) {
+        try {
+          _subscriptions[entry.key] = _listenToStream(entry.key);
+        } catch (_) {
+          _subscriptions.remove(entry.key);
+        }
+      }
+    }
 
     return _controller.done;
   }

pkgs/async/test/stream_group_test.dart

@@ -491,6 +491,22 @@ void main() {
       controller.add('first');
       expect(streamGroup.close(), completes);
     });
+
+    test('completes close() when streams close without being removed',
+        () async {
+      var controller = StreamController.broadcast();
+      var group = StreamGroup.broadcast();
+      group.add(controller.stream);
+      var closeCompleted = false;
+      group.close().then((_) => closeCompleted = true);
+
+      await flushMicrotasks();
+      expect(closeCompleted, isFalse);
+
+      await controller.close();
+      await flushMicrotasks();
+      expect(closeCompleted, isTrue);
+    });
   });
 
   group('regardless of type', () {