@Test public void createWithoutFunction() { assertThrows(NullPointerException.class, () -> factory.create(null, () -> null)); }
@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 -> (Multi<O>) source.onFailure().invoke(consumer); }
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 -> (Multi<O>) source.onFailure().invoke(consumer); } }
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 -> (Multi<O>) source.onFailure().invoke(consumer); }  }
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 -> (Multi<O>) source.onFailure().invoke(consumer); }  @SuppressWarnings("unchecked") @Override ProcessingStage<I, O> create(Engine engine, Stage.OnError stage); }
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 -> (Multi<O>) source.onFailure().invoke(consumer); }  @SuppressWarnings("unchecked") @Override ProcessingStage<I, O> create(Engine engine, Stage.OnError stage);  }