@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); }
public static <T> StreamFilter<T> create(Predicate<T> predicate) { return new StreamFilter<>(predicate); }
StreamFilter implements StreamTransformer<T, T> { public static <T> StreamFilter<T> create(Predicate<T> predicate) { return new StreamFilter<>(predicate); } }
StreamFilter implements StreamTransformer<T, T> { public static <T> StreamFilter<T> create(Predicate<T> predicate) { return new StreamFilter<>(predicate); } private  StreamFilter(Predicate<T> predicate); }
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(); }
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();  }