class DistinctStageFactory implements ProcessingStageFactory<Stage.Distinct> {

    @SuppressWarnings("unchecked")
    @Override
    public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.Distinct stage) {
        Objects.requireNonNull(stage);
        return source -> (Flowable<O>) source.distinct();
    }

}