class 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) {
                    // Do nothing.
                }

                @Override
                public void onError(Throwable t) {
                    // Do nothing.
                }

                @Override
                public void onComplete() {
                    // Do nothing.
                }
            });
            return CompletableFuture.completedFuture(null);
        };
    }

}