anchor.diff:    /// The stream which will contain the remaining events of [_streamQueue].
     Stream<T> get stream => _completer.stream;
   
  -  bool addEvents(Queue<Result> events) {
  -    _completeStream(events);
  -    return true;
  -  }
  -
  -  void close(Queue<Result> events) {
  -    _completeStream(events);
  -  }
  -
  -  void _completeStream(Queue<Result> events) {
  +  bool update(Queue<Result> events, bool isDone) {
       if (events.isEmpty) {
         if (_streamQueue._isDone) {
           _completer.setEmpty();
         } else {
  -        _completer.setSourceStream(_getRestStream());
  +        _completer.setSourceStream(_streamQueue._extractStream());
         }
       } else {
         // There are prefetched events which needs to be added before the
  
positive.code:   /// The future completed when the cancel request is completed.
    Future get future => _completer.future;
  
    bool addEvents(Queue<Result> events) {
      _shutdown();
      return true;
    }
  
    void close(_) {
      _shutdown();
    }
  
    void _shutdown() {
      if (_streamQueue._isDone) {
        _completer.complete();
      } else {
        _streamQueue._ensureListening();
        _completer.complete(_streamQueue._dispose().cancel());
      }
    }
  }
  
  
negative.code: 
    Future<T> get future => _completer.future;
  
    bool update(Queue<Result> events, bool isDone) {
      if (events.isNotEmpty) {
        events.removeFirst().complete(_completer);
        return true;
      }
      if (isDone) {
        var errorFuture =
            new Future.sync(() => throw new StateError("No elements"));
        _completer.complete(errorFuture);
        return true;
      }
      return false;
    }
  }
  
  