@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<>();
    Flowable<Long> flowable = Flowable.interval(1000, TimeUnit.MILLISECONDS)
            .observeOn(Schedulers.io())
            .doOnNext(list::add)
            .doOnCancel(() -> cancelled.set(true));
    CompletionStage<Void> stage = terminal.apply(flowable);
    stage.toCompletableFuture().get();
    await().untilAtomic(cancelled, is(true));
    assertThat(list).isEmpty();
    assertThat(cancelled).isTrue();
}