Revert revision 31820

pkg/scheduled_test/lib/scheduled_stream.dart

Index: pkg/scheduled_test/lib/scheduled_stream.dart
===================================================================
--- pkg/scheduled_test/lib/scheduled_stream.dart	(revision 31820)
+++ pkg/scheduled_test/lib/scheduled_stream.dart	(working copy)
@@ -40,7 +40,6 @@
 
   /// The set of all streams forked from this one.
   final _forks = new Set<ScheduledStream<T>>();
-  final _forkControllers = new Set<StreamController<T>>();
 
   /// The queue of values emitted by [_stream] but not yet emitted through
   /// [next].
@@ -73,11 +72,9 @@
   bool _isNextPending = false;
 
   /// Creates a new scheduled stream wrapping [stream].
-  ScheduledStream(this._stream) {
+  ScheduledStream(Stream<T> stream)
+      : _stream = stream.asBroadcastStream() {
     _subscription = _stream.listen((value) {
-      for (var c in _forkControllers) {
-        c.add(value);
-      }
       if (_hasNextCompleter != null) {
         _hasNextCompleter.complete(true);
         _hasNextCompleter = null;
@@ -92,9 +89,6 @@
         _pendingValues.add(new Fallible.withValue(value));
       }
     }, onError: (error, stackTrace) {
-      for (var c in _forkControllers) {
-        c.addError(error, stackTrace);
-      }
       if (_hasNextCompleter != null) {
         _hasNextCompleter.completeError(error, stackTrace);
         _hasNextCompleter = null;
@@ -211,10 +205,11 @@
         controller.addError(valueOrError.error, valueOrError.stackTrace);
       }
     }
+
     if (_isDone) {
       controller.close();
     } else {
-      _forkControllers.add(controller);
+      _stream.pipe(controller);
     }
 
     var fork = new ScheduledStream<T>(controller.stream);
@@ -242,11 +237,6 @@
   /// Handles a "done" event from the underlying stream, as well as [this] being
   /// closed.
   void _onDone() {
-    for (var c in _forkControllers) {
-      c.close();
-    }
-    _forkControllers.clear();
-
     if (_hasNextCompleter != null) {
       _hasNextCompleter.complete(false);
       _hasNextCompleter = null;