class 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;
        });
    }

}