anchor.diff:    /// 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;
  
positive.code:   bool _isNextPending = false;
  
    /// Creates a new scheduled stream wrapping [stream].
    ScheduledStream(Stream<T> stream)
        : _stream = stream.asBroadcastStream() {
      _subscription = _stream.listen((value) {
        if (_hasNextCompleter != null) {
          _hasNextCompleter.complete(true);
          _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;
  