@Test public void create() throws ExecutionException, InterruptedException { TerminalStage<Integer, Integer> terminal = factory.create(null, () -> Collectors.summingInt((ToIntFunction<Integer>) value -> value)); List<Integer> list = new ArrayList<>(); Flowable<Integer> flowable = Flowable.fromArray(1, 2, 3, 4, 5, 6, 7, 8, 9, 10) .doOnNext(list::add) .subscribeOn(Schedulers.computation()); CompletionStage<Integer> stage = terminal.apply(flowable); Integer result = stage.toCompletableFuture().get(); assertThat(result).isEqualTo(1 + 2 + 3 + 4 + 5 + 6 + 7 + 8 + 9 + 10); assertThat(list).hasSize(10); }
@SuppressWarnings("unchecked") @Override public <I, O> TerminalStage<I, O> create(Engine engine, Stage.Collect stage) { Collector<I, Object, O> collector = (Collector<I, Object, O>) Objects.requireNonNull(stage).getCollector(); Objects.requireNonNull(collector); return new CollectStage<>(collector); }
CollectStageFactory implements TerminalStageFactory<Stage.Collect> { @SuppressWarnings("unchecked") @Override public <I, O> TerminalStage<I, O> create(Engine engine, Stage.Collect stage) { Collector<I, Object, O> collector = (Collector<I, Object, O>) Objects.requireNonNull(stage).getCollector(); Objects.requireNonNull(collector); return new CollectStage<>(collector); } }
CollectStageFactory implements TerminalStageFactory<Stage.Collect> { @SuppressWarnings("unchecked") @Override public <I, O> TerminalStage<I, O> create(Engine engine, Stage.Collect stage) { Collector<I, Object, O> collector = (Collector<I, Object, O>) Objects.requireNonNull(stage).getCollector(); Objects.requireNonNull(collector); return new CollectStage<>(collector); }  }
CollectStageFactory implements TerminalStageFactory<Stage.Collect> { @SuppressWarnings("unchecked") @Override public <I, O> TerminalStage<I, O> create(Engine engine, Stage.Collect stage) { Collector<I, Object, O> collector = (Collector<I, Object, O>) Objects.requireNonNull(stage).getCollector(); Objects.requireNonNull(collector); return new CollectStage<>(collector); }  @SuppressWarnings("unchecked") @Override TerminalStage<I, O> create(Engine engine, Stage.Collect stage); }
CollectStageFactory implements TerminalStageFactory<Stage.Collect> { @SuppressWarnings("unchecked") @Override public <I, O> TerminalStage<I, O> create(Engine engine, Stage.Collect stage) { Collector<I, Object, O> collector = (Collector<I, Object, O>) Objects.requireNonNull(stage).getCollector(); Objects.requireNonNull(collector); return new CollectStage<>(collector); }  @SuppressWarnings("unchecked") @Override TerminalStage<I, O> create(Engine engine, Stage.Collect stage);  }