OnErrorStageFactory implements ProcessingStageFactory<Stage.OnError> { @SuppressWarnings("unchecked") @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.OnError stage) { Consumer<Throwable> consumer = Objects.requireNonNull(stage).getConsumer(); Objects.requireNonNull(consumer); return source -> (Flowable<O>) source.doOnError(consumer::accept); }  @SuppressWarnings("unchecked") @Override ProcessingStage<I, O> create(Engine engine, Stage.OnError stage);  }
@Test public void create() throws ExecutionException, InterruptedException { Flowable<Integer> flowable = Flowable.fromArray(1, 2, 3, 4, 5, 6, 7, 8, 9, 10) .subscribeOn(Schedulers.computation()); AtomicReference<Throwable> error = new AtomicReference<>(); ReactiveStreams.fromPublisher(flowable) .filter(i -> i < 4) .map(this::squareOrFailed) .onError(error::set) .map(this::asString) .toList() .run().toCompletableFuture().exceptionally(x -> Collections.emptyList()).get(); assertThat(error.get()).hasMessage("failed"); }
@Test(expected = NullPointerException.class) public void createWithoutStage() { factory.create(null, null); }
@Test(expected = NullPointerException.class) public void createWithoutFunction() { factory.create(null, () -> null); }