Properly close watcher streams during tests.

This also makes [futureStream] pass cancellations through to the underlying
stream, and adds the ability for it to return broadcast streams.

R=rnystrom@google.com
BUG=14943

Review URL: https://codereview.chromium.org//66293008

git-svn-id: https://dart.googlecode.com/svn/branches/bleeding_edge/dart/pkg/watcher@30180 260f80e4-7a28-3924-810f-c04153c831b5

pkgs/watcher/lib/src/utils.dart

diff --git a/pkgs/watcher/lib/src/utils.dart b/pkgs/watcher/lib/src/utils.dart
index 66c0f09..a64575b 100644
--- a/pkgs/watcher/lib/src/utils.dart
+++ b/pkgs/watcher/lib/src/utils.dart
@@ -27,17 +27,44 @@
 ///
 /// 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;
 }
 

pkgs/watcher/test/utils.dart

diff --git a/pkgs/watcher/test/utils.dart b/pkgs/watcher/test/utils.dart
index 136c8b0..567bdb2 100644
--- a/pkgs/watcher/test/utils.dart
+++ b/pkgs/watcher/test/utils.dart
@@ -143,7 +143,7 @@
     }, "reset watcher");
 
     return _watcher.events;
-  }, "create watcher")).asBroadcastStream();
+  }, "create watcher"), broadcast: true);
 
   schedule(() => _watcher.ready, "wait for watcher to be ready");
 }