@Test(expected = NullPointerException.class) public void createWithoutStage() { factory.create(null, null); }
@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; }); }
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; }); } }
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; }); }  }
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); }
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);  }