### Question:
CancelStageFactory implements TerminalStageFactory<Stage.Cancel> { @Override public <I, O> TerminalStage<I, O> create(Engine engine, Stage.Cancel stage) { Objects.requireNonNull(stage); return (Multi<I> flow) -> { flow.subscribe(new Subscriber<I>() { @Override public void onSubscribe(Subscription s) { s.cancel(); } @Override public void onNext(I in) { } @Override public void onError(Throwable t) { } @Override public void onComplete() { } }); return Infrastructure.wrapCompletableFuture(CompletableFuture.completedFuture(null)); }; }  @Override TerminalStage<I, O> create(Engine engine, Stage.Cancel stage);  }### Answer:
@Test public void create() throws ExecutionException, InterruptedException { TerminalStage<Long, Void> terminal = factory.create(null, new Stage.Cancel() { }); AtomicBoolean cancelled = new AtomicBoolean(); List<Long> list = new ArrayList<>(); Multi<Long> publisher = Multi.createFrom().ticks().every(Duration.ofMillis(1000)) .emitOn(Infrastructure.getDefaultExecutor()) .onItem().invoke(list::add) .onCancellation().invoke(() -> cancelled.set(true)); CompletionStage<Void> stage = terminal.apply(publisher); stage.toCompletableFuture().get(); await().untilAtomic(cancelled, is(true)); assertThat(list).isEmpty(); assertThat(cancelled).isTrue(); }
@Test public void createWithoutStage() { assertThrows(NullPointerException.class, () -> factory.create(null, null)); }