FilteringStateChangeListenable implements StateChangeListenable<C> { @Override public Stream<C> stream() { return backingListenable.stream().map(mapAndFilter).filter(Objects::nonNull); } 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_thenStreamOnlyReturnsThemInOrder() { cache.add(state1); cache.add(extended1); cache.add(state2); cache.add(extended2); ImmutableList<ExtendedTestState> states = (new FilteringStateChangeListenable<>(cache, this::asExtended)).stream().collect(ImmutableList.toImmutableList()); assertEquals(2, states.size()); assertEquals(extended1, states.get(0)); assertEquals(extended2, states.get(1)); }