Properly close watcher streams during tests.

pkg/barback/lib/src/utils.dart

Index: pkg/barback/lib/src/utils.dart
diff --git a/pkg/barback/lib/src/utils.dart b/pkg/barback/lib/src/utils.dart
index 38a596eaa0f7bd5da1acd6ccb636b9c796db841d..51d98eef0b0b0aa89611bfc4ebaa2edcad23fd81 100644
--- a/pkg/barback/lib/src/utils.dart
+++ b/pkg/barback/lib/src/utils.dart
@@ -181,19 +181,48 @@ Future pumpEventQueue([int times=20]) {
 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;
 }