CancelStageFactory implements TerminalStageFactory<Stage.Cancel> { @Override public <I, O> TerminalStage<I, O> create(Engine engine, Stage.Cancel stage) { Objects.requireNonNull(stage); return (Flowable<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 CompletableFuture.completedFuture(null); }; }  @Override TerminalStage<I, O> create(Engine engine, Stage.Cancel stage);  }
@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(); }
@Test(expected = NullPointerException.class) public void createWithoutStage() { factory.create(null, null); }