@Test public void addIndexFlow_Test() throws Exception { final Main example = new Main(new Helper()); List<String> messages = Arrays.asList( "I say high, you say low", "You say why and I say I don't know", "Oh, no", "You say goodbye and I say hello" ); final Flow<String, String, NotUsed> addIndexFlow = example.addIndexFlow(); final CompletionStage<List<String>> future = Source.from(messages) .via(addIndexFlow) .runWith(Sink.seq(), system); final List<String> result = future.toCompletableFuture().get(3, TimeUnit.SECONDS); assert(result.size() == messages.size()); final Pattern pattern = Pattern.compile("index: \\d+, message: (.*)"); Streams.zip( messages.stream(), result.stream(), Pair::create) .forEachOrdered(pair -> { String message = pair.first(); String resultMessage = pair.second(); Matcher matcher = pattern.matcher(resultMessage); assertTrue(matcher.find()); assert(matcher.groupCount() == 1); assert(matcher.group(1).equals(message)); }); }
public Flow<String, String, NotUsed> addIndexFlow() { final Pair<Integer, String> seed = Pair.create(0, "start"); return Flow.of(String.class) .scan(seed, (acc, message) -> { Integer index = acc.first(); return Pair.create(index + 1, String.format("index: %s, message: %s", index, message)); }) .filterNot(p -> p == seed) .map(Pair::second); }
Main extends AllDirectives { public Flow<String, String, NotUsed> addIndexFlow() { final Pair<Integer, String> seed = Pair.create(0, "start"); return Flow.of(String.class) .scan(seed, (acc, message) -> { Integer index = acc.first(); return Pair.create(index + 1, String.format("index: %s, message: %s", index, message)); }) .filterNot(p -> p == seed) .map(Pair::second); } }
Main extends AllDirectives { public Flow<String, String, NotUsed> addIndexFlow() { final Pair<Integer, String> seed = Pair.create(0, "start"); return Flow.of(String.class) .scan(seed, (acc, message) -> { Integer index = acc.first(); return Pair.create(index + 1, String.format("index: %s, message: %s", index, message)); }) .filterNot(p -> p == seed) .map(Pair::second); } Main(Helper helper); }
Main extends AllDirectives { public Flow<String, String, NotUsed> addIndexFlow() { final Pair<Integer, String> seed = Pair.create(0, "start"); return Flow.of(String.class) .scan(seed, (acc, message) -> { Integer index = acc.first(); return Pair.create(index + 1, String.format("index: %s, message: %s", index, message)); }) .filterNot(p -> p == seed) .map(Pair::second); } Main(Helper helper); static void main(String[] args); Flow<String, String, NotUsed> addIndexFlow(); }
Main extends AllDirectives { public Flow<String, String, NotUsed> addIndexFlow() { final Pair<Integer, String> seed = Pair.create(0, "start"); return Flow.of(String.class) .scan(seed, (acc, message) -> { Integer index = acc.first(); return Pair.create(index + 1, String.format("index: %s, message: %s", index, message)); }) .filterNot(p -> p == seed) .map(Pair::second); } Main(Helper helper); static void main(String[] args); Flow<String, String, NotUsed> addIndexFlow();  }