anchor.diff:          controller.addError(valueOrError.error, valueOrError.stackTrace);
         }
       }
  -
       if (_isDone) {
         controller.close();
       } else {
  -      _stream.pipe(controller);
  +      _forkControllers.add(controller);
       }
   
       var fork = new ScheduledStream<T>(controller.stream);
  
positive.code:   bool _isNextPending = false;
  
    /// Creates a new scheduled stream wrapping [stream].
    ScheduledStream(Stream<T> stream)
        : _stream = stream.asBroadcastStream() {
      _subscription = _stream.listen((value) {
        if (_hasNextCompleter != null) {
          _hasNextCompleter.complete(true);
          _hasNextCompleter = null;
  
negative.code: 
    /// 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].
  