anchor.diff:  
     /// 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].
  
positive.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;
  
negative.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;
  