Properly close watcher streams during tests.

pkg/watcher/lib/src/utils.dart

Index: pkg/watcher/lib/src/utils.dart
diff --git a/pkg/watcher/lib/src/utils.dart b/pkg/watcher/lib/src/utils.dart
index 66c0f09cf7b2ea121f9dad780b8a7321d947a076..a64575b9f796eab3536f792c95ea5c397822c20e 100644
--- a/pkg/watcher/lib/src/utils.dart
+++ b/pkg/watcher/lib/src/utils.dart
@@ -27,17 +27,44 @@ Set unionAll(Iterable<Set> sets) =>
 ///
 /// 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) {
+///
+/// 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;
 }