@Test public void testBasics() throws Exception { final Duration READ_DELAY = Duration.ofMillis(100L); ListDataSink<Set<String>> out = ListDataSink.get(); Fluent.flow("Test") .read(ListDataSource.unbounded( asList("0-one 1-two 0-three 1-four 0-five 1-six 0-seven".split(" "))) .withReadDelay(READ_DELAY)) .apply(input -> ReduceByKey.of(input) .keyBy(e -> "") .valueBy(e -> e) .reduceBy(s -> s.collect(Collectors.toSet())) .windowBy(Count.of(3))) .flatMap((Pair<String, Set<String>> e, Collector<String> c) -> e.getSecond().stream().forEachOrdered(c::collect)) .apply(input -> ReduceByKey.of(input) .keyBy(e -> "") .valueBy(e -> e) .reduceBy(s -> s.collect(Collectors.toSet()))) .mapElements(Pair::getSecond) .persist(out) .execute(new LocalExecutor()); }
public static Flow flow(String name) { return Flow.create(name); }
Fluent { public static Flow flow(String name) { return Flow.create(name); } }
Fluent { public static Flow flow(String name) { return Flow.create(name); }  }
Fluent { public static Flow flow(String name) { return Flow.create(name); }  static Flow flow(String name); static Flow flow(String name, Settings settings); static Dataset<T> lift(cz.seznam.euphoria.core.client.dataset.Dataset<T> xs); }
Fluent { public static Flow flow(String name) { return Flow.create(name); }  static Flow flow(String name); static Flow flow(String name, Settings settings); static Dataset<T> lift(cz.seznam.euphoria.core.client.dataset.Dataset<T> xs);  }