anchor.diff:          _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;
  
positive.code: 
    /// The set of all streams forked from this one.
    final _forks = new Set<ScheduledStream<T>>();
  
    /// The queue of values emitted by [_stream] but not yet emitted through
    /// [next].
  
negative.code:   /// 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;
  