@Test public void createWithoutStage() { assertThrows(NullPointerException.class, () -> 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 -> Multi.createFrom().deferred(() -> { Multi<O> multi = Multi.createFrom().publisher(processor); source.subscribe().withSubscriber(processor); return multi; }); }
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 -> Multi.createFrom().deferred(() -> { Multi<O> multi = Multi.createFrom().publisher(processor); source.subscribe().withSubscriber(processor); return multi; }); } }
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 -> Multi.createFrom().deferred(() -> { Multi<O> multi = Multi.createFrom().publisher(processor); source.subscribe().withSubscriber(processor); return multi; }); }  }
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 -> Multi.createFrom().deferred(() -> { Multi<O> multi = Multi.createFrom().publisher(processor); source.subscribe().withSubscriber(processor); return multi; }); }  @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 -> Multi.createFrom().deferred(() -> { Multi<O> multi = Multi.createFrom().publisher(processor); source.subscribe().withSubscriber(processor); return multi; }); }  @Override ProcessingStage<I, O> create(Engine engine, Stage.ProcessorStage stage);  }