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