class 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))));
                }));
    }

}