anchor.diff:          controller.addError(valueOrError.error, valueOrError.stackTrace);
         }
       }
  +
       if (_isDone) {
         controller.close();
       } else {
  -      _forkControllers.add(controller);
  +      _stream.pipe(controller);
       }
   
       var fork = new ScheduledStream<T>(controller.stream);
  
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;
  