@Test public void create() throws ExecutionException, InterruptedException { Flowable<Integer> flowable = Flowable.fromArray(1, 2, 3, 4, 5, 6, 7, 8, 9, 10) .subscribeOn(Schedulers.computation()); List<String> list = ReactiveStreams.fromPublisher(flowable) .filter(i -> i < 4) .flatMap(this::duplicate) .flatMapCompletionStage(this::asString) .toList() .run().toCompletableFuture().get(); assertThat(list).containsExactly("1", "1", "2", "2", "3", "3"); }
@Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.FlatMap stage) { Function<I, Graph> mapper = Casts.cast(stage.getMapper()); return new FlatMapStage<>(engine, mapper); }
FlatMapStageFactory implements ProcessingStageFactory<Stage.FlatMap> { @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.FlatMap stage) { Function<I, Graph> mapper = Casts.cast(stage.getMapper()); return new FlatMapStage<>(engine, mapper); } }
FlatMapStageFactory implements ProcessingStageFactory<Stage.FlatMap> { @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.FlatMap stage) { Function<I, Graph> mapper = Casts.cast(stage.getMapper()); return new FlatMapStage<>(engine, mapper); }  }
FlatMapStageFactory implements ProcessingStageFactory<Stage.FlatMap> { @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.FlatMap stage) { Function<I, Graph> mapper = Casts.cast(stage.getMapper()); return new FlatMapStage<>(engine, mapper); }  @Override ProcessingStage<I, O> create(Engine engine, Stage.FlatMap stage); }
FlatMapStageFactory implements ProcessingStageFactory<Stage.FlatMap> { @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.FlatMap stage) { Function<I, Graph> mapper = Casts.cast(stage.getMapper()); return new FlatMapStage<>(engine, mapper); }  @Override ProcessingStage<I, O> create(Engine engine, Stage.FlatMap stage);  }