@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); }
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();  }