ProcessorStageFactory implements ProcessingStageFactory<Stage.ProcessorStage> { @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.ProcessorStage stage) { Processor<I, O> processor = Casts.cast(Objects.requireNonNull( Objects.requireNonNull(stage).getRsProcessor())); return source -> Flowable.defer(() -> { Flowable<O> flowable = Flowable.fromPublisher(processor); source.safeSubscribe(processor); return flowable; }); }  @Override ProcessingStage<I, O> create(Engine engine, Stage.ProcessorStage stage);  }
@Test(expected = NullPointerException.class) public void createWithoutStage() { factory.create(null, null); }
@Test(expected = NullPointerException.class) public void createWithoutFunction() { factory.create(null, () -> null); }