FilteringStateChangeListenable implements StateChangeListenable<C> { @Override public void registerAtomicStateChangeListener(AtomicStateChangeListener<? super C> listener) { backingListenable.registerAtomicStateChangeListener(changes -> { ImmutableList<Change<C>> plannableChanges = StreamSupport.stream(changes.spliterator(), false) .map(this::wrapChange) .filter(Objects::nonNull) .collect(ImmutableList.toImmutableList()); ((AtomicStateChangeListener<C>)listener).stateChanged(plannableChanges); }); } FilteringStateChangeListenable(StateChangeListenable<T> backingListenable, Function<? super T, C> mapAndFilter); @Override T registerCustomIndex(T index); @Override void registerStateAddedOrRemovedListener(Consumer<? super C> consumer); @Override void registerStateChangeListener(CacheStateChangeListener<? super C> listener); @Override void registerStateAddedListener(CacheStateAddedListener<? super C> listener); @Override void registerStateRemovedListener(CacheStateRemovedListener<? super C> listener); @Override Stream<C> stream(); @Override UnmodifiableIterator<C> iterator(); @Override void registerAtomicStateChangeListener(AtomicStateChangeListener<? super C> listener); @Override void removeStateChangeListener(CacheStateChangeListener<? super C> listener);  }
@Test void whenOnlyListeningForExtendedStates_thenAtomicChangeReceivesAllExtendedStateChanges() { cache.add(state1); cache.add(extended1); List<Change> updates = new ArrayList<>(); (new FilteringStateChangeListenable<>(cache, this::asExtended)).registerAtomicStateChangeListener(changes -> { result = true; changes.forEach(updates::add); }); cache.updateAll(ImmutableList.of( Change.change(extended1, extended1Updated), Change.add(state2), Change.add(extended2), Change.delete(state1))); assertTrue(result); assertEquals(2, updates.size()); assertEquals(extended1, updates.get(0).originalObject); assertEquals(extended1Updated, updates.get(0).newObject); assertNull(updates.get(1).originalObject); assertEquals(extended2, updates.get(1).newObject); }
@Test void whenOnlyListeningForExtendedStates_thenAtomicChangeReceivesAllExtendedStateDeletions() { cache.add(state1); cache.add(extended1); cache.add(state2); cache.add(extended2); List<Change> updates = new ArrayList<>(); (new FilteringStateChangeListenable<>(cache, this::asExtended)).registerAtomicStateChangeListener(changes -> { result = true; changes.forEach(updates::add); }); cache.updateAll(ImmutableList.of( Change.delete(state1), Change.delete(state2), Change.delete(extended1), Change.delete(extended2))); assertTrue(result); assertEquals(2, updates.size()); assertEquals(extended1, updates.get(0).originalObject); assertNull(updates.get(0).newObject); assertEquals(extended2, updates.get(1).originalObject); assertNull(updates.get(0).newObject); }