Properly close watcher streams during tests.

pkg/barback/lib/src/utils.dart

 // Copyright (c) 2013, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
 library barback.utils;
 
 import 'dart:async';
 
 /// A pair of values.
 class Pair<E, F> {
@@ skipping 163 matching lines @@
   // invoking this method.
   return new Future.delayed(Duration.ZERO, () => pumpEventQueue(times - 1));
 }
 
 /// Like `new Future`, but avoids issue 11911 by using `new Future.value` under
 /// the covers.
 // TODO(jmesserly): doc comment changed to due 14601.
 Future newFuture(callback()) => new Future.value().then((_) => callback());
 
 /// Returns a buffered stream that will emit the same values as the stream
-/// returned by [future] once [future] completes. If [future] completes to an
-/// error, the return value will emit that error and then close.
-Stream futureStream(Future<Stream> future) {
-  var controller = new StreamController(sync: true);
-  future.then((stream) {
-    stream.listen(
-        controller.add,
-        onError: controller.addError,
-        onDone: controller.close);
-  }).catchError((e, stackTrace) {
+/// returned by [future] once [future] completes.
+///
+/// If [future] completes to an error, the return value will emit that error and
+/// then close.
+///
+/// If [broadcast] is true, a broadcast stream is returned. This assumes that
+/// the stream returned by [future] will be a broadcast stream as well.
+/// [broadcast] defaults to false.
+Stream futureStream(Future<Stream> future, {bool broadcast: false}) {
+  var subscription;
+  var controller;
+
+  future = future.catchError((e, stackTrace) {
+    if (controller == null) return;
     controller.addError(e, stackTrace);
     controller.close();
+    controller = null;
   });
+
+  onListen() {
+    future.then((stream) {
+      if (controller == null) return;
+      subscription = stream.listen(
+          controller.add,
+          onError: controller.addError,
+          onDone: controller.close);
+    });
+  }
+
+  onCancel() {
+    if (subscription != null) subscription.cancel();
+    subscription = null;
+    controller = null;
+  }
+
+  if (broadcast) {
+    controller = new StreamController.broadcast(
+        sync: true, onListen: onListen, onCancel: onCancel);
+  } else {
+    controller = new StreamController(
+        sync: true, onListen: onListen, onCancel: onCancel);
+  }
   return controller.stream;
 }
 
 /// Returns a [Stream] that will emit the same values as the stream returned by
 /// [callback].
 ///
 /// [callback] will only be called when the returned [Stream] gets a subscriber.
 Stream callbackStream(Stream callback()) {
   var subscription;
   var controller;
   controller = new StreamController(onListen: () {
     subscription = callback().listen(controller.add,
         onError: controller.addError,
         onDone: controller.close);
   },
       onCancel: () => subscription.cancel(),
       onPause: () => subscription.pause(),
       onResume: () => subscription.resume(),
       sync: true);
   return controller.stream;
 }