Revert revision 31820

pkg/scheduled_test/lib/scheduled_stream.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 scheduled_test.scheduled_stream;
 
 import 'dart:async';
 import 'dart:collection';
 
 import 'package:stack_trace/stack_trace.dart';
@@ skipping 22 matching lines @@
   ///
   /// If this is non-null, [_pendingValues] will always be empty, since any
   /// value coming in will be passed to this completer.
   Completer<T> _nextCompleter;
 
   /// The completer for emitting a value requested by [hasNext].
   Completer<bool> _hasNextCompleter;
 
   /// 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].
   final _pendingValues = new Queue<Fallible<T>>();
 
   /// All values emitted by this stream so far.
   ///
   /// This does not include values emitted by the underlying stream but not yet
   /// emitted through [next].
   List<T> get emittedValues => new UnmodifiableListView(_emittedValues);
@@ skipping 12 matching lines @@
   /// Whether the wrapped stream has been closed.
   bool _isDone = false;
 
   /// Whether [next] has been called but has not yet returned.
   ///
   /// This is distinct from `_nextCompleter != null` when [next] is called while
   /// there are pending values available, until the future it returns completes.
   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;
       }
 
       if (_nextCompleter != null) {
         _nextCompleter.complete(value);
         _emittedValues.add(value);
         _nextCompleter = null;
         _isNextPending = false;
       } else {
         _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;
       }
 
       if (_nextCompleter != null) {
         _nextCompleter.completeError(error, stackTrace);
         _nextCompleter = null;
       } else {
         _pendingValues.add(new Fallible.withError(error, stackTrace));
@@ skipping 96 matching lines @@
   /// closed at whatever point it's currently at.
   ScheduledStream<T> fork() {
     var controller = new StreamController<T>();
     for (var valueOrError in _pendingValues) {
       if (valueOrError.hasValue) {
         controller.add(valueOrError.value);
       } else {
         controller.addError(valueOrError.error, valueOrError.stackTrace);
       }
     }
+
     if (_isDone) {
       controller.close();
     } else {
-      _forkControllers.add(controller);
+      _stream.pipe(controller);
     }
 
     var fork = new ScheduledStream<T>(controller.stream);
     _forks.add(fork);
     return fork;
   }
 
   /// Closes this stream.
   ///
   /// This cancels the subscription to the underlying stream and acts as though
   /// [this] was closed immediately after the current position, regardless of
   /// whether the underlying stream has emitted additional events.
   void close() {
     _subscription.cancel();
     _pendingValues.clear();
 
     for (var fork in _forks) {
       fork.close();
     }
     _forks.clear();
 
     if (!_isDone) _onDone();
   }
 
   /// 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;
     }
 
     if (_nextCompleter != null) {
       _nextCompleter.completeError(
           new StateError("ScheduledStream has no more elements."),
           new Chain.current());
       _nextCompleter = null;
       _isNextPending = false;
     }
 
     _isDone = true;
   }
 }