anchor.diff:    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;
  
positive.code:         _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;
  
negative.code:   /// Handles a "done" event from the underlying stream, as well as [this] being
    /// closed.
    void _onDone() {
      if (_hasNextCompleter != null) {
        _hasNextCompleter.complete(false);
        _hasNextCompleter = null;
  