class FilterStageFactory implements ProcessingStageFactory<Stage.Filter> {

    @SuppressWarnings("unchecked")
    @Override
    public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.Filter stage) {
        Objects.requireNonNull(stage);
        Predicate predicate = Objects.requireNonNull(stage.getPredicate());
        return source -> (Flowable<O>) source.filter(predicate::test);
    }

}