StreamFilter implements StreamTransformer<T, T> { public static <T> StreamFilter<T> create(Predicate<T> predicate) { return new StreamFilter<>(predicate); } private  StreamFilter(Predicate<T> predicate); static StreamFilter<T> create(Predicate<T> predicate); @Override StreamConsumer<T> getInput(); @Override StreamSupplier<T> getOutput();  }
@Test public void test1() { StreamSupplier<Integer> supplier = StreamSupplier.of(1, 2, 3, 4, 5, 6); StreamFilter<Integer> filter = StreamFilter.create(input -> input % 2 == 1); StreamConsumerToList<Integer> consumer = StreamConsumerToList.create(); await(supplier.transformWith(filter) .streamTo(consumer.transformWith(randomlySuspending()))); assertEquals(asList(1, 3, 5), consumer.getList()); assertEndOfStream(supplier); assertEndOfStream(filter); assertEndOfStream(consumer); }
@Test public void testWithError() { StreamSupplier<Integer> source = StreamSupplier.of(1, 2, 3, 4, 5, 6); StreamFilter<Integer> streamFilter = StreamFilter.create(input -> input % 2 != 1); StreamConsumerToList<Integer> consumer = StreamConsumerToList.create(); ExpectedException exception = new ExpectedException("Test Exception"); Throwable e = awaitException(source.transformWith(streamFilter) .streamTo(consumer .transformWith(decorate(promise -> promise.then(item -> item == 4 ? Promise.ofException(exception) : Promise.of(item)))))); assertSame(exception, e); assertEquals(asList(2, 4), consumer.getList()); assertClosedWithError(source); assertClosedWithError(consumer); assertClosedWithError(streamFilter); }
@Test public void testSupplierDisconnectWithError() { ExpectedException exception = new ExpectedException("Test Exception"); StreamSupplier<Integer> source = StreamSupplier.concat( StreamSupplier.ofIterable(Arrays.asList(1, 2, 3, 4, 5, 6)), StreamSupplier.closingWithError(exception)); StreamFilter<Integer> streamFilter = StreamFilter.create(input -> input % 2 != 1); StreamConsumerToList<Integer> consumer = StreamConsumerToList.create(); Throwable e = awaitException(source.transformWith(streamFilter) .streamTo(consumer.transformWith(randomlySuspending()))); assertSame(exception, e); assertClosedWithError(consumer); assertClosedWithError(streamFilter); }
@Test public void testFilterConsumer() { StreamSupplier<Integer> supplier = StreamSupplier.of(1, 2, 3, 4, 5, 6); StreamFilter<Integer> filter = StreamFilter.create(input -> input % 2 == 1); StreamConsumerToList<Integer> consumer = StreamConsumerToList.create(); StreamConsumer<Integer> transformedConsumer = consumer .transformWith(filter) .transformWith(randomlySuspending()); await(supplier.streamTo(transformedConsumer)); assertEquals(asList(1, 3, 5), consumer.getList()); assertEndOfStream(supplier); assertEndOfStream(filter); }