Refactor tracking of multiple stream subscriptions (dart-lang/stream_transform#92)

Instead of using a variable with a bigger scope than the `onListen`
handler, and assigning to null in between calls to the callback, scope
the variable down. Remove the subscriptions when they are no longer
needed and use the size of the list as the subscription count rather
than tracking in a separate int variable.

Refactor a `..addAll` and a `.map` to use a list literal with a for-loop
element.

Make more local variables final since the details of which are changed
is very relevant to behavior in this code.

pkgs/stream_transform/lib/src/combine_latest.dart

diff --git a/pkgs/stream_transform/lib/src/combine_latest.dart b/pkgs/stream_transform/lib/src/combine_latest.dart
index bacc7d6..7fee05d 100644
--- a/pkgs/stream_transform/lib/src/combine_latest.dart
+++ b/pkgs/stream_transform/lib/src/combine_latest.dart
@@ -192,22 +192,21 @@
   _CombineLatestAll(this._others);
 
   @override
-  Stream<List<T>> bind(Stream<T> source) {
-    final controller = source.isBroadcast
+  Stream<List<T>> bind(Stream<T> first) {
+    final controller = first.isBroadcast
         ? StreamController<List<T>>.broadcast(sync: true)
         : StreamController<List<T>>(sync: true);
 
-    var allStreams = [source]..addAll(_others);
-    if (source.isBroadcast) {
-      allStreams = allStreams
-          .map((s) => s.isBroadcast ? s : s.asBroadcastStream())
-          .toList();
-    }
-
-    List<StreamSubscription<T>> subscriptions;
+    final allStreams = [
+      first,
+      for (final other in _others)
+        !first.isBroadcast || other.isBroadcast
+            ? other
+            : other.asBroadcastStream(),
+    ];
 
     controller.onListen = () {
-      assert(subscriptions == null);
+      final subscriptions = <StreamSubscription<T>>[];
 
       final latestData = List<T>(allStreams.length);
       final hasEmitted = <int>{};
@@ -219,35 +218,39 @@
         }
       }
 
-      var activeStreamCount = 0;
-      subscriptions = allStreams.map((stream) {
-        final index = activeStreamCount;
-        activeStreamCount++;
-        return stream.listen((data) => handleData(index, data),
-            onError: controller.addError, onDone: () {
-          if (--activeStreamCount <= 0 || !hasEmitted.contains(index)) {
+      var streamId = 0;
+      for (final stream in allStreams) {
+        final index = streamId;
+
+        final subscription = stream.listen((data) => handleData(index, data),
+            onError: controller.addError);
+        subscription.onDone(() {
+          assert(subscriptions.contains(subscription));
+          subscriptions.remove(subscription);
+          if (subscriptions.isEmpty || !hasEmitted.contains(index)) {
             controller.close();
           }
         });
-      }).toList();
-      if (!source.isBroadcast) {
+        subscriptions.add(subscription);
+
+        streamId++;
+      }
+      if (!first.isBroadcast) {
         controller
           ..onPause = () {
-            for (var subscription in subscriptions) {
+            for (final subscription in subscriptions) {
               subscription.pause();
             }
           }
           ..onResume = () {
-            for (var subscription in subscriptions) {
+            for (final subscription in subscriptions) {
               subscription.resume();
             }
           };
       }
       controller.onCancel = () {
-        final toCancel = subscriptions;
-        subscriptions = null;
-        if (activeStreamCount <= 0) return null;
-        return Future.wait(toCancel.map((s) => s.cancel()));
+        if (subscriptions.isEmpty) return null;
+        return Future.wait(subscriptions.map((s) => s.cancel()));
       };
     };
     return controller.stream;

pkgs/stream_transform/lib/src/merge.dart

diff --git a/pkgs/stream_transform/lib/src/merge.dart b/pkgs/stream_transform/lib/src/merge.dart
index 5d2c864..99d19c9 100644
--- a/pkgs/stream_transform/lib/src/merge.dart
+++ b/pkgs/stream_transform/lib/src/merge.dart
@@ -87,47 +87,45 @@
 
   @override
   Stream<T> bind(Stream<T> first) {
-    var controller = first.isBroadcast
+    final controller = first.isBroadcast
         ? StreamController<T>.broadcast(sync: true)
         : StreamController<T>(sync: true);
 
-    var allStreams = [first]..addAll(_others);
-    if (first.isBroadcast) {
-      allStreams = allStreams
-          .map((s) => s.isBroadcast ? s : s.asBroadcastStream())
-          .toList();
-    }
-
-    List<StreamSubscription<T>> subscriptions;
+    final allStreams = [
+      first,
+      for (final other in _others)
+        !first.isBroadcast || other.isBroadcast
+            ? other
+            : other.asBroadcastStream(),
+    ];
 
     controller.onListen = () {
-      assert(subscriptions == null);
-      var activeStreamCount = 0;
-      subscriptions = allStreams.map((stream) {
-        activeStreamCount++;
-        return stream.listen(controller.add, onError: controller.addError,
-            onDone: () {
-          if (--activeStreamCount <= 0) controller.close();
+      final subscriptions = <StreamSubscription<T>>[];
+      for (final stream in allStreams) {
+        final subscription =
+            stream.listen(controller.add, onError: controller.addError);
+        subscription.onDone(() {
+          subscriptions.remove(subscription);
+          if (subscriptions.isEmpty) controller.close();
         });
-      }).toList();
+        subscriptions.add(subscription);
+      }
       if (!first.isBroadcast) {
         controller
           ..onPause = () {
-            for (var subscription in subscriptions) {
+            for (final subscription in subscriptions) {
               subscription.pause();
             }
           }
           ..onResume = () {
-            for (var subscription in subscriptions) {
+            for (final subscription in subscriptions) {
               subscription.resume();
             }
           };
       }
       controller.onCancel = () {
-        var toCancel = subscriptions;
-        subscriptions = null;
-        if (activeStreamCount <= 0) return null;
-        return Future.wait(toCancel.map((s) => s.cancel()));
+        if (subscriptions.isEmpty) return null;
+        return Future.wait(subscriptions.map((s) => s.cancel()));
       };
     };
     return controller.stream;
@@ -150,14 +148,12 @@
         final subscription =
             inner.listen(controller.add, onError: controller.addError);
         subscription.onDone(() {
-          assert(subscriptions.contains(subscription));
           subscriptions.remove(subscription);
           if (subscriptions.isEmpty) controller.close();
         });
         subscriptions.add(subscription);
       }, onError: controller.addError);
       outerSubscription.onDone(() {
-        assert(subscriptions.contains(outerSubscription));
         subscriptions.remove(outerSubscription);
         if (subscriptions.isEmpty) controller.close();
       });