anchor.diff:      _allMatching = allMatching;
       var list = await sortAndLimit(_allMatching, finder, cooperator);
   
  -    if (isClosed) {
  -      return;
  -    }
  +    if (!_shouldAdd) {
  +      return;
  +    }
  +
       // devPrint('adding $allMatching / limit $list / $finder');
       _streamController?.add(immutableListToSnapshots<K, V>(list));
     }
   
     /// Add error.
     void addError(dynamic error, StackTrace stackTrace) {
  -    if (isClosed) {
  +    if (!_shouldAdd) {
         return;
       }
       _streamController.addError(error, stackTrace);
  
positive.code:   /// stream.
    Stream<List<RecordSnapshot<K, V>>> get stream => _streamController.stream;
  
    /// Add data to stream.
    Future add(
        List<ImmutableSembastRecord> allMatching, Cooperator cooperator) async {
      if (isClosed) {
        return;
      }
  
  
negative.code:   /// stream.
    Stream<RecordSnapshot<K, V>> get stream => _streamController.stream;
  
    bool get _shouldAdd => !isClosed && _streamController.hasListener;
  
    /// Add a snapshot.
    void add(RecordSnapshot snapshot) {
      if (!_shouldAdd) {
        return;
      }
      hasInitialData = true;
      _streamController.add(snapshot?.cast<K, V>());
    }
  
    /// Add an error.
    void addError(dynamic error, StackTrace stackTrace) {
      if (!_shouldAdd) {
        return;
      }
      _streamController.addError(error, stackTrace);
    }
  }
  
  