FlatMapCompletionStageFactory implements ProcessingStageFactory<Stage.FlatMapCompletionStage> { @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.FlatMapCompletionStage stage) { Function<I, CompletionStage<O>> mapper = Casts.cast( Objects.requireNonNull(stage).getMapper()); return new FlatMapCompletionStage<>(mapper); }  @Override ProcessingStage<I, O> create(Engine engine,
            Stage.FlatMapCompletionStage stage);  }
@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) .flatMapCompletionStage(this::square) .flatMapCompletionStage(this::asString) .toList() .run().toCompletableFuture().get(); assertThat(list).containsExactly("1", "4", "9"); }
@Test(expected = NullPointerException.class) public void createWithoutStage() { factory.create(null, null); }
@Test(expected = NullPointerException.class) public void createWithoutFunction() { factory.create(null, () -> null); }