FilterDataGenerator extends DataGenerator<A> { @Override public DataGenerator<A> apply() { DataGenerator<A> nextGen = source.apply(); if (nextGen.isClosed()) { return (DataGenerator<A>) EOS; } return new FilterDataGenerator<>(nextGen, predicate); } private  FilterDataGenerator(DataGenerator<A> source, Function<A, Boolean> predicate); @Override DataGenerator<A> apply(); @Override Optional<A> getOptionalValue(); static DataGenerator<A> newInstance(DataGenerator<A> source, Function<A, Boolean> predicate);  }
@Test public void testFilter() throws Exception { DataGenerator<List<Long>> filteredBatch = DataGenerator.range(0, 9).filter(v -> v % 2 == 0).batch(5); assertThat(filteredBatch.getValue()).contains(0L, 2L, 4L, 6L, 8L); assertThat(filteredBatch.isClosed()).isFalse(); assertThat(filteredBatch.apply().isClosed()).isTrue(); }