anchor.diff:          controller.addError(valueOrError.error, valueOrError.stackTrace);
         }
       }
  -
       if (_isDone) {
         controller.close();
       } else {
  -      _stream.pipe(controller);
  +      _forkControllers.add(controller);
       }
   
       var fork = new ScheduledStream<T>(controller.stream);
  
positive.code:         _pendingValues.add(new Fallible.withValue(value));
        }
      }, onError: (error, stackTrace) {
        if (_hasNextCompleter != null) {
          _hasNextCompleter.completeError(error, stackTrace);
          _hasNextCompleter = null;
  
negative.code:   bool _isNextPending = false;
  
    /// Creates a new scheduled stream wrapping [stream].
    ScheduledStream(this._stream) {
      _subscription = _stream.listen((value) {
        for (var c in _forkControllers) {
          c.add(value);
        }
        if (_hasNextCompleter != null) {
          _hasNextCompleter.complete(true);
          _hasNextCompleter = null;
  