### Question:
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);  }### Answer:
@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<>(); Multi<Integer> publisher = Multi.createFrom().items(1, 2, 3, 4, 5, 6, 7, 8, 9, 10) .onItem().invoke(list::add) .emitOn(computation); CompletionStage<Integer> stage = terminal.apply(publisher); Integer result = stage.toCompletableFuture().get(); assertThat(result).isEqualTo(1 + 2 + 3 + 4 + 5 + 6 + 7 + 8 + 9 + 10); assertThat(list).hasSize(10); }
@Test public void createWithoutStage() { assertThrows(NullPointerException.class, () -> factory.create(new Engine(), null)); }
@Test public void createWithoutCollector() { assertThrows(NullPointerException.class, () -> factory.create(null, () -> null)); }