@Test public void createWithoutStage() { assertThrows(NullPointerException.class, () -> factory.create(null, null)); }
@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 -> (Multi<O>) source.onFailure().recoverWithMulti(failure -> { Graph graph = function.apply(failure); Publisher<I> publisher = engine.buildPublisher(Objects.requireNonNull(graph)); return Multi.createFrom().publisher(publisher); }); }
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 -> (Multi<O>) source.onFailure().recoverWithMulti(failure -> { Graph graph = function.apply(failure); Publisher<I> publisher = engine.buildPublisher(Objects.requireNonNull(graph)); return Multi.createFrom().publisher(publisher); }); } }
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 -> (Multi<O>) source.onFailure().recoverWithMulti(failure -> { Graph graph = function.apply(failure); Publisher<I> publisher = engine.buildPublisher(Objects.requireNonNull(graph)); return Multi.createFrom().publisher(publisher); }); }  }
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 -> (Multi<O>) source.onFailure().recoverWithMulti(failure -> { Graph graph = function.apply(failure); Publisher<I> publisher = engine.buildPublisher(Objects.requireNonNull(graph)); return Multi.createFrom().publisher(publisher); }); }  @SuppressWarnings("unchecked") @Override ProcessingStage<I, O> create(Engine engine, Stage.OnErrorResumeWith stage); }
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 -> (Multi<O>) source.onFailure().recoverWithMulti(failure -> { Graph graph = function.apply(failure); Publisher<I> publisher = engine.buildPublisher(Objects.requireNonNull(graph)); return Multi.createFrom().publisher(publisher); }); }  @SuppressWarnings("unchecked") @Override ProcessingStage<I, O> create(Engine engine, Stage.OnErrorResumeWith stage);  }