OnErrorResumeWithStageFactory implements ProcessingStageFactory<Stage.OnErrorResumeWith> { @SuppressWarnings("unchecked") @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.OnErrorResumeWith stage) { Function<Throwable, Graph> function = Objects.requireNonNull(stage).getFunction(); Objects.requireNonNull(function); return source -> (Flowable<O>) RxJavaPlugins.onAssembly( new OnErrorResumeWith<>(source, (Throwable err) -> { Graph graph = function.apply(err); return Flowable.fromPublisher( Objects.requireNonNull(engine.buildPublisher(Objects.requireNonNull(graph)))); })); }  @SuppressWarnings("unchecked") @Override ProcessingStage<I, O> create(Engine engine, Stage.OnErrorResumeWith 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()); List<Integer> list = ReactiveStreams.<Integer> failed(new Exception("BOOM")) .onErrorResumeWith(t -> ReactiveStreams.fromPublisher(flowable)) .toList() .run().toCompletableFuture().exceptionally(x -> Collections.emptyList()).get(); assertThat(list).hasSize(10); }
@Test(expected = NullPointerException.class) public void createWithoutStage() { factory.create(null, null); }
@Test(expected = NullPointerException.class) public void createWithoutFunction() { factory.create(null, () -> null); }