class FlatMapIterableStageFactory implements ProcessingStageFactory<Stage.FlatMapIterable> {

    @Override
    public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.FlatMapIterable stage) {
        Function<I, Iterable<O>> mapper = Casts.cast(stage.getMapper());
        return new FlatMapIterable<>(mapper);
    }

}

class FlatMapIterableStageFactoryTest  {

    private final FlatMapIterableStageFactory factory;
    private final ExecutorService executor;

    @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)
                .flatMapIterable(this::duplicate)
                .flatMapCompletionStage(this::asString)
                .toList()
                .run().toCompletableFuture().get();

        assertThat(list).containsExactly("1", "1", "2", "2", "3", "3");
    }

}