anchor.diff:    bool _isNextPending = false;
   
     /// Creates a new scheduled stream wrapping [stream].
  -  ScheduledStream(Stream<T> stream)
  -      : _stream = stream.asBroadcastStream() {
  +  ScheduledStream(this._stream) {
       _subscription = _stream.listen((value) {
  +      for (var c in _forkControllers) {
  +        c.add(value);
  +      }
         if (_hasNextCompleter != null) {
           _hasNextCompleter.complete(true);
           _hasNextCompleter = null;
  
positive.code:         _pendingValues.add(new Fallible.withValue(value));
        }
      }, onError: (error, stackTrace) {
        if (_hasNextCompleter != null) {
          _hasNextCompleter.completeError(error, stackTrace);
          _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].
  