class TakeWhileStageFactory implements ProcessingStageFactory<Stage.TakeWhile> {

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

}