@Test(expected = NullPointerException.class) public void createWithoutStage() { factory.create(null, null); }
@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); }; }
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); }; } }
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); }; }  }
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); }
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);  }