class LimitStageFactory implements ProcessingStageFactory<Stage.Limit> {

    @Override
    public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.Limit stage) {
        long limit = stage.getLimit();
        return source -> (Flowable<O>) source.limit(limit);
    }

}